LU-6968 llog: Only update necessary part of header

During llog_cancel_rec(), only update the part of header,
instead of the whole header, otherwise it might need pack
the whole header (32KB) into the RPC for each record
cancellation.

Signed-off-by: wang di <di.wang@intel.com>
Change-Id: I1e6e04b5b725ca8ed500b6c4d767719d7ff5f658
Reviewed-on: http://review.whamcloud.com/15905
Tested-by: Jenkins
Reviewed-by: Mike Pershin <mike.pershin@intel.com>
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: James Simmons <uja.ornl@yahoo.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index 3aeea1d..7bdd90f 100644 (file)
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -197,6 +197,7 @@ EXPORT_SYMBOL(llog_destroy);
 int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
                    int index)
 {
+       struct llog_thread_info *lgi = llog_info(env);
        struct dt_device        *dt;
        struct llog_log_hdr     *llh = loghandle->lgh_hdr;
        struct thandle          *th;
@@ -241,13 +242,19 @@ int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
                CDEBUG(D_RPCTRACE, "Catalog index %u already clear?\n", index);
                GOTO(out_unlock, rc);
        }
+
+       loghandle->lgh_hdr->llh_count--;
+       /* Pass this index to llog_osd_write_rec(), which will use the index
+        * to only update the necesary bitmap. */
+       lgi->lgi_cookie.lgc_index = index;
        /* update header */
-       rc = llog_write_rec(env, loghandle, &llh->llh_hdr, NULL,
+       rc = llog_write_rec(env, loghandle, &llh->llh_hdr, &lgi->lgi_cookie,
                            LLOG_HEADER_IDX, th);
-       if (rc == 0)
-               loghandle->lgh_hdr->llh_count--;
-       else
+       if (rc != 0) {
+               loghandle->lgh_hdr->llh_count++;
                ext2_set_bit(index, LLOG_HDR_BITMAP(llh));
+               GOTO(out_unlock, rc);
+       }
 
        if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
            (llh->llh_count == 1) &&

diff --git a/lustre/obdclass/llog_internal.h b/lustre/obdclass/llog_internal.h
index 2950864..c968422 100644 (file)
--- a/lustre/obdclass/llog_internal.h
+++ b/lustre/obdclass/llog_internal.h
@@ -58,6 +58,7 @@ struct llog_thread_info {
        struct llog_logid_rec            lgi_logid;
        struct dt_insert_rec             lgi_dt_rec;
        struct lu_seq_range              lgi_range;
+       struct llog_cookie               lgi_cookie;
        char                             lgi_name[32];
 };
 

diff --git a/lustre/obdclass/llog_osd.c b/lustre/obdclass/llog_osd.c
index a8a53e4..bc9951b 100644 (file)
--- a/lustre/obdclass/llog_osd.c
+++ b/lustre/obdclass/llog_osd.c
@@ -422,11 +422,44 @@ static int llog_osd_write_rec(const struct lu_env *env,
 
                if (idx == LLOG_HEADER_IDX) {
                        /* llog header update */
+                       __u32   *bitmap = LLOG_HDR_BITMAP(llh);
+
                        lgi->lgi_off = 0;
-                       lgi->lgi_buf.lb_len = reclen;
-                       lgi->lgi_buf.lb_buf = rec;
+
+                       /* If it does not indicate the bitmap index
+                        * (reccookie == NULL), then it means update
+                        * the whole update header. Otherwise only
+                        * update header and bits needs to be updated,
+                        * and in DNE cases, it will signaficantly
+                        * shrink the RPC size.
+                        * see distribute_txn_cancel_records()*/
+                       if (reccookie == NULL) {
+                               lgi->lgi_buf.lb_len = reclen;
+                               lgi->lgi_buf.lb_buf = rec;
+                               rc = dt_record_write(env, o, &lgi->lgi_buf,
+                                                    &lgi->lgi_off, th);
+                               RETURN(rc);
+                       }
+
+                       /* update the header */
+                       lgi->lgi_buf.lb_len = llh->llh_bitmap_offset;
+                       lgi->lgi_buf.lb_buf = llh;
                        rc = dt_record_write(env, o, &lgi->lgi_buf,
                                             &lgi->lgi_off, th);
+                       if (rc != 0)
+                               RETURN(rc);
+
+                       /* update the bitmap */
+                       index = reccookie->lgc_index;
+                       lgi->lgi_off = llh->llh_bitmap_offset +
+                                     (index / (sizeof(*bitmap) * 8)) *
+                                                       sizeof(*bitmap);
+                       lgi->lgi_buf.lb_len = sizeof(*bitmap);
+                       lgi->lgi_buf.lb_buf =
+                                       &bitmap[index/(sizeof(*bitmap)*8)];
+                       rc = dt_record_write(env, o, &lgi->lgi_buf,
+                                            &lgi->lgi_off, th);
+
                        RETURN(rc);
                } else if (loghandle->lgh_cur_idx > 0) {
                        /**