[Cluster-devel] [PATCH 16/19] gfs2: Only remove revokes that we've submitted

[Bob Peterson <rpeterso@redhat.com>]

Before this patch function revoke_lo_after_commit would run the
list of revokes (sd_log_num_revoke) and free them all. That assumes
they've all been submitted. When revoke_lo_before_commit is called
by the log flush, that's true. But then all the io is submitted,
which could take some time. In the meantime, another process might
be able to add new revokes to the list, and those might be wiped
out. Also, since both functions run the list run head to tail,
function gfs2_add_revoke should add to the tail, not the head.

Signed-off-by: Bob Peterson <rpeterso@redhat.com>
---
 fs/gfs2/incore.h |  1 +
 fs/gfs2/log.c    |  4 +++-
 fs/gfs2/lops.c   | 13 +++++++++++--
 3 files changed, 15 insertions(+), 3 deletions(-)

diff --git a/fs/gfs2/incore.h b/fs/gfs2/incore.h
index bcda69880ec1..7e1af6bc5990 100644
--- a/fs/gfs2/incore.h
+++ b/fs/gfs2/incore.h
@@ -814,6 +814,7 @@ struct gfs2_sbd {
 
 	atomic_t sd_log_pinned;
 	unsigned int sd_log_num_revoke;
+	unsigned int sd_log_num_revoked; /* number of revokes written */
 
 	struct list_head sd_log_le_revoke;
 	struct list_head sd_log_le_ordered;
diff --git a/fs/gfs2/log.c b/fs/gfs2/log.c
index e573bdbb9634..2dc0a6b2e0c1 100644
--- a/fs/gfs2/log.c
+++ b/fs/gfs2/log.c
@@ -617,7 +617,7 @@ void gfs2_add_revoke(struct gfs2_sbd *sdp, struct gfs2_bufdata *bd)
 	sdp->sd_log_num_revoke++;
 	atomic_inc(&gl->gl_revokes);
 	set_bit(GLF_LFLUSH, &gl->gl_flags);
-	list_add(&bd->bd_list, &sdp->sd_log_le_revoke);
+	list_add_tail(&bd->bd_list, &sdp->sd_log_le_revoke);
 }
 
 void gfs2_write_revokes(struct gfs2_sbd *sdp)
@@ -627,6 +627,8 @@ void gfs2_write_revokes(struct gfs2_sbd *sdp)
 	gfs2_log_lock(sdp);
 	while (sdp->sd_log_num_revoke > max_revokes)
 		max_revokes += (sdp->sd_sb.sb_bsize - sizeof(struct gfs2_meta_header)) / sizeof(u64);
+	/* Subtract out the revokes we already have queued, to get the number
+	 * of additional revokes we can add at this time. */
 	max_revokes -= sdp->sd_log_num_revoke;
 	if (!sdp->sd_log_num_revoke) {
 		atomic_dec(&sdp->sd_log_blks_free);
diff --git a/fs/gfs2/lops.c b/fs/gfs2/lops.c
index 11d2fdeed8ac..f442ef60431d 100644
--- a/fs/gfs2/lops.c
+++ b/fs/gfs2/lops.c
@@ -656,9 +656,10 @@ static void revoke_lo_before_commit(struct gfs2_sbd *sdp, struct gfs2_trans *tr)
 	length = gfs2_struct2blk(sdp, sdp->sd_log_num_revoke, sizeof(u64));
 	page = gfs2_get_log_desc(sdp, GFS2_LOG_DESC_REVOKE, length, sdp->sd_log_num_revoke);
 	offset = sizeof(struct gfs2_log_descriptor);
+	sdp->sd_log_num_revoked = 0;
 
 	list_for_each_entry(bd, head, bd_list) {
-		sdp->sd_log_num_revoke--;
+		sdp->sd_log_num_revoked++;
 
 		if (offset + sizeof(u64) > sdp->sd_sb.sb_bsize) {
 
@@ -675,7 +676,8 @@ static void revoke_lo_before_commit(struct gfs2_sbd *sdp, struct gfs2_trans *tr)
 		*(__be64 *)(page_address(page) + offset) = cpu_to_be64(bd->bd_blkno);
 		offset += sizeof(u64);
 	}
-	gfs2_assert_withdraw(sdp, !sdp->sd_log_num_revoke);
+	gfs2_assert_withdraw(sdp, sdp->sd_log_num_revoke ==
+			     sdp->sd_log_num_revoked);
 
 	gfs2_log_write_page(sdp, page);
 }
@@ -686,6 +688,8 @@ static void revoke_lo_after_commit(struct gfs2_sbd *sdp, struct gfs2_trans *tr)
 	struct gfs2_bufdata *bd;
 	struct gfs2_glock *gl;
 
+	BUG_ON(sdp->sd_log_num_revoked > sdp->sd_log_num_revoke);
+
 	while (!list_empty(head)) {
 		bd = list_entry(head->next, struct gfs2_bufdata, bd_list);
 		list_del_init(&bd->bd_list);
@@ -693,7 +697,12 @@ static void revoke_lo_after_commit(struct gfs2_sbd *sdp, struct gfs2_trans *tr)
 		atomic_dec(&gl->gl_revokes);
 		clear_bit(GLF_LFLUSH, &gl->gl_flags);
 		kmem_cache_free(gfs2_bufdata_cachep, bd);
+		sdp->sd_log_num_revoke--;
+		sdp->sd_log_num_revoked--;
+		if (sdp->sd_log_num_revoked == 0)
+			break;
 	}
+	WARN_ON(sdp->sd_log_num_revoke);
 }
 
 static void revoke_lo_before_scan(struct gfs2_jdesc *jd,
-- 
2.20.1