| Message ID | 20190905084717.30308-6-david@fromorbit.com (mailing list archive) |
|---|---|
| State | Accepted |
| Headers | show |
| Series | [1/8] xfs: push the AIL in xlog_grant_head_wake | expand |
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 >
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.
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);