[5/8] xfs: factor callbacks out of xlog_state_do_callback()
diff mbox series

Message ID 20190905084717.30308-6-david@fromorbit.com
State New
Headers show
Series
  • [1/8] xfs: push the AIL in xlog_grant_head_wake
Related show

Commit Message

Dave Chinner Sept. 5, 2019, 8:47 a.m. UTC
From: Dave Chinner <dchinner@redhat.com>

Simplify the code flow by lifting the iclog callback work out of
the main iclog iteration loop. This isolates the log juggling and
callbacks from the iclog state change logic in the loop.

Note that the loopdidcallbacks variable is not actually tracking
whether callbacks are actually run - it is tracking whether the
icloglock was dropped during the loop and so determines if we
completed the entire iclog scan loop atomically. Hence we know for
certain there are either no more ordered completions to run or
that the next completion will run the remaining ordered iclog
completions. Hence rename that variable appropriately for it's
function.

Signed-off-by: Dave Chinner <dchinner@redhat.com>
---
 fs/xfs/xfs_log.c | 70 +++++++++++++++++++++++++++++++-----------------
 1 file changed, 45 insertions(+), 25 deletions(-)

Comments

Darrick J. Wong Sept. 5, 2019, 3:39 p.m. UTC | #1
On Thu, Sep 05, 2019 at 06:47:14PM +1000, Dave Chinner wrote:
> From: Dave Chinner <dchinner@redhat.com>
> 
> Simplify the code flow by lifting the iclog callback work out of
> the main iclog iteration loop. This isolates the log juggling and
> callbacks from the iclog state change logic in the loop.
> 
> Note that the loopdidcallbacks variable is not actually tracking
> whether callbacks are actually run - it is tracking whether the
> icloglock was dropped during the loop and so determines if we
> completed the entire iclog scan loop atomically. Hence we know for
> certain there are either no more ordered completions to run or
> that the next completion will run the remaining ordered iclog
> completions. Hence rename that variable appropriately for it's
> function.
> 
> Signed-off-by: Dave Chinner <dchinner@redhat.com>
> ---
>  fs/xfs/xfs_log.c | 70 +++++++++++++++++++++++++++++++-----------------
>  1 file changed, 45 insertions(+), 25 deletions(-)
> 
> diff --git a/fs/xfs/xfs_log.c b/fs/xfs/xfs_log.c
> index 2904bf0d17f3..73aa8e152c83 100644
> --- a/fs/xfs/xfs_log.c
> +++ b/fs/xfs/xfs_log.c
> @@ -2628,6 +2628,42 @@ xlog_get_lowest_lsn(
>  	return lowest_lsn;
>  }
>  
> +/*
> + * Keep processing entries in the iclog callback list until we come around and
> + * it is empty.  We need to atomically see that the list is empty and change the
> + * state to DIRTY so that we don't miss any more callbacks being added.
> + *
> + * This function is called with the icloglock held and returns with it held. We
> + * drop it while running callbacks, however, as holding it over thousands of
> + * callbacks is unnecessary and causes excessive contention if we do.
> + */
> +static void
> +xlog_state_do_iclog_callbacks(
> +	struct xlog		*log,
> +	struct xlog_in_core	*iclog,
> +	bool			aborted)
> +{
> +	spin_unlock(&log->l_icloglock);
> +	spin_lock(&iclog->ic_callback_lock);
> +	while (!list_empty(&iclog->ic_callbacks)) {
> +		LIST_HEAD(tmp);
> +
> +		list_splice_init(&iclog->ic_callbacks, &tmp);
> +
> +		spin_unlock(&iclog->ic_callback_lock);
> +		xlog_cil_process_committed(&tmp, aborted);
> +		spin_lock(&iclog->ic_callback_lock);
> +	}
> +
> +	/*
> +	 * Pick up the icloglock while still holding the callback lock so we
> +	 * serialise against anyone trying to add more callbacks to this iclog
> +	 * now we've finished processing.
> +	 */
> +	spin_lock(&log->l_icloglock);
> +	spin_unlock(&iclog->ic_callback_lock);
> +}
> +
>  #ifdef DEBUG
>  /*
>   * Make one last gasp attempt to see if iclogs are being left in limbo.  If the
> @@ -2682,7 +2718,7 @@ xlog_state_do_callback(
>  	int		   flushcnt = 0;
>  	xfs_lsn_t	   lowest_lsn;
>  	int		   ioerrors;	/* counter: iclogs with errors */
> -	int		   loopdidcallbacks; /* flag: inner loop did callbacks*/
> +	bool		   cycled_icloglock;
>  	int		   funcdidcallbacks; /* flag: function did callbacks */
>  	int		   repeats;	/* for issuing console warnings if
>  					 * looping too many times */
> @@ -2704,7 +2740,7 @@ xlog_state_do_callback(
>  		 */
>  		first_iclog = log->l_iclog;
>  		iclog = log->l_iclog;
> -		loopdidcallbacks = 0;
> +		cycled_icloglock = false;
>  		repeats++;
>  
>  		do {
> @@ -2795,31 +2831,13 @@ xlog_state_do_callback(
>  			} else
>  				ioerrors++;
>  
> -			spin_unlock(&log->l_icloglock);
> -
>  			/*
> -			 * Keep processing entries in the callback list until
> -			 * we come around and it is empty.  We need to
> -			 * atomically see that the list is empty and change the
> -			 * state to DIRTY so that we don't miss any more
> -			 * callbacks being added.
> +			 * Running callbacks will drop the icloglock which means
> +			 * we'll have to run at least one more complete loop.
>  			 */
> -			spin_lock(&iclog->ic_callback_lock);
> -			while (!list_empty(&iclog->ic_callbacks)) {
> -				LIST_HEAD(tmp);
> +			cycled_icloglock = true;
> +			xlog_state_do_iclog_callbacks(log, iclog, aborted);
>  
> -				list_splice_init(&iclog->ic_callbacks, &tmp);
> -
> -				spin_unlock(&iclog->ic_callback_lock);
> -				xlog_cil_process_committed(&tmp, aborted);
> -				spin_lock(&iclog->ic_callback_lock);
> -			}
> -
> -			loopdidcallbacks++;
> -			funcdidcallbacks++;
> -
> -			spin_lock(&log->l_icloglock);
> -			spin_unlock(&iclog->ic_callback_lock);
>  			if (!(iclog->ic_state & XLOG_STATE_IOERROR))
>  				iclog->ic_state = XLOG_STATE_DIRTY;
>  
> @@ -2835,6 +2853,8 @@ xlog_state_do_callback(
>  			iclog = iclog->ic_next;
>  		} while (first_iclog != iclog);
>  
> +		funcdidcallbacks += cycled_icloglock;

funcdidcallbacks is effectively a yes/no state flag, so maybe it should
be turned into a boolean and this statement becomes:

	funcdidcallbacks |= cycled_icloglock;

Though I guess we're not at huge risk of integer overflow and it
controls whether or not we run a debugging check so maybe we don't care?

--D

> +
>  		if (repeats > 5000) {
>  			flushcnt += repeats;
>  			repeats = 0;
> @@ -2842,7 +2862,7 @@ xlog_state_do_callback(
>  				"%s: possible infinite loop (%d iterations)",
>  				__func__, flushcnt);
>  		}
> -	} while (!ioerrors && loopdidcallbacks);
> +	} while (!ioerrors && cycled_icloglock);
>  
>  	if (funcdidcallbacks)
>  		xlog_state_callback_check_state(log);
> -- 
> 2.23.0.rc1
>
Dave Chinner Sept. 5, 2019, 10:17 p.m. UTC | #2
On Thu, Sep 05, 2019 at 08:39:07AM -0700, Darrick J. Wong wrote:
> On Thu, Sep 05, 2019 at 06:47:14PM +1000, Dave Chinner wrote:
> > @@ -2795,31 +2831,13 @@ xlog_state_do_callback(
> >  			} else
> >  				ioerrors++;
> >  
> > -			spin_unlock(&log->l_icloglock);
> > -
> >  			/*
> > -			 * Keep processing entries in the callback list until
> > -			 * we come around and it is empty.  We need to
> > -			 * atomically see that the list is empty and change the
> > -			 * state to DIRTY so that we don't miss any more
> > -			 * callbacks being added.
> > +			 * Running callbacks will drop the icloglock which means
> > +			 * we'll have to run at least one more complete loop.
> >  			 */
> > -			spin_lock(&iclog->ic_callback_lock);
> > -			while (!list_empty(&iclog->ic_callbacks)) {
> > -				LIST_HEAD(tmp);
> > +			cycled_icloglock = true;
> > +			xlog_state_do_iclog_callbacks(log, iclog, aborted);
> >  
> > -				list_splice_init(&iclog->ic_callbacks, &tmp);
> > -
> > -				spin_unlock(&iclog->ic_callback_lock);
> > -				xlog_cil_process_committed(&tmp, aborted);
> > -				spin_lock(&iclog->ic_callback_lock);
> > -			}
> > -
> > -			loopdidcallbacks++;
> > -			funcdidcallbacks++;
> > -
> > -			spin_lock(&log->l_icloglock);
> > -			spin_unlock(&iclog->ic_callback_lock);
> >  			if (!(iclog->ic_state & XLOG_STATE_IOERROR))
> >  				iclog->ic_state = XLOG_STATE_DIRTY;
> >  
> > @@ -2835,6 +2853,8 @@ xlog_state_do_callback(
> >  			iclog = iclog->ic_next;
> >  		} while (first_iclog != iclog);
> >  
> > +		funcdidcallbacks += cycled_icloglock;
> 
> funcdidcallbacks is effectively a yes/no state flag, so maybe it should
> be turned into a boolean and this statement becomes:
> 
> 	funcdidcallbacks |= cycled_icloglock;

Fixed. I renamed it to did_callbacks at the same time, just to be a
little less eye-bleedy...

> Though I guess we're not at huge risk of integer overflow and it
> controls whether or not we run a debugging check so maybe we don't care?

All that really matters is we don't need a branch to calculate it :P

Cheers,

Dave.

Patch
diff mbox series

diff --git a/fs/xfs/xfs_log.c b/fs/xfs/xfs_log.c
index 2904bf0d17f3..73aa8e152c83 100644
--- a/fs/xfs/xfs_log.c
+++ b/fs/xfs/xfs_log.c
@@ -2628,6 +2628,42 @@  xlog_get_lowest_lsn(
 	return lowest_lsn;
 }
 
+/*
+ * Keep processing entries in the iclog callback list until we come around and
+ * it is empty.  We need to atomically see that the list is empty and change the
+ * state to DIRTY so that we don't miss any more callbacks being added.
+ *
+ * This function is called with the icloglock held and returns with it held. We
+ * drop it while running callbacks, however, as holding it over thousands of
+ * callbacks is unnecessary and causes excessive contention if we do.
+ */
+static void
+xlog_state_do_iclog_callbacks(
+	struct xlog		*log,
+	struct xlog_in_core	*iclog,
+	bool			aborted)
+{
+	spin_unlock(&log->l_icloglock);
+	spin_lock(&iclog->ic_callback_lock);
+	while (!list_empty(&iclog->ic_callbacks)) {
+		LIST_HEAD(tmp);
+
+		list_splice_init(&iclog->ic_callbacks, &tmp);
+
+		spin_unlock(&iclog->ic_callback_lock);
+		xlog_cil_process_committed(&tmp, aborted);
+		spin_lock(&iclog->ic_callback_lock);
+	}
+
+	/*
+	 * Pick up the icloglock while still holding the callback lock so we
+	 * serialise against anyone trying to add more callbacks to this iclog
+	 * now we've finished processing.
+	 */
+	spin_lock(&log->l_icloglock);
+	spin_unlock(&iclog->ic_callback_lock);
+}
+
 #ifdef DEBUG
 /*
  * Make one last gasp attempt to see if iclogs are being left in limbo.  If the
@@ -2682,7 +2718,7 @@  xlog_state_do_callback(
 	int		   flushcnt = 0;
 	xfs_lsn_t	   lowest_lsn;
 	int		   ioerrors;	/* counter: iclogs with errors */
-	int		   loopdidcallbacks; /* flag: inner loop did callbacks*/
+	bool		   cycled_icloglock;
 	int		   funcdidcallbacks; /* flag: function did callbacks */
 	int		   repeats;	/* for issuing console warnings if
 					 * looping too many times */
@@ -2704,7 +2740,7 @@  xlog_state_do_callback(
 		 */
 		first_iclog = log->l_iclog;
 		iclog = log->l_iclog;
-		loopdidcallbacks = 0;
+		cycled_icloglock = false;
 		repeats++;
 
 		do {
@@ -2795,31 +2831,13 @@  xlog_state_do_callback(
 			} else
 				ioerrors++;
 
-			spin_unlock(&log->l_icloglock);
-
 			/*
-			 * Keep processing entries in the callback list until
-			 * we come around and it is empty.  We need to
-			 * atomically see that the list is empty and change the
-			 * state to DIRTY so that we don't miss any more
-			 * callbacks being added.
+			 * Running callbacks will drop the icloglock which means
+			 * we'll have to run at least one more complete loop.
 			 */
-			spin_lock(&iclog->ic_callback_lock);
-			while (!list_empty(&iclog->ic_callbacks)) {
-				LIST_HEAD(tmp);
+			cycled_icloglock = true;
+			xlog_state_do_iclog_callbacks(log, iclog, aborted);
 
-				list_splice_init(&iclog->ic_callbacks, &tmp);
-
-				spin_unlock(&iclog->ic_callback_lock);
-				xlog_cil_process_committed(&tmp, aborted);
-				spin_lock(&iclog->ic_callback_lock);
-			}
-
-			loopdidcallbacks++;
-			funcdidcallbacks++;
-
-			spin_lock(&log->l_icloglock);
-			spin_unlock(&iclog->ic_callback_lock);
 			if (!(iclog->ic_state & XLOG_STATE_IOERROR))
 				iclog->ic_state = XLOG_STATE_DIRTY;
 
@@ -2835,6 +2853,8 @@  xlog_state_do_callback(
 			iclog = iclog->ic_next;
 		} while (first_iclog != iclog);
 
+		funcdidcallbacks += cycled_icloglock;
+
 		if (repeats > 5000) {
 			flushcnt += repeats;
 			repeats = 0;
@@ -2842,7 +2862,7 @@  xlog_state_do_callback(
 				"%s: possible infinite loop (%d iterations)",
 				__func__, flushcnt);
 		}
-	} while (!ioerrors && loopdidcallbacks);
+	} while (!ioerrors && cycled_icloglock);
 
 	if (funcdidcallbacks)
 		xlog_state_callback_check_state(log);