[Cluster-devel] [PATCH] GFS2: aggressively issue revokes in gfs2_log_flush

[Steven Whitehouse <swhiteho@redhat.com>]

Hi,

On Tue, 2013-06-11 at 09:49 -0500, Benjamin Marzinski wrote:
> This patch looks at all the outstanding blocks in all the transactions
> on the log, and moves the completed ones to the ail2 list.  Then it
> issues revokes for these blocks.  This will hopefully speed things up
> in situations where there is a lot of contention for glocks, especially
> if they are acquired serially.
> 
> revoke_lo_before_commit will issue at most one log block's full of these
> preemptive revokes. The amount of reserved log space that
> gfs2_log_reserve() ignores has been incremented to allow for this extra
> block.
> 
Looks good, just a couple of comments...

> Signed-off-by: Benjamin Marzinski <bmarzins@redhat.com>
> ---
>  fs/gfs2/log.c  |   74 +++++++++++++++++++++++++++++++++++++++++++++++++++++----
>  fs/gfs2/log.h  |    1 
>  fs/gfs2/lops.c |    1 
>  3 files changed, 72 insertions(+), 4 deletions(-)
> 
> Index: gfs2-3.0-nmw-130611/fs/gfs2/log.c
> ===================================================================
> --- gfs2-3.0-nmw-130611.orig/fs/gfs2/log.c
> +++ gfs2-3.0-nmw-130611/fs/gfs2/log.c
> @@ -211,15 +211,16 @@ static void gfs2_ail1_empty_one(struct g
>  static int gfs2_ail1_empty(struct gfs2_sbd *sdp)
>  {
>  	struct gfs2_trans *tr, *s;
> +	int oldest_tr = 1;
>  	int ret;
>  
>  	spin_lock(&sdp->sd_ail_lock);
>  	list_for_each_entry_safe_reverse(tr, s, &sdp->sd_ail1_list, tr_list) {
>  		gfs2_ail1_empty_one(sdp, tr);
> -		if (list_empty(&tr->tr_ail1_list))
> +		if (list_empty(&tr->tr_ail1_list) && oldest_tr)
>  			list_move(&tr->tr_list, &sdp->sd_ail2_list);
>  		else
> -			break;
> +			oldest_tr = 0;
>  	}
>  	ret = list_empty(&sdp->sd_ail1_list);
>  	spin_unlock(&sdp->sd_ail_lock);
> @@ -317,7 +318,7 @@ static void ail2_empty(struct gfs2_sbd *
>  
>  int gfs2_log_reserve(struct gfs2_sbd *sdp, unsigned int blks)
>  {
> -	unsigned reserved_blks = 6 * (4096 / sdp->sd_vfs->s_blocksize);
> +	unsigned reserved_blks = 7 * (4096 / sdp->sd_vfs->s_blocksize);
>  	unsigned wanted = blks + reserved_blks;
>  	DEFINE_WAIT(wait);
>  	int did_wait = 0;
> @@ -545,6 +546,72 @@ void gfs2_ordered_del_inode(struct gfs2_
>  	spin_unlock(&sdp->sd_ordered_lock);
>  }
>  
> +void write_revokes(struct gfs2_sbd *sdp)
> +{
Since this is not static, it should really be called gfs2_write_revokes

> +	struct gfs2_trans *tr;
> +	struct gfs2_bufdata *bd, *tmp;
> +	struct buffer_head *bh;
> +	struct gfs2_glock *gl;
> +	int have_revokes = 0;
> +	int max_revokes = (sdp->sd_sb.sb_bsize - sizeof(struct gfs2_log_descriptor)) / sizeof(u64);
> +
> +	gfs2_ail1_empty(sdp);
> +	spin_lock(&sdp->sd_ail_lock);
> +	list_for_each_entry(tr, &sdp->sd_ail1_list, tr_list) {
> +		list_for_each_entry(bd, &tr->tr_ail2_list, bd_ail_st_list) {
> +			if (list_empty(&bd->bd_list)) {
> +				have_revokes = 1;
> +				goto done;
> +			}
> +		}
> +	}
> +done:
> +	spin_unlock(&sdp->sd_ail_lock);
> +	if (have_revokes == 0)
> +		return;
> +	while (sdp->sd_log_num_revoke > max_revokes)
> +		max_revokes += (sdp->sd_sb.sb_bsize - sizeof(struct gfs2_meta_header)) / sizeof(u64);
> +	max_revokes -= sdp->sd_log_num_revoke;
> +	if (!sdp->sd_log_num_revoke) {
> +		atomic_dec(&sdp->sd_log_blks_free);
> +		/* If no blocks have been reserved, we need to also
> +		 * reserve a block for the header */
> +		if (!sdp->sd_log_blks_reserved)
> +			atomic_dec(&sdp->sd_log_blks_free);
> +	}
> +	gfs2_log_lock(sdp);
> +	spin_lock(&sdp->sd_ail_lock);
> +	list_for_each_entry(tr, &sdp->sd_ail1_list, tr_list) {
> +		list_for_each_entry_safe(bd, tmp, &tr->tr_ail2_list, bd_ail_st_list) {
> +			if (max_revokes == 0)
> +				goto out_of_blocks;
> +			if (!list_empty(&bd->bd_list))
> +				continue;
> +			gl = bd->bd_gl;
> +			bh = bd->bd_bh;
> +			bd->bd_blkno = bh->b_blocknr;
> +			bh->b_private = NULL;
> +			gfs2_remove_from_ail(bd);
> +			bd->bd_bh = NULL;
> +			bd->bd_ops = &gfs2_revoke_lops;
> +			sdp->sd_log_num_revoke++;
> +			atomic_inc(&gl->gl_revokes);
> +			set_bit(GLF_LFLUSH, &gl->gl_flags);
> +			list_add_tail(&bd->bd_list, &sdp->sd_log_le_revoke);
This looks very similar to code elsewhere which creates revokes - is it
possible to make this bit into a common function perhaps?


Otherwise I think it looks very good,

Steve.

> +			max_revokes--;
> +		}
> +	}
> +out_of_blocks:
> +	spin_unlock(&sdp->sd_ail_lock);
> +	gfs2_log_unlock(sdp);
> +
> +	if (!sdp->sd_log_num_revoke) {
> +		atomic_inc(&sdp->sd_log_blks_free);
> +		if (!sdp->sd_log_blks_reserved)
> +			atomic_inc(&sdp->sd_log_blks_free);
> +	}
> +}
> +
>  /**
>   * log_write_header - Get and initialize a journal header buffer
>   * @sdp: The GFS2 superblock
> @@ -562,7 +629,6 @@ static void log_write_header(struct gfs2
>  	lh = page_address(page);
>  	clear_page(lh);
>  
> -	gfs2_ail1_empty(sdp);
>  	tail = current_tail(sdp);
>  
>  	lh->lh_header.mh_magic = cpu_to_be32(GFS2_MAGIC);
> Index: gfs2-3.0-nmw-130611/fs/gfs2/log.h
> ===================================================================
> --- gfs2-3.0-nmw-130611.orig/fs/gfs2/log.h
> +++ gfs2-3.0-nmw-130611/fs/gfs2/log.h
> @@ -72,5 +72,6 @@ extern void gfs2_ail1_flush(struct gfs2_
>  extern void gfs2_log_shutdown(struct gfs2_sbd *sdp);
>  extern void gfs2_meta_syncfs(struct gfs2_sbd *sdp);
>  extern int gfs2_logd(void *data);
> +extern void write_revokes(struct gfs2_sbd *sdp);
>  
>  #endif /* __LOG_DOT_H__ */
> Index: gfs2-3.0-nmw-130611/fs/gfs2/lops.c
> ===================================================================
> --- gfs2-3.0-nmw-130611.orig/fs/gfs2/lops.c
> +++ gfs2-3.0-nmw-130611/fs/gfs2/lops.c
> @@ -606,6 +606,7 @@ static void revoke_lo_before_commit(stru
>  	struct page *page;
>  	unsigned int length;
>  
> +	write_revokes(sdp);
>  	if (!sdp->sd_log_num_revoke)
>  		return;
>  
>