Re: [PATCH 3/3] xfs: flush the CIL via a workqueue

[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

 



Did you plan to resubmit a version of this?

On Mon, Jul 18, 2011 at 01:49:49PM +1000, Dave Chinner wrote:
> From: Dave Chinner <dchinner@xxxxxxxxxx>
> 
> Doing background CIL flushes adds significant latency to whatever
> async transaction that triggers it. To avoid blocking async
> transactions on things like waiting for log buffer IO to complete,
> move the CIL push off into a workqueue.  By moving the push work
> into a workqueue, we remove all the latency that the commit adds
> from the foreground transaction commit path. This also means that
> single threaded workloads won't do the CIL push procssing, leaving
> them more CPU to do more async transactions.
> 
> To do this, we need to keep track of the sequence nnumber we have
> pushed work for. This avoids having many transaction commits
> attempting to schedule work for the same sequence, and ensures that
> we only ever have one push (background or forced) in progress at a
> time. It also means that we don't need to take the CIL lock in write
> mode to check for potential background push races, which reduces
> lock contention.
> 
> Signed-off-by: Dave Chinner <dchinner@xxxxxxxxxx>
> ---
>  fs/xfs/linux-2.6/xfs_super.c |    7 +
>  fs/xfs/xfs_log_cil.c         |  260 +++++++++++++++++++++++++-----------------
>  fs/xfs/xfs_log_priv.h        |    4 +
>  3 files changed, 166 insertions(+), 105 deletions(-)
> 
> diff --git a/fs/xfs/linux-2.6/xfs_super.c b/fs/xfs/linux-2.6/xfs_super.c
> index 6a6d4d9..b3ace86 100644
> --- a/fs/xfs/linux-2.6/xfs_super.c
> +++ b/fs/xfs/linux-2.6/xfs_super.c
> @@ -1683,8 +1683,14 @@ xfs_init_workqueues(void)
>  	if (!xfs_alloc_wq)
>  		goto out_destroy_ail;
>  
> +	xfs_cil_wq = alloc_workqueue("xfscil", WQ_MEM_RECLAIM, 8);
> +	if (!xfs_ail_wq)
> +		goto out_destroy_alloc;
> +
>  	return 0;
>  
> +out_destroy_alloc:
> +	destroy_workqueue(xfs_alloc_wq);
>  out_destroy_ail:
>  	destroy_workqueue(xfs_ail_wq);
>  out_destroy_syncd:
> @@ -1696,6 +1702,7 @@ out:
>  STATIC void
>  xfs_destroy_workqueues(void)
>  {
> +	destroy_workqueue(xfs_cil_wq);
>  	destroy_workqueue(xfs_ail_wq);
>  	destroy_workqueue(xfs_syncd_wq);
>  }
> diff --git a/fs/xfs/xfs_log_cil.c b/fs/xfs/xfs_log_cil.c
> index c7755d5..9e652d2 100644
> --- a/fs/xfs/xfs_log_cil.c
> +++ b/fs/xfs/xfs_log_cil.c
> @@ -31,67 +31,7 @@
>  #include "xfs_alloc.h"
>  #include "xfs_discard.h"
>  
> -/*
> - * Perform initial CIL structure initialisation. If the CIL is not
> - * enabled in this filesystem, ensure the log->l_cilp is null so
> - * we can check this conditional to determine if we are doing delayed
> - * logging or not.
> - */
> -int
> -xlog_cil_init(
> -	struct log	*log)
> -{
> -	struct xfs_cil	*cil;
> -	struct xfs_cil_ctx *ctx;
> -
> -	log->l_cilp = NULL;
> -	if (!(log->l_mp->m_flags & XFS_MOUNT_DELAYLOG))
> -		return 0;
> -
> -	cil = kmem_zalloc(sizeof(*cil), KM_SLEEP|KM_MAYFAIL);
> -	if (!cil)
> -		return ENOMEM;
> -
> -	ctx = kmem_zalloc(sizeof(*ctx), KM_SLEEP|KM_MAYFAIL);
> -	if (!ctx) {
> -		kmem_free(cil);
> -		return ENOMEM;
> -	}
> -
> -	INIT_LIST_HEAD(&cil->xc_cil);
> -	INIT_LIST_HEAD(&cil->xc_committing);
> -	spin_lock_init(&cil->xc_cil_lock);
> -	init_rwsem(&cil->xc_ctx_lock);
> -	init_waitqueue_head(&cil->xc_commit_wait);
> -
> -	INIT_LIST_HEAD(&ctx->committing);
> -	INIT_LIST_HEAD(&ctx->busy_extents);
> -	ctx->sequence = 1;
> -	ctx->cil = cil;
> -	cil->xc_ctx = ctx;
> -	cil->xc_current_sequence = ctx->sequence;
> -
> -	cil->xc_log = log;
> -	log->l_cilp = cil;
> -	return 0;
> -}
> -
> -void
> -xlog_cil_destroy(
> -	struct log	*log)
> -{
> -	if (!log->l_cilp)
> -		return;
> -
> -	if (log->l_cilp->xc_ctx) {
> -		if (log->l_cilp->xc_ctx->ticket)
> -			xfs_log_ticket_put(log->l_cilp->xc_ctx->ticket);
> -		kmem_free(log->l_cilp->xc_ctx);
> -	}
> -
> -	ASSERT(list_empty(&log->l_cilp->xc_cil));
> -	kmem_free(log->l_cilp);
> -}
> +struct workqueue_struct *xfs_cil_wq;
>  
>  /*
>   * Allocate a new ticket. Failing to get a new ticket makes it really hard to
> @@ -401,12 +341,58 @@ xlog_cil_committed(
>   * get a race between multiple pushes for the same sequence they will block on
>   * the first one and then abort, hence avoiding needless pushes.
>   */
> -STATIC int
> +static void
>  xlog_cil_push(
> -	struct log		*log,
> -	xfs_lsn_t		push_seq)
> +	struct log	*log,
> +	xfs_lsn_t	push_seq)
>  {
> -	struct xfs_cil		*cil = log->l_cilp;
> +	struct xfs_cil	*cil = log->l_cilp;
> +
> +	if (!cil)
> +		return;
> +
> +	ASSERT(!push_seq || push_seq <= cil->xc_current_sequence);
> +
> +	/*
> +	 * don't do a background push if we haven't used up all the
> +	 * space available yet.
> +	 */
> +	if (!push_seq && cil->xc_ctx->space_used < XLOG_CIL_SPACE_LIMIT(log))
> +		return;
> +
> +	/*
> +	 * if we are being asked to push to a specific sequence, and we have
> +	 * already queued a larger push, then nothing to do.
> +	 */
> +	if (push_seq && push_seq <= cil->xc_push_seq)
> +		return;
> +
> +	spin_lock(&cil->xc_cil_lock);
> +	if (!push_seq)
> +		push_seq = cil->xc_current_sequence;
> +
> +	/*
> +	 * if the CIL is empty,  or we've already pushed the sequence, then
> +	 * there's no work we need to do.
> +	 */
> +	if (list_empty(&cil->xc_cil) || push_seq <= cil->xc_push_seq) {
> +		spin_unlock(&cil->xc_cil_lock);
> +		return;
> +	}
> +
> +	cil->xc_push_seq = push_seq;
> +	queue_work(xfs_cil_wq, &cil->xc_push_work);
> +	spin_unlock(&cil->xc_cil_lock);
> +
> +}
> +
> +static void
> +xlog_cil_push_work(
> +	struct work_struct	*work)
> +{
> +	struct xfs_cil		*cil = container_of(work, struct xfs_cil,
> +							xc_push_work);
> +	struct log		*log = cil->xc_log;
>  	struct xfs_log_vec	*lv;
>  	struct xfs_cil_ctx	*ctx;
>  	struct xfs_cil_ctx	*new_ctx;
> @@ -419,40 +405,34 @@ xlog_cil_push(
>  	struct xfs_trans_header thdr;
>  	struct xfs_log_iovec	lhdr;
>  	struct xfs_log_vec	lvhdr = { NULL };
> +	xfs_lsn_t		push_seq;
>  	xfs_lsn_t		commit_lsn;
>  
> -	if (!cil)
> -		return 0;
> -
> -	ASSERT(!push_seq || push_seq <= cil->xc_ctx->sequence);
> -
>  	new_ctx = kmem_zalloc(sizeof(*new_ctx), KM_SLEEP|KM_NOFS);
>  	new_ctx->ticket = xlog_cil_ticket_alloc(log);
>  
> -	/*
> -	 * Lock out transaction commit, but don't block for background pushes
> -	 * unless we are well over the CIL space limit. See the definition of
> -	 * XLOG_CIL_HARD_SPACE_LIMIT() for the full explanation of the logic
> -	 * used here.
> -	 */
> -	if (!down_write_trylock(&cil->xc_ctx_lock)) {
> -		if (!push_seq &&
> -		    cil->xc_ctx->space_used < XLOG_CIL_HARD_SPACE_LIMIT(log))
> -			goto out_free_ticket;
> -		down_write(&cil->xc_ctx_lock);
> -	}
> +	/* Lock out transaction commiti until we've switch contexts */
> +	down_write(&cil->xc_ctx_lock);
>  	ctx = cil->xc_ctx;
>  
> -	/* check if we've anything to push */
> -	if (list_empty(&cil->xc_cil))
> -		goto out_skip;
> +	spin_lock(&cil->xc_cil_lock);
> +	push_seq = cil->xc_push_seq;
> +	ASSERT(push_seq > 0 && push_seq <= ctx->sequence);
>  
> -	/* check for spurious background flush */
> -	if (!push_seq && cil->xc_ctx->space_used < XLOG_CIL_SPACE_LIMIT(log))
> +	/*
> +	 * Check if we've anything to push. If there is nothing, then we don't
> +	 * move on to a new sequence number and so we have to be able to push
> +	 * this sequence again later.
> +	 */
> +	if (list_empty(&cil->xc_cil)) {
> +		cil->xc_push_seq = 0;
> +		spin_unlock(&cil->xc_cil_lock);
>  		goto out_skip;
> +	}
> +	spin_unlock(&cil->xc_cil_lock);
>  
>  	/* check for a previously pushed seqeunce */
> -	if (push_seq && push_seq < cil->xc_ctx->sequence)
> +	if (push_seq < ctx->sequence)
>  		goto out_skip;
>  
>  	/*
> @@ -602,20 +582,19 @@ restart:
>  	spin_unlock(&cil->xc_cil_lock);
>  
>  	/* release the hounds! */
> -	return xfs_log_release_iclog(log->l_mp, commit_iclog);
> +	xfs_log_release_iclog(log->l_mp, commit_iclog);
> +	return;
>  
>  out_skip:
>  	up_write(&cil->xc_ctx_lock);
> -out_free_ticket:
>  	xfs_log_ticket_put(new_ctx->ticket);
>  	kmem_free(new_ctx);
> -	return 0;
> +	return;
>  
>  out_abort_free_ticket:
>  	xfs_log_ticket_put(tic);
>  out_abort:
>  	xlog_cil_committed(ctx, XFS_LI_ABORTED);
> -	return XFS_ERROR(EIO);
>  }
>  
>  /*
> @@ -645,7 +624,6 @@ xfs_log_commit_cil(
>  {
>  	struct log		*log = mp->m_log;
>  	int			log_flags = 0;
> -	int			push = 0;
>  
>  	if (flags & XFS_TRANS_RELEASE_LOG_RES)
>  		log_flags = XFS_LOG_REL_PERM_RESERV;
> @@ -694,12 +672,6 @@ xfs_log_commit_cil(
>  	 */
>  	xfs_trans_free_items(tp, *commit_lsn, 0);
>  
> -	/* check for background commit before unlock */
> -	if (log->l_cilp->xc_ctx->space_used > XLOG_CIL_SPACE_LIMIT(log))
> -		push = 1;
> -
> -	up_read(&log->l_cilp->xc_ctx_lock);
> -
>  	/*
>  	 * We need to push CIL every so often so we don't cache more than we
>  	 * can fit in the log. The limit really is that a checkpoint can't be
> @@ -707,8 +679,8 @@ xfs_log_commit_cil(
>  	 * overwrite the previous checkpoint), but commit latency and memory
>  	 * usage limit this to a smaller size in most cases.
>  	 */
> -	if (push)
> -		xlog_cil_push(log, 0);
> +	xlog_cil_push(log, 0);
> +	up_read(&log->l_cilp->xc_ctx_lock);
>  }
>  
>  /*
> @@ -720,9 +692,6 @@ xfs_log_commit_cil(
>   *
>   * We return the current commit lsn to allow the callers to determine if a
>   * iclog flush is necessary following this call.
> - *
> - * XXX: Initially, just push the CIL unconditionally and return whatever
> - * commit lsn is there. It'll be empty, so this is broken for now.
>   */
>  xfs_lsn_t
>  xlog_cil_force_lsn(
> @@ -733,6 +702,8 @@ xlog_cil_force_lsn(
>  	struct xfs_cil_ctx	*ctx;
>  	xfs_lsn_t		commit_lsn = NULLCOMMITLSN;
>  
> +	/* lock out background commit */
> +	down_read(&log->l_cilp->xc_ctx_lock);
>  	ASSERT(sequence <= cil->xc_current_sequence);
>  
>  	/*
> @@ -740,8 +711,23 @@ xlog_cil_force_lsn(
>  	 * xlog_cil_push() handles racing pushes for the same sequence,
>  	 * so no need to deal with it here.
>  	 */
> -	if (sequence == cil->xc_current_sequence)
> +	if (sequence == cil->xc_current_sequence) {
>  		xlog_cil_push(log, sequence);
> +		up_read(&log->l_cilp->xc_ctx_lock);
> +
> +		/*
> +		 * We have to block waiting for the push to execute even if we
> +		 * didn't push the sequence out as we need to wait for the push
> +		 * to get queued into the committing list. Once it is in the
> +		 * committing list, we can harvest the commit_lsn of the
> +		 * checkpoint issued by the push.
> +		 *
> +		 * We don't hold the ctx lock while doing this as the push work
> +		 * needs to hold it.
> +		 */
> +		flush_work_sync(&cil->xc_push_work);
> +	} else
> +		up_read(&log->l_cilp->xc_ctx_lock);
>  
>  	/*
>  	 * See if we can find a previous sequence still committing.
> @@ -802,3 +788,67 @@ xfs_log_item_in_current_chkpt(
>  		return false;
>  	return true;
>  }
> +
> +/*
> + * Perform initial CIL structure initialisation. If the CIL is not
> + * enabled in this filesystem, ensure the log->l_cilp is null so
> + * we can check this conditional to determine if we are doing delayed
> + * logging or not.
> + */
> +int
> +xlog_cil_init(
> +	struct log	*log)
> +{
> +	struct xfs_cil	*cil;
> +	struct xfs_cil_ctx *ctx;
> +
> +	log->l_cilp = NULL;
> +	if (!(log->l_mp->m_flags & XFS_MOUNT_DELAYLOG))
> +		return 0;
> +
> +	cil = kmem_zalloc(sizeof(*cil), KM_SLEEP|KM_MAYFAIL);
> +	if (!cil)
> +		return ENOMEM;
> +
> +	ctx = kmem_zalloc(sizeof(*ctx), KM_SLEEP|KM_MAYFAIL);
> +	if (!ctx) {
> +		kmem_free(cil);
> +		return ENOMEM;
> +	}
> +
> +	INIT_WORK(&cil->xc_push_work, xlog_cil_push_work);
> +	INIT_LIST_HEAD(&cil->xc_cil);
> +	INIT_LIST_HEAD(&cil->xc_committing);
> +	spin_lock_init(&cil->xc_cil_lock);
> +	init_rwsem(&cil->xc_ctx_lock);
> +	init_waitqueue_head(&cil->xc_commit_wait);
> +
> +	INIT_LIST_HEAD(&ctx->committing);
> +	INIT_LIST_HEAD(&ctx->busy_extents);
> +	ctx->sequence = 1;
> +	ctx->cil = cil;
> +	cil->xc_ctx = ctx;
> +	cil->xc_current_sequence = ctx->sequence;
> +
> +	cil->xc_log = log;
> +	log->l_cilp = cil;
> +	return 0;
> +}
> +
> +void
> +xlog_cil_destroy(
> +	struct log	*log)
> +{
> +	if (!log->l_cilp)
> +		return;
> +
> +	if (log->l_cilp->xc_ctx) {
> +		if (log->l_cilp->xc_ctx->ticket)
> +			xfs_log_ticket_put(log->l_cilp->xc_ctx->ticket);
> +		kmem_free(log->l_cilp->xc_ctx);
> +	}
> +
> +	ASSERT(list_empty(&log->l_cilp->xc_cil));
> +	kmem_free(log->l_cilp);
> +}
> +
> diff --git a/fs/xfs/xfs_log_priv.h b/fs/xfs/xfs_log_priv.h
> index 2d3b6a4..61d55f9 100644
> --- a/fs/xfs/xfs_log_priv.h
> +++ b/fs/xfs/xfs_log_priv.h
> @@ -417,8 +417,12 @@ struct xfs_cil {
>  	struct list_head	xc_committing;
>  	wait_queue_head_t	xc_commit_wait;
>  	xfs_lsn_t		xc_current_sequence;
> +	struct work_struct	xc_push_work;
> +	xfs_lsn_t		xc_push_seq;
>  };
>  
> +extern struct workqueue_struct *xfs_cil_wq;
> +
>  /*
>   * The amount of log space we allow the CIL to aggregate is difficult to size.
>   * Whatever we choose, we have to make sure we can get a reservation for the
> -- 
> 1.7.5.1
> 
> _______________________________________________
> xfs mailing list
> xfs@xxxxxxxxxxx
> http://oss.sgi.com/mailman/listinfo/xfs
---end quoted text---

_______________________________________________
xfs mailing list
xfs@xxxxxxxxxxx
http://oss.sgi.com/mailman/listinfo/xfs


[Index of Archives]     [Linux XFS Devel]     [Linux Filesystem Development]     [Filesystem Testing]     [Linux USB Devel]     [Linux Audio Users]     [Yosemite News]     [Linux Kernel]     [Linux SCSI]

  Powered by Linux