From: wang di <di.wang@intel.com>
Date: Wed, 5 Aug 2015 08:03:42 +0000 (-0700)
Subject: LU-6968 llog: Only update necessary part of header
X-Git-Tag: 2.7.60~56
X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=commitdiff_plain;h=339e9adc2ce671c948c0f278f522fa62e51d8371

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
--- 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
--- 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
--- 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) {
 			/**