xfs: clean up xfs_buf io interfaces

From: Dave Chinner <[email protected]>

When synchronous IO runs IO completion work, it does so without an IO reference or a hold reference on the buffer. The IO "hold reference" is owned by the submitter, and released when the submission is complete. The IO reference is released when both the submitter and the bio end_io processing is run, and so if the io completion work is run from IO completion context, it is run without an IO reference.

Hence we can get the situation where the submitter can submit the IO, see an error on the buffer and unlock and free the buffer while there is still IO in progress. This leads to use-after-free and memory corruption.

Fix this by taking a "sync IO hold" reference that is owned by the IO and not released until after the buffer completion calls are run to wake up synchronous waiters. This means that the buffer will not be freed in any circumstance until all IO processing is completed.

Signed-off-by Dave Chinner <[email protected]>

---
 fs/xfs/xfs_buf.c | 46 ++++++++++++++++++++++++++++++++++++++--------
 1 file changed, 38 insertions(+), 8 deletions(-)

fs/xfs/xfs_buf.c +0 -0

index cd7b8ca..5d86bbd 100644

fs/xfs/xfs_buf.c +24 -6

 	}
 }
 

@@ -1383,22 +1385,50 @@ xfs_buf_iorequest(
 
 	if (bp->b_flags & XBF_WRITE)
 		xfs_buf_wait_unpin(bp);
+
+	/*
+	 * Take references to the buffer. For XBF_ASYNC buffers, holding a
+	 * reference for as long as submission takes is all that is necessary
+	 * here. The IO inherits the lock and hold count from the submitter,
+	 * and these are release during IO completion processing. Taking a hold
+	 * over submission ensures that the buffer is not freed until we have
+	 * completed all processing, regardless of when IO errors occur or are
+	 * reported.
+	 *
+	 * However, for synchronous IO, the IO does not inherit the submitters
+	 * reference count, nor the buffer lock. Hence we need to take an extra
+	 * reference to the buffer for the for the IO context so that we can
+	 * guarantee the buffer is not freed until all IO completion processing
+	 * is done. Otherwise the caller can drop their reference while the IO
+	 * is still in progress and hence trigger a use-after-free situation.
+	 */
 	xfs_buf_hold(bp);
+	if (!(bp->b_flags & XBF_ASYNC))
+		xfs_buf_hold(bp);
+
 
 	/*
-	 * Set the count to 1 initially, this will stop an I/O
-	 * completion callout which happens before we have started
-	 * all the I/O from calling xfs_buf_ioend too early.
+	 * Set the count to 1 initially, this will stop an I/O completion
+	 * callout which happens before we have started all the I/O from calling
+	 * xfs_buf_ioend too early.
 	 */
 	atomic_set(&bp->b_io_remaining, 1);
 	_xfs_buf_ioapply(bp);
+
 	/*
-	 * If _xfs_buf_ioapply failed, we'll get back here with
-	 * only the reference we took above.  _xfs_buf_ioend will
-	 * drop it to zero, so we'd better not queue it for later,

fs/xfs/xfs_buf.c +0 -0

 	xfs_buf_rele(bp);
 }

Signature · 6 lines

2.0.0

_______________________________________________
xfs mailing list
[email protected]
http://oss.sgi.com/mailman/listinfo/xfs

On Fri, Aug 15, 2014 at 04:38:59PM +1000, Dave Chinner wrote:

Quoted 44 lines

fs/xfs/xfs_buf.c +28 -7

 	if (bp->b_flags & XBF_WRITE)
 		xfs_buf_wait_unpin(bp);
+
+	/*
+	 * Take references to the buffer. For XBF_ASYNC buffers, holding a
+	 * reference for as long as submission takes is all that is necessary
+	 * here. The IO inherits the lock and hold count from the submitter,
+	 * and these are release during IO completion processing. Taking a hold
+	 * over submission ensures that the buffer is not freed until we have
+	 * completed all processing, regardless of when IO errors occur or are
+	 * reported.
+	 *
+	 * However, for synchronous IO, the IO does not inherit the submitters
+	 * reference count, nor the buffer lock. Hence we need to take an extra
+	 * reference to the buffer for the for the IO context so that we can
+	 * guarantee the buffer is not freed until all IO completion processing
+	 * is done. Otherwise the caller can drop their reference while the IO
+	 * is still in progress and hence trigger a use-after-free situation.
+	 */
 	xfs_buf_hold(bp);
+	if (!(bp->b_flags & XBF_ASYNC))
+		xfs_buf_hold(bp);
+
 
 	/*
-	 * Set the count to 1 initially, this will stop an I/O
-	 * completion callout which happens before we have started
-	 * all the I/O from calling xfs_buf_ioend too early.
+	 * Set the count to 1 initially, this will stop an I/O completion
+	 * callout which happens before we have started all the I/O from calling
+	 * xfs_buf_ioend too early.
 	 */
 	atomic_set(&bp->b_io_remaining, 1);
 	_xfs_buf_ioapply(bp);
+
 	/*
-	 * If _xfs_buf_ioapply failed, we'll get back here with
-	 * only the reference we took above.  _xfs_buf_ioend will
-	 * drop it to zero, so we'd better not queue it for later,
-	 * or we'll free it before it's done.
+	 * If _xfs_buf_ioapply failed or we are doing synchronous IO that
+	 * completes extremely quickly, we can get back here with only the IO
+	 * reference we took above.  _xfs_buf_ioend will drop it to zero, so
+	 * we'd better run completion processing synchronously so that the we

fs/xfs/xfs_buf.c xfs_buf_iorequest( · line 1427

Quoted 5 lines

fs/xfs/xfs_buf.c +4 -0

+	 * don't return to the caller with completion still pending. In the
+	 * error case, this allows the caller to check b_error safely without
+	 * waiting, and in the synchronous IO case it avoids unnecessary context
+	 * switches an latency for high-peformance devices.
 	 */

AFAICT there is no real wait if the buf has completed at this point. The wait just decrements the completion counter.

So what's the benefit of
"not waiting?" Where is the potential context switch?

Are you referring
to the case where error is set but I/O is not complete? Are you saying the advantage to the caller is it doesn't have to care about the state of further I/O once it has been determined at least one error has occurred? (If so, who cares about latency given that some operation that depends on this I/O is already doomed to fail?).

The code looks fine, but I'm trying to understand the reasoning better (and I suspect we can clarify the comment).

fs/xfs/xfs_buf.c xfs_buf_iorequest( · line 1431

Quoted 5 lines

fs/xfs/xfs_buf.c +4 -1

-	_xfs_buf_ioend(bp, bp->b_error ? 0 : 1);
+	if (bp->b_error || !(bp->b_flags & XBF_ASYNC))
+		_xfs_buf_ioend(bp, 0);
+	else
+		_xfs_buf_ioend(bp, 1);

Not related to this patch, but it seems like the problem this code tries to address is still possible.

Perhaps this papers over a particular
instance. Consider the case where an I/O fails immediately after this call completes, but not before. We have an extra reference now for completion, but we can still return to the caller with completion pending. I suppose its fine if we consider the "problem" to be that the reference goes away underneath the completion, as opposed to the caller caring about the status of completion.

Quoted 1 line · Brian Foster

On Fri, Aug 15, 2014 at 04:38:59PM +1000, Dave Chinner wrote:

....

Quoted 44 lines

Quoted 44 lines

fs/xfs/xfs_buf.c +28 -7

 	if (bp->b_flags & XBF_WRITE)
 		xfs_buf_wait_unpin(bp);
+
+	/*
+	 * Take references to the buffer. For XBF_ASYNC buffers, holding a
+	 * reference for as long as submission takes is all that is necessary
+	 * here. The IO inherits the lock and hold count from the submitter,
+	 * and these are release during IO completion processing. Taking a hold
+	 * over submission ensures that the buffer is not freed until we have
+	 * completed all processing, regardless of when IO errors occur or are
+	 * reported.
+	 *
+	 * However, for synchronous IO, the IO does not inherit the submitters
+	 * reference count, nor the buffer lock. Hence we need to take an extra
+	 * reference to the buffer for the for the IO context so that we can
+	 * guarantee the buffer is not freed until all IO completion processing
+	 * is done. Otherwise the caller can drop their reference while the IO
+	 * is still in progress and hence trigger a use-after-free situation.
+	 */
 	xfs_buf_hold(bp);
+	if (!(bp->b_flags & XBF_ASYNC))
+		xfs_buf_hold(bp);
+
 
 	/*
-	 * Set the count to 1 initially, this will stop an I/O
-	 * completion callout which happens before we have started
-	 * all the I/O from calling xfs_buf_ioend too early.
+	 * Set the count to 1 initially, this will stop an I/O completion
+	 * callout which happens before we have started all the I/O from calling
+	 * xfs_buf_ioend too early.
 	 */
 	atomic_set(&bp->b_io_remaining, 1);
 	_xfs_buf_ioapply(bp);
+
 	/*
-	 * If _xfs_buf_ioapply failed, we'll get back here with
-	 * only the reference we took above.  _xfs_buf_ioend will
-	 * drop it to zero, so we'd better not queue it for later,
-	 * or we'll free it before it's done.
+	 * If _xfs_buf_ioapply failed or we are doing synchronous IO that
+	 * completes extremely quickly, we can get back here with only the IO
+	 * reference we took above.  _xfs_buf_ioend will drop it to zero, so
+	 * we'd better run completion processing synchronously so that the we

Quoted 8 lines

fs/xfs/xfs_buf.c xfs_buf_iorequest( · line 1427

Quoted 5 lines

fs/xfs/xfs_buf.c +4 -0

+	 * don't return to the caller with completion still pending. In the
+	 * error case, this allows the caller to check b_error safely without
+	 * waiting, and in the synchronous IO case it avoids unnecessary context
+	 * switches an latency for high-peformance devices.
 	 */

AFAICT there is no real wait if the buf has completed at this point. The wait just decrements the completion counter.

If the IO has completed, then we run the completion code.

Quoted 2 lines

So what's the benefit of
"not waiting?" Where is the potential context switch?

async work for completion processing on synchrnous IO means we queue the work, then sleep in xfs_buf_iowait(). Two context switches, plus a work queue execution

Quoted 6 lines

Are you referring
to the case where error is set but I/O is not complete? Are you saying the advantage to the caller is it doesn't have to care about the state of further I/O once it has been determined at least one error has occurred? (If so, who cares about latency given that some operation that depends on this I/O is already doomed to fail?).

No, you're reading *way* too much into this. For sync IO, it's always best to process completion inline. For async, it doesn't matter, but if there's a submission error is *more effecient* to process it in the current context.

Quoted 11 lines

The code looks fine, but I'm trying to understand the reasoning better (and I suspect we can clarify the comment).

fs/xfs/xfs_buf.c xfs_buf_iorequest( · line 1431

Quoted 5 lines

fs/xfs/xfs_buf.c +4 -1

-	_xfs_buf_ioend(bp, bp->b_error ? 0 : 1);
+	if (bp->b_error || !(bp->b_flags & XBF_ASYNC))
+		_xfs_buf_ioend(bp, 0);
+	else
+		_xfs_buf_ioend(bp, 1);

Not related to this patch, but it seems like the problem this code tries to address is still possible.

The race condition is still possible - it just won't result in a use-after-free. The race condition is not fixed until patch 8, but as a backportable fix, this patch is much, much simpler.

Quoted 7 lines

Perhaps this papers over a particular
instance. Consider the case where an I/O fails immediately after this call completes, but not before. We have an extra reference now for completion, but we can still return to the caller with completion pending. I suppose its fine if we consider the "problem" to be that the reference goes away underneath the completion, as opposed to the caller caring about the status of completion.

Precisely.

Cheers,

Dave.

Signature · 7 lines

Dave Chinner
[email protected]

_______________________________________________
xfs mailing list
[email protected]
http://oss.sgi.com/mailman/listinfo/xfs

From: Dave Chinner <[email protected]>

We do some work in xfs_buf_ioend, and some work in xfs_buf_iodone_work, but much of that functionality is the same. This work can all be done in a single function, leaving xfs_buf_iodone just a wrapper to determine if we should execute it by workqueue or directly. hence rename xfs_buf_iodone_work to xfs_buf_ioend(), and add a new xfs_buf_ioend_async() for places that need async processing.

Signed-off-by Dave Chinner <[email protected]>

---
 fs/xfs/xfs_buf.c         | 79 +++++++++++++++++++++---------------------------
 fs/xfs/xfs_buf.h         |  2 +-
 fs/xfs/xfs_buf_item.c    |  4 +--
 fs/xfs/xfs_inode.c       |  2 +-
 fs/xfs/xfs_log.c         |  2 +-
 fs/xfs/xfs_log_recover.c |  2 +-
 6 files changed, 40 insertions(+), 51 deletions(-)

fs/xfs/xfs_buf.c +0 -0

index 5d86bbd..1b7f0bc 100644

fs/xfs/xfs_buf.c +3 -3

 {
-	struct xfs_buf		*bp =
-		container_of(work, xfs_buf_t, b_iodone_work);
-	bool			read = !!(bp->b_flags & XBF_READ);
+	bool		read = !!(bp->b_flags & XBF_READ);
+
+	trace_xfs_buf_iodone(bp, _RET_IP_);
 
 	bp->b_flags &= ~(XBF_READ | XBF_WRITE | XBF_READ_AHEAD);
 

fs/xfs/xfs_buf.c +16 -33

 	if (bp->b_iodone)
 		(*(bp->b_iodone))(bp);
 	else if (bp->b_flags & XBF_ASYNC)
 		xfs_buf_relse(bp);
 	else {
-		ASSERT(read && bp->b_ops);
 		complete(&bp->b_iowait);
 		xfs_buf_rele(bp);
 	}
 }
 
-void
-xfs_buf_ioend(
-	struct xfs_buf	*bp,
-	int		schedule)
+static void
+xfs_buf_ioend_work(
+	struct work_struct	*work)
 {
-	bool		read = !!(bp->b_flags & XBF_READ);
-
-	trace_xfs_buf_iodone(bp, _RET_IP_);
+	struct xfs_buf		*bp =
+		container_of(work, xfs_buf_t, b_iodone_work);
 
-	if (bp->b_error == 0)
-		bp->b_flags |= XBF_DONE;
+	xfs_buf_ioend(bp);
+}
 
-	if (bp->b_iodone || (read && bp->b_ops) || (bp->b_flags & XBF_ASYNC)) {
-		if (schedule) {
-			INIT_WORK(&bp->b_iodone_work, xfs_buf_iodone_work);
-			queue_work(xfslogd_workqueue, &bp->b_iodone_work);
-		} else {
-			xfs_buf_iodone_work(&bp->b_iodone_work);
-		}
-	} else {
-		bp->b_flags &= ~(XBF_READ | XBF_WRITE | XBF_READ_AHEAD);
-		complete(&bp->b_iowait);
-		xfs_buf_rele(bp);
-	}
+void
+xfs_buf_ioend_async(
+	struct xfs_buf	*bp)
+{
+	INIT_WORK(&bp->b_iodone_work, xfs_buf_ioend_work);
+	queue_work(xfslogd_workqueue, &bp->b_iodone_work);
 }
 
 void

@@ -1094,7 +1089,7 @@ xfs_bioerror(
 	XFS_BUF_UNDONE(bp);
 	xfs_buf_stale(bp);
 
-	xfs_buf_ioend(bp, 0);
+	xfs_buf_ioend(bp);
 
 	return -EIO;
 }

@@ -1181,15 +1176,6 @@ xfs_bwrite(
 }
 
 STATIC void
-_xfs_buf_ioend(
-	xfs_buf_t		*bp,
-	int			schedule)
-{
-	if (atomic_dec_and_test(&bp->b_io_remaining) == 1)
-		xfs_buf_ioend(bp, schedule);
-}
-
-STATIC void
 xfs_buf_bio_end_io(
 	struct bio		*bio,
 	int			error)

@@ -1206,7 +1192,8 @@ xfs_buf_bio_end_io(
 	if (!bp->b_error && xfs_buf_is_vmapped(bp) && (bp->b_flags & XBF_READ))
 		invalidate_kernel_vmap_range(bp->b_addr, xfs_buf_vmap_len(bp));
 
-	_xfs_buf_ioend(bp, 1);
+	if (atomic_dec_and_test(&bp->b_io_remaining) == 1)
+		xfs_buf_ioend_async(bp);
 	bio_put(bio);
 }
 

@@ -1425,10 +1412,12 @@ xfs_buf_iorequest(
 	 * waiting, and in the synchronous IO case it avoids unnecessary context
 	 * switches an latency for high-peformance devices.
 	 */

fs/xfs/xfs_buf.c +0 -0

 	xfs_buf_rele(bp);
 }

fs/xfs/xfs_buf.h +1 -1

index c753183..4585c15 100644
@@ -286,7 +286,7 @@ extern void xfs_buf_unlock(xfs_buf_t *);
 
 /* Buffer Read and Write Routines */
 extern int xfs_bwrite(struct xfs_buf *bp);
-extern void xfs_buf_ioend(xfs_buf_t *,	int);
+extern void xfs_buf_ioend(struct xfs_buf *bp);
 extern void xfs_buf_ioerror(xfs_buf_t *, int);
 extern void xfs_buf_ioerror_alert(struct xfs_buf *, const char *func);
 extern void xfs_buf_iorequest(xfs_buf_t *);

fs/xfs/xfs_buf_item.c +2 -2

index 76007de..4fd41b5 100644
@@ -491,7 +491,7 @@ xfs_buf_item_unpin(
 		xfs_buf_ioerror(bp, -EIO);
 		XFS_BUF_UNDONE(bp);
 		xfs_buf_stale(bp);
-		xfs_buf_ioend(bp, 0);
+		xfs_buf_ioend(bp);
 	}
 }
 

@@ -1115,7 +1115,7 @@ do_callbacks:
 	xfs_buf_do_callbacks(bp);
 	bp->b_fspriv = NULL;
 	bp->b_iodone = NULL;
-	xfs_buf_ioend(bp, 0);
+	xfs_buf_ioend(bp);
 }
 
 /*

fs/xfs/xfs_inode.c +1 -1

index fea3c92..00d210b 100644
@@ -3056,7 +3056,7 @@ cluster_corrupt_out:
 			XFS_BUF_UNDONE(bp);
 			xfs_buf_stale(bp);
 			xfs_buf_ioerror(bp, -EIO);
-			xfs_buf_ioend(bp, 0);
+			xfs_buf_ioend(bp);
 		} else {
 			xfs_buf_stale(bp);
 			xfs_buf_relse(bp);

fs/xfs/xfs_log.c +1 -1

index 8eaa8f5..e4665db 100644
@@ -1689,7 +1689,7 @@ xlog_bdstrat(
 	if (iclog->ic_state & XLOG_STATE_IOERROR) {
 		xfs_buf_ioerror(bp, -EIO);
 		xfs_buf_stale(bp);
-		xfs_buf_ioend(bp, 0);
+		xfs_buf_ioend(bp);
 		/*
 		 * It would seem logical to return EIO here, but we rely on
 		 * the log state machine to propagate I/O errors instead of

fs/xfs/xfs_log_recover.c +1 -1

index 1fd5787..4ba19bf 100644
@@ -383,7 +383,7 @@ xlog_recover_iodone(
 					SHUTDOWN_META_IO_ERROR);
 	}
 	bp->b_iodone = NULL;
-	xfs_buf_ioend(bp, 0);
+	xfs_buf_ioend(bp);
 }
 
 /*

Signature · 6 lines

2.0.0

_______________________________________________
xfs mailing list
[email protected]
http://oss.sgi.com/mailman/listinfo/xfs

Quoted 25 lines · Dave Chinner

From: Dave Chinner <[email protected]>

We do some work in xfs_buf_ioend, and some work in xfs_buf_iodone_work, but much of that functionality is the same. This work can all be done in a single function, leaving xfs_buf_iodone just a wrapper to determine if we should execute it by workqueue or directly. hence rename xfs_buf_iodone_work to xfs_buf_ioend(), and add a new xfs_buf_ioend_async() for places that need async processing.

Signed-off-by Dave Chinner <[email protected]>

---
 fs/xfs/xfs_buf.c         | 79 +++++++++++++++++++++---------------------------
 fs/xfs/xfs_buf.h         |  2 +-
 fs/xfs/xfs_buf_item.c    |  4 +--
 fs/xfs/xfs_inode.c       |  2 +-
 fs/xfs/xfs_log.c         |  2 +-
 fs/xfs/xfs_log_recover.c |  2 +-
 6 files changed, 40 insertions(+), 51 deletions(-)

fs/xfs/xfs_buf.c +0 -0

index 5d86bbd..1b7f0bc 100644

fs/xfs/xfs_buf.c xfs_buf_wait_unpin( · line 1003

Quoted 7 lines

fs/xfs/xfs_buf.c +2 -2

@@ -999,54 +999,49 @@ xfs_buf_wait_unpin(
  */
 
 STATIC void
-xfs_buf_iodone_work(
-	struct work_struct	*work)
+xfs_buf_ioend(
+	struct xfs_buf	*bp)

Compile failure here due to STATIC.

Quoted 10 lines

fs/xfs/xfs_buf.c +3 -3

 {
-	struct xfs_buf		*bp =
-		container_of(work, xfs_buf_t, b_iodone_work);
-	bool			read = !!(bp->b_flags & XBF_READ);
+	bool		read = !!(bp->b_flags & XBF_READ);
+
+	trace_xfs_buf_iodone(bp, _RET_IP_);
 
 	bp->b_flags &= ~(XBF_READ | XBF_WRITE | XBF_READ_AHEAD);
 

fs/xfs/xfs_buf.c xfs_buf_wait_unpin( · line 1017

Quoted 10 lines

fs/xfs/xfs_buf.c +7 -3

-	/* only validate buffers that were read without errors */
-	if (read && bp->b_ops && !bp->b_error && (bp->b_flags & XBF_DONE))
-		bp->b_ops->verify_read(bp);
+	if (!bp->b_error) {
+		bp->b_flags |= XBF_DONE;
+
+		/* only validate buffers that were read without errors */
+		if (read && bp->b_ops)
+			bp->b_ops->verify_read(bp);
+	}

Probably not a cause of errors, but this code is now executed twice for I/O with b_iodone callbacks.

Once for the initial call from bio_end_io,
again from the callback via the b_iodone handler. The flags bits are probably fine, but we don't want to be running the verifiers multiple times unnecessarily.

Quoted 8 lines

fs/xfs/xfs_buf.c +0 -4

@@ -1425,10 +1412,12 @@ xfs_buf_iorequest(
 	 * waiting, and in the synchronous IO case it avoids unnecessary context
 	 * switches an latency for high-peformance devices.
 	 */
-	if (bp->b_error || !(bp->b_flags & XBF_ASYNC))
-		_xfs_buf_ioend(bp, 0);
-	else
-		_xfs_buf_ioend(bp, 1);

fs/xfs/xfs_buf.c xfs_buf_iorequest( · line 1420

Quoted 6 lines

fs/xfs/xfs_buf.c +6 -0

+	if (atomic_dec_and_test(&bp->b_io_remaining) == 1) {
+		if (bp->b_error || !(bp->b_flags & XBF_ASYNC))
+			xfs_buf_ioend(bp);
+		else
+			xfs_buf_ioend_async(bp);
+	}

This looks cleaner, but the comment is out of whack at this point.

Quoted 46 lines · Brian Foster

Quoted 25 lines · Dave Chinner

From: Dave Chinner <[email protected]>

We do some work in xfs_buf_ioend, and some work in xfs_buf_iodone_work, but much of that functionality is the same. This work can all be done in a single function, leaving xfs_buf_iodone just a wrapper to determine if we should execute it by workqueue or directly. hence rename xfs_buf_iodone_work to xfs_buf_ioend(), and add a new xfs_buf_ioend_async() for places that need async processing.

Signed-off-by Dave Chinner <[email protected]>

---
 fs/xfs/xfs_buf.c         | 79 +++++++++++++++++++++---------------------------
 fs/xfs/xfs_buf.h         |  2 +-
 fs/xfs/xfs_buf_item.c    |  4 +--
 fs/xfs/xfs_inode.c       |  2 +-
 fs/xfs/xfs_log.c         |  2 +-
 fs/xfs/xfs_log_recover.c |  2 +-
 6 files changed, 40 insertions(+), 51 deletions(-)

fs/xfs/xfs_buf.c +0 -0

index 5d86bbd..1b7f0bc 100644

fs/xfs/xfs_buf.c xfs_buf_wait_unpin( · line 1003

Quoted 7 lines

fs/xfs/xfs_buf.c +2 -2

@@ -999,54 +999,49 @@ xfs_buf_wait_unpin(
  */
 
 STATIC void
-xfs_buf_iodone_work(
-	struct work_struct	*work)
+xfs_buf_ioend(
+	struct xfs_buf	*bp)

Compile failure here due to STATIC.

Quoted 10 lines

fs/xfs/xfs_buf.c +3 -3

 {
-	struct xfs_buf		*bp =
-		container_of(work, xfs_buf_t, b_iodone_work);
-	bool			read = !!(bp->b_flags & XBF_READ);
+	bool		read = !!(bp->b_flags & XBF_READ);
+
+	trace_xfs_buf_iodone(bp, _RET_IP_);
 
 	bp->b_flags &= ~(XBF_READ | XBF_WRITE | XBF_READ_AHEAD);
 

Quoted 13 lines

fs/xfs/xfs_buf.c xfs_buf_wait_unpin( · line 1017

Quoted 10 lines

fs/xfs/xfs_buf.c +7 -3

-	/* only validate buffers that were read without errors */
-	if (read && bp->b_ops && !bp->b_error && (bp->b_flags & XBF_DONE))
-		bp->b_ops->verify_read(bp);
+	if (!bp->b_error) {
+		bp->b_flags |= XBF_DONE;
+
+		/* only validate buffers that were read without errors */
+		if (read && bp->b_ops)
+			bp->b_ops->verify_read(bp);
+	}

Probably not a cause of errors, but this code is now executed twice for I/O with b_iodone callbacks.

reads don't have b_iodone callbacks.

Quoted 4 lines

Once for the initial call from bio_end_io,
again from the callback via the b_iodone handler. The flags bits are probably fine, but we don't want to be running the verifiers multiple times unnecessarily.

Which we don't ;)

Quoted 8 lines

Quoted 8 lines

fs/xfs/xfs_buf.c +0 -4

@@ -1425,10 +1412,12 @@ xfs_buf_iorequest(
 	 * waiting, and in the synchronous IO case it avoids unnecessary context
 	 * switches an latency for high-peformance devices.
 	 */
-	if (bp->b_error || !(bp->b_flags & XBF_ASYNC))
-		_xfs_buf_ioend(bp, 0);
-	else
-		_xfs_buf_ioend(bp, 1);

Quoted 8 lines

fs/xfs/xfs_buf.c xfs_buf_iorequest( · line 1420

Quoted 6 lines

fs/xfs/xfs_buf.c +6 -0

+	if (atomic_dec_and_test(&bp->b_io_remaining) == 1) {
+		if (bp->b_error || !(bp->b_flags & XBF_ASYNC))
+			xfs_buf_ioend(bp);
+		else
+			xfs_buf_ioend_async(bp);
+	}

This looks cleaner, but the comment is out of whack at this point.

The code is functionally identical, so the comment didn't get changed. As it is, the behaviour that exists in this patch goes away in later patches, so it's mostly irrelevant that a comment is absoultely correct in an intermediate point within the patch set.

Cheers,

Dave.

Signature · 7 lines

Dave Chinner
[email protected]

_______________________________________________
xfs mailing list
[email protected]
http://oss.sgi.com/mailman/listinfo/xfs

STATIC void

Quoted 1 line

STATIC void

needs the static removed as pointed out by Brian.

fs/xfs/xfs_buf.c xfs_buf_wait_unpin( · line 1005

Quoted 2 lines

fs/xfs/xfs_buf.c +1 -1

-	bool			read = !!(bp->b_flags & XBF_READ);
+	bool		read = !!(bp->b_flags & XBF_READ);

We don't really need the double negation here.

From: Dave Chinner <[email protected]>

Currently the report of a bio error from completion immediately marks the buffer with an error. The issue is that this is racy w.r.t. synchronous IO - the submitter can see b_error being set before the IO is complete, and hence we cannot differentiate between submission failures and completion failures.

Add an internal b_io_error field protected by the b_lock to catch IO completion errors, and only propagate that to the buffer during final IO completion handling. Hence we can tell in xfs_buf_iorequest if we've had a submission failure bey checking bp->b_error before dropping our b_io_remaining reference - that reference will prevent b_io_error values from being propagated to b_error in the event that completion races with submission.

In doing so, allow xfs_buf_iorequest to return an error. That way, the caller can check for submission errors safely if required, and easily distinguish them from completion errors that come from xfs_buf_iowait().

Signed-off-by Dave Chinner <[email protected]>

---
 fs/xfs/xfs_buf.c | 18 ++++++++++++++++--
 fs/xfs/xfs_buf.h |  1 +
 2 files changed, 17 insertions(+), 2 deletions(-)

fs/xfs/xfs_buf.c +16 -2

index 1b7f0bc..58ae34c 100644
@@ -1008,6 +1008,13 @@ xfs_buf_ioend(
 
 	bp->b_flags &= ~(XBF_READ | XBF_WRITE | XBF_READ_AHEAD);
 
+	/*
+	 * Pull in IO completion errors now. We are guaranteed to be running
+	 * single threaded, so we don't need the lock to read b_io_error.
+	 */
+	if (!bp->b_error && bp->b_io_error)
+		xfs_buf_ioerror(bp, bp->b_io_error);
+
 	if (!bp->b_error) {
 		bp->b_flags |= XBF_DONE;
 

@@ -1186,8 +1193,12 @@ xfs_buf_bio_end_io(
 	 * don't overwrite existing errors - otherwise we can lose errors on
 	 * buffers that require multiple bios to complete.
 	 */
-	if (!bp->b_error)
-		xfs_buf_ioerror(bp, error);
+	if (error) {
+		spin_lock(&bp->b_lock);
+		if (!bp->b_io_error)
+			bp->b_io_error = error;
+		spin_unlock(&bp->b_lock);
+	}
 
 	if (!bp->b_error && xfs_buf_is_vmapped(bp) && (bp->b_flags & XBF_READ))
 		invalidate_kernel_vmap_range(bp->b_addr, xfs_buf_vmap_len(bp));

@@ -1373,6 +1384,9 @@ xfs_buf_iorequest(
 	if (bp->b_flags & XBF_WRITE)
 		xfs_buf_wait_unpin(bp);
 
+	/* clear the internal error state to avoid spurious errors */
+	bp->b_io_error = 0;
+
 	/*
 	 * Take references to the buffer. For XBF_ASYNC buffers, holding a
 	 * reference for as long as submission takes is all that is necessary

fs/xfs/xfs_buf.h +1 -0

index 4585c15..44db8cd 100644
@@ -158,6 +158,7 @@ typedef struct xfs_buf {
 	struct list_head	b_lru;		/* lru list */
 	spinlock_t		b_lock;		/* internal state lock */
 	unsigned int		b_state;	/* internal state flags */
+	int			b_io_error;	/* internal IO error state */
 	wait_queue_head_t	b_waiters;	/* unpin waiters */
 	struct list_head	b_list;
 	struct xfs_perag	*b_pag;		/* contains rbtree root */

Signature · 6 lines

2.0.0

_______________________________________________
xfs mailing list
[email protected]
http://oss.sgi.com/mailman/listinfo/xfs

Quoted 16 lines · Dave Chinner

From: Dave Chinner <[email protected]>

Currently the report of a bio error from completion immediately marks the buffer with an error. The issue is that this is racy w.r.t. synchronous IO - the submitter can see b_error being set before the IO is complete, and hence we cannot differentiate between submission failures and completion failures.

Add an internal b_io_error field protected by the b_lock to catch IO completion errors, and only propagate that to the buffer during final IO completion handling. Hence we can tell in xfs_buf_iorequest if we've had a submission failure bey checking bp->b_error before dropping our b_io_remaining reference - that reference will prevent b_io_error values from being propagated to b_error in the event that completion races with submission.

Quoted 5 lines

In doing so, allow xfs_buf_iorequest to return an error. That way, the caller can check for submission errors safely if required, and easily distinguish them from completion errors that come from xfs_buf_iowait().

I don't see any changes to xfs_buf_iorequest() at all in this one. That aside, it looks fine. It's not how I was thinking about it, but a clear separation of submission error regardless.

In doing so, allow xfs_buf_iorequest to return an error. That way, the caller can check for submission errors safely if required, and easily distinguish them from completion errors that come from xfs_buf_iowait().

From: Dave Chinner <[email protected]>

Only has two callers, and is just a shutdown check and error handler around xfs_buf_iorequest. However, the error handling is a mess of read and write semantics, and both internal callers only call it for writes. Hence kill the wrapper, and follow up with a patch to sanitise the error handling.

Signed-off-by Dave Chinner <[email protected]>

From: Dave Chinner <[email protected]>

Internal buffer write error handling is a mess due to the unnatural split between xfs_bioerror and xfs_bioerror_relse(). The buffer reference counting is also wrong for the xfs_bioerror path for xfs_bwrite - it uses sync IO and so xfs_buf_ioend will release a hold count that was never taken.

Further, xfs_bwrite only does sync IO and determines the handler to call based on b_iodone, so for this caller the only difference between xfs_bioerror() and xfs_bioerror_release() is the XBF_DONE flag. We don't care what the XBF_DONE flag state is because we stale the buffer in both paths - the next buffer lookup will clear XBF_DONE anyway because XBF_STALE is set. Hence we can use common error handling for xfs_bwrite().

__xfs_buf_delwri_submit() is a similar - it's only ever called on writes - all sync or async - and again there's no reason to handle them any differently at all.

Clean up the nasty error handling and remove xfs_bioerror().

Signed-off-by Dave Chinner <[email protected]>

---
 fs/xfs/xfs_buf.c | 67 +++++++++++++++++---------------------------------------
 1 file changed, 20 insertions(+), 47 deletions(-)

fs/xfs/xfs_buf.c +2 -34

index 98fcf5a..f444285 100644
@@ -1072,36 +1072,6 @@ xfs_buf_ioerror_alert(
 }
 
 /*
- * Called when we want to stop a buffer from getting written or read.
- * We attach the EIO error, muck with its flags, and call xfs_buf_ioend
- * so that the proper iodone callbacks get called.
- */
-STATIC int
-xfs_bioerror(
-	xfs_buf_t *bp)
-{
-#ifdef XFSERRORDEBUG
-	ASSERT(XFS_BUF_ISREAD(bp) || bp->b_iodone);
-#endif
-
-	/*
-	 * No need to wait until the buffer is unpinned, we aren't flushing it.
-	 */
-	xfs_buf_ioerror(bp, -EIO);
-
-	/*
-	 * We're calling xfs_buf_ioend, so delete XBF_DONE flag.
-	 */
-	XFS_BUF_UNREAD(bp);
-	XFS_BUF_UNDONE(bp);
-	xfs_buf_stale(bp);
-
-	xfs_buf_ioend(bp);
-
-	return -EIO;
-}
-
-/*
  * Same as xfs_bioerror, except that we are releasing the buffer
  * here ourselves, and avoiding the xfs_buf_ioend call.
  * This is meant for userdata errors; metadata bufs come with

@@ -1149,19 +1119,19 @@ xfs_bwrite(
 	ASSERT(xfs_buf_islocked(bp));
 
 	bp->b_flags |= XBF_WRITE;
-	bp->b_flags &= ~(XBF_ASYNC | XBF_READ | _XBF_DELWRI_Q | XBF_WRITE_FAIL);
+	bp->b_flags &= ~(XBF_ASYNC | XBF_READ | _XBF_DELWRI_Q |
+			 XBF_WRITE_FAIL | XBF_DONE);
 
 	if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
 		trace_xfs_bdstrat_shut(bp, _RET_IP_);
 
-		/*
-		 * Metadata write that didn't get logged but written anyway.
-		 * These aren't associated with a transaction, and can be

fs/xfs/xfs_buf.c +13 -8

+		xfs_buf_ioend(bp);
+		return -EIO;
 	}
 
 	xfs_buf_iorequest(bp);

@@ -1848,16 +1818,19 @@ __xfs_buf_delwri_submit(
 		if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
 			trace_xfs_bdstrat_shut(bp, _RET_IP_);
 
+			xfs_buf_ioerror(bp, -EIO);
+			xfs_buf_stale(bp);
+
 			/*
-			 * Metadata write that didn't get logged but written anyway.
-			 * These aren't associated with a transaction, and can be
-			 * ignored.
+			 * if sync IO, xfs_buf_ioend is going to remove a
+			 * ref here. We need to add that so we can collect
+			 * all the buffer errors in the wait loop.
 			 */
-			if (!bp->b_iodone)
-				return xfs_bioerror_relse(bp);
-			return xfs_bioerror(bp);
-		}
-		xfs_buf_iorequest(bp);
+			if (wait)
+				xfs_buf_hold(bp);
+			xfs_buf_ioend(bp);
+		} else
+			xfs_buf_iorequest(bp);
 	}
 	blk_finish_plug(&plug);
 

Signature · 6 lines

2.0.0

_______________________________________________
xfs mailing list
[email protected]
http://oss.sgi.com/mailman/listinfo/xfs

Quoted 21 lines · Dave Chinner

fs/xfs/xfs_buf.c +4 -9

@@ -1149,19 +1119,19 @@ xfs_bwrite(
 	ASSERT(xfs_buf_islocked(bp));
 
 	bp->b_flags |= XBF_WRITE;
-	bp->b_flags &= ~(XBF_ASYNC | XBF_READ | _XBF_DELWRI_Q | XBF_WRITE_FAIL);
+	bp->b_flags &= ~(XBF_ASYNC | XBF_READ | _XBF_DELWRI_Q |
+			 XBF_WRITE_FAIL | XBF_DONE);
 
 	if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
 		trace_xfs_bdstrat_shut(bp, _RET_IP_);
 
-		/*
-		 * Metadata write that didn't get logged but written anyway.
-		 * These aren't associated with a transaction, and can be
-		 * ignored.
-		 */
-		if (!bp->b_iodone)
-			return xfs_bioerror_relse(bp);
-		return xfs_bioerror(bp);
+		xfs_buf_ioerror(bp, -EIO);
+		xfs_buf_stale(bp);

fs/xfs/xfs_buf.c xfs_bwrite( · line 1132

Quoted 3 lines

fs/xfs/xfs_buf.c +3 -0

+
+		/* sync IO, xfs_buf_ioend is going to remove a ref here */
+		xfs_buf_hold(bp);

Looks correct, but that's kind of ugly. The reference serves no purpose except to appease the error sequence. It gives the impression that the reference management should be handled at a higher level (and with truly synchronous I/O primitives/mechanisms, it is ;).

At the very least, can we reorganize the ioend side of things to handle this? We already have the duplicate execution issue previously mentioned that has to get resolved. Some duplicate code is fine if it improves clarity, imo.

Internal buffer write error handling is a mess due to the unnatural split between xfs_bioerror and xfs_bioerror_relse(). The buffer reference counting is also wrong for the xfs_bioerror path for xfs_bwrite - it uses sync IO and so xfs_buf_ioend will release a hold count that was never taken.

Quoted 5 lines · Dave Chinner

Internal buffer write error handling is a mess due to the unnatural split between xfs_bioerror and xfs_bioerror_relse(). The buffer reference counting is also wrong for the xfs_bioerror path for xfs_bwrite - it uses sync IO and so xfs_buf_ioend will release a hold count that was never taken.

Which seems to be cause by patch 1 in the series, and thus the additional hold for the error case should be introduced there.

index 96c898e..758c07d 100644

fs/xfs/xfs_trans_buf.c +0 -0

@@ -324,11 +324,13 @@ xfs_trans_read_buf_map(
 			 */

Quoted 12 lines

index 96c898e..758c07d 100644

fs/xfs/xfs_trans_buf.c +2 -3

@@ -324,11 +324,13 @@ xfs_trans_read_buf_map(
 			 */
 			if (XFS_FORCED_SHUTDOWN(mp)) {
 				trace_xfs_bdstrat_shut(bp, _RET_IP_);
-				xfs_bioerror_relse(bp);
-			} else {
-				xfs_buf_iorequest(bp);
+				bp->b_flags &= ~(XBF_READ | XBF_DONE);
+				xfs_buf_ioerror(bp, -EIO);

fs/xfs/xfs_trans_buf.c xfs_trans_read_buf_map( · line 331

Quoted 3 lines

fs/xfs/xfs_trans_buf.c +2 -0

+				xfs_buf_stale(bp);
+				return -EIO;
 			}

This is a large change of behavior as it doesn't hit the error path after the xfs_buf_iowait anymore. While I don't think that that path was entirely correct this version seems to be even less so by not releasing the buffer reference or forcing the shutdown.

Quoted 4 lines

This is a large change of behavior as it doesn't hit the error path after the xfs_buf_iowait anymore. While I don't think that that path was entirely correct this version seems to be even less so by not releasing the buffer reference or forcing the shutdown.

The IO is synchronous, so the previous behaviour did not release the buffer here. But, yes, it needs to because we're not running the io wait on it anymore. And this happens only during a shutdown, so i don't see any need to trigger a shutdown ;)

As it is, I think this gets properly fixed by the next patch....

Quoted 10 lines · Dave Chinner

Quoted 4 lines

This is a large change of behavior as it doesn't hit the error path after the xfs_buf_iowait anymore. While I don't think that that path was entirely correct this version seems to be even less so by not releasing the buffer reference or forcing the shutdown.

The IO is synchronous, so the previous behaviour did not release the buffer here. But, yes, it needs to because we're not running the io wait on it anymore. And this happens only during a shutdown, so i don't see any need to trigger a shutdown ;)

Quoted 1 line

As it is, I think this gets properly fixed by the next patch....

Do you have any scenario that might actually exercise this path? I enabled xfs_trans_read_buf_io trace events and run xfstests as well as a few other workloads and couldn't hit it at all. And when you think of it: when would be do a trans_get_buf, then not actually updating it with data and then recurse into a trans_read_buf in the same transaction?

Maybe it's just time do a bit more of an audit and put this whole code path to rest.

Quoted 4 lines

fs/xfs/xfs_bmap_util.c +0 -0

index 2f1e30d..c53cc03 100644

fs/xfs/xfs_bmap_util.c xfs_zero_remaining_bytes(

Quoted 1 line

fs/xfs/xfs_bmap_util.c +0 -0

@@ -1157,12 +1157,7 @@ xfs_zero_remaining_bytes(

xfs_zero_remaining_bytes really should be switched to xfs_buf_read_uncached + xfs_bwrite instead of all this mess just to micro-optimize a few memory allocation away that don't even happen for the common case of blocksize == PAGE_SIZE. I'm not even sure we should be using the buffer cache at all for something inside a regular file, but that's a discussion for another time.

fs/xfs/xfs_bmap_util.c xfs_zero_remaining_bytes( · line 1161

Quoted 6 lines

fs/xfs/xfs_bmap_util.c +2 -0

 void
+xfs_buf_submit(
+	struct xfs_buf	*bp)
 {
 	trace_xfs_buf_iorequest(bp, _RET_IP_);

I suspect these two should have properly name and separate trace points now?

It also seems like a lot of the guts of the two functions are still the same, so factoring that into a __xfs_buf_submit helper would be useful.

Quoted 4 lines

fs/xfs/xfs_bmap_util.c +2 -0

+	 * If _xfs_buf_ioapply failed, 
+	 * we can get back here with only the IO
 	 * reference we took above.  _xfs_buf_ioend will drop it to zero, so
 	 * we'd better run completion processing synchronously so that the we

fs/xfs/xfs_bmap_util.c

Quoted 6 lines

fs/xfs/xfs_bmap_util.c +3 -0

+	 * don't return to the caller with completion still pending.
+	 * this allows the caller to check b_error safely without
+	 * waiting
 	 */
 	if (atomic_dec_and_test(&bp->b_io_remaining) == 1) {
 		if (bp->b_error || !(bp->b_flags & XBF_ASYNC))

I don't think the !ASYNC case can happen here, can it?

fs/xfs/xfs_bmap_util.c

Quoted 5 lines

fs/xfs/xfs_bmap_util.c +3 -0

+		if (!wait)
 			list_del_init(&bp->b_list);
+		else
+			xfs_buf_hold(bp);

Maybe switch this around to avoid the negated condition in the if else?

Also might this be worth a change of it's own?

xfs_zero_remaining_bytes is uses an uncached buffer, so we're not using the buffer cache at all for the blocks being zeroed. That is why it does the flag twiddling dance it does. However, consolidation all the different block zeroing functions we have is an exercise for a different day....

From: Dave Chinner <[email protected]>

There is a lot of cookie-cutter code that looks like:

if (shutdown)
handle buffer error
xfs_buf_iorequest(bp)
error = xfs_buf_iowait(bp)
if (error)
handle buffer error

spread through XFS. There's significant complexity now in xfs_buf_iorequest() to specifically handle this sort of synchronous IO pattern, but there's all sorts of nasty surprises in different error handling code dependent on who owns the buffer references and the locks.

Pull this pattern into a single helper, where we can hide all the synchronous IO warts and hence make the error handling for all the callers much saner. This removes the need for a special extra reference to protect IO completion processing, as we can now hold a single reference across dispatch and waiting, simplifying the sync IO smeantics and error handling.

In doing this, also rename xfs_buf_iorequest to xfs_buf_submit and make it explicitly handle on asynchronous IO. This forces all users to be switched specifically to one interface or the other and removes any ambiguity between how the interfaces are to be used. It also means that xfs_buf_iowait() goes away.

For the special case of delwri buffer submission and waiting, we don't need to issue IO synchronously at all. The second pass to cal xfs_buf_iowait() can now just block on xfs_buf_lock() - the buffer will be unlocked when the async IO is complete. This formalises a sane the method of waiting for async IO - take an extra reference, submit the IO, call xfs_buf_lock() when you want to wait for IO completion. i.e.:

	bp = xfs_buf_find();
	xfs_buf_hold(bp);
	bp->b_flags |= XBF_ASYNC;
	xfs_buf_iosubmit(bp);
	xfs_buf_lock(bp)
	error = bp->b_error;
	....
	xfs_buf_relse(bp);

Signed-off-by Dave Chinner <[email protected]>

---

 fs/xfs/xfs_bmap_util.c   |  14 +---
 fs/xfs/xfs_buf.c         | 186 ++++++++++++++++++++++++++---------------------
 fs/xfs/xfs_buf.h         |   4 +-
 fs/xfs/xfs_buf_item.c    |   4 +-
 fs/xfs/xfs_log.c         |   2 +-
 fs/xfs/xfs_log_recover.c |  22 ++----
 fs/xfs/xfs_trans_buf.c   |  17 ++---
 7 files changed, 122 insertions(+), 127 deletions(-)

fs/xfs/xfs_bmap_util.c +2 -12

index 2f1e30d..c53cc03 100644
@@ -1157,12 +1157,7 @@ xfs_zero_remaining_bytes(
 		XFS_BUF_READ(bp);
 		XFS_BUF_SET_ADDR(bp, xfs_fsb_to_db(ip, imap.br_startblock));
 
-		if (XFS_FORCED_SHUTDOWN(mp)) {
-			error = -EIO;
-			break;
-		}
-		xfs_buf_iorequest(bp);
-		error = xfs_buf_iowait(bp);
+		error = xfs_buf_submit_wait(bp);
 		if (error) {
 			xfs_buf_ioerror_alert(bp,
 					"xfs_zero_remaining_bytes(read)");

@@ -1175,12 +1170,7 @@ xfs_zero_remaining_bytes(
 		XFS_BUF_UNREAD(bp);
 		XFS_BUF_WRITE(bp);
 
-		if (XFS_FORCED_SHUTDOWN(mp)) {
-			error = -EIO;
-			break;
-		}
-		xfs_buf_iorequest(bp);
-		error = xfs_buf_iowait(bp);
+		error = xfs_buf_submit_wait(bp);
 		if (error) {
 			xfs_buf_ioerror_alert(bp,
 					"xfs_zero_remaining_bytes(write)");

fs/xfs/xfs_buf.c +102 -84

index 352e9219..a2599f9 100644
@@ -623,10 +623,11 @@ _xfs_buf_read(
 	bp->b_flags &= ~(XBF_WRITE | XBF_ASYNC | XBF_READ_AHEAD);
 	bp->b_flags |= flags & (XBF_READ | XBF_ASYNC | XBF_READ_AHEAD);
 
-	xfs_buf_iorequest(bp);
-	if (flags & XBF_ASYNC)
+	if (flags & XBF_ASYNC) {
+		xfs_buf_submit(bp);
 		return 0;
-	return xfs_buf_iowait(bp);
+	}
+	return xfs_buf_submit_wait(bp);
 }
 
 xfs_buf_t *

@@ -708,12 +709,7 @@ xfs_buf_read_uncached(
 	bp->b_flags |= XBF_READ;
 	bp->b_ops = ops;
 
-	if (XFS_FORCED_SHUTDOWN(target->bt_mount)) {
-		xfs_buf_relse(bp);
-		return NULL;
-	}
-	xfs_buf_iorequest(bp);
-	xfs_buf_iowait(bp);
+	xfs_buf_submit_wait(bp);
 	return bp;
 }
 

@@ -1027,10 +1023,8 @@ xfs_buf_ioend(
 		(*(bp->b_iodone))(bp);
 	else if (bp->b_flags & XBF_ASYNC)
 		xfs_buf_relse(bp);
-	else {
+	else
 		complete(&bp->b_iowait);
-		xfs_buf_rele(bp);
-	}
 }
 
 static void

@@ -1083,21 +1077,7 @@ xfs_bwrite(
 	bp->b_flags &= ~(XBF_ASYNC | XBF_READ | _XBF_DELWRI_Q |
 			 XBF_WRITE_FAIL | XBF_DONE);
 
-	if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
-		trace_xfs_bdstrat_shut(bp, _RET_IP_);
-
-		xfs_buf_ioerror(bp, -EIO);
-		xfs_buf_stale(bp);
-
-		/* sync IO, xfs_buf_ioend is going to remove a ref here */
-		xfs_buf_hold(bp);
-		xfs_buf_ioend(bp);
-		return -EIO;
-	}
-
-	xfs_buf_iorequest(bp);
-
-	error = xfs_buf_iowait(bp);
+	error = xfs_buf_submit_wait(bp);
 	if (error) {
 		xfs_force_shutdown(bp->b_target->bt_mount,
 				   SHUTDOWN_META_IO_ERROR);

@@ -1206,7 +1186,7 @@ next_chunk:
 	} else {
 		/*
 		 * This is guaranteed not to be the last io reference count
-		 * because the caller (xfs_buf_iorequest) holds a count itself.
+		 * because the caller (xfs_buf_submit) holds a count itself.
 		 */
 		atomic_dec(&bp->b_io_remaining);
 		xfs_buf_ioerror(bp, -EIO);

@@ -1296,13 +1276,29 @@ _xfs_buf_ioapply(
 	blk_finish_plug(&plug);
 }
 
+/*
+ * Asynchronous IO submission path. This transfers the buffer lock ownership and
+ * the current reference to the IO. It is not safe to reference the buffer after
+ * a call to this function unless the caller holds an additional reference
+ * itself.
+ */
 void
-xfs_buf_iorequest(
-	xfs_buf_t		*bp)
+xfs_buf_submit(
+	struct xfs_buf	*bp)
 {
 	trace_xfs_buf_iorequest(bp, _RET_IP_);
 
 	ASSERT(!(bp->b_flags & _XBF_DELWRI_Q));
+	ASSERT(bp->b_flags & XBF_ASYNC);
+
+	/* on shutdown we stale and complete the buffer immediately */
+	if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
+		xfs_buf_ioerror(bp, -EIO);
+		bp->b_flags &= ~XBF_DONE;
+		xfs_buf_stale(bp);
+		xfs_buf_ioend(bp);
+		return;
+	}
 
 	if (bp->b_flags & XBF_WRITE)
 		xfs_buf_wait_unpin(bp);

@@ -1311,25 +1307,10 @@ xfs_buf_iorequest(
 	bp->b_io_error = 0;
 
 	/*
-	 * Take references to the buffer. For XBF_ASYNC buffers, holding a
-	 * reference for as long as submission takes is all that is necessary
-	 * here. The IO inherits the lock and hold count from the submitter,
-	 * and these are release during IO completion processing. Taking a hold
-	 * over submission ensures that the buffer is not freed until we have
-	 * completed all processing, regardless of when IO errors occur or are
-	 * reported.
-	 *
-	 * However, for synchronous IO, the IO does not inherit the submitters
-	 * reference count, nor the buffer lock. Hence we need to take an extra
-	 * reference to the buffer for the for the IO context so that we can
-	 * guarantee the buffer is not freed until all IO completion processing
-	 * is done. Otherwise the caller can drop their reference while the IO
-	 * is still in progress and hence trigger a use-after-free situation.
+	 * Take an extra reference so that we don't have to guess when it's no
+	 * longer safe to reference the buffer that was passed to us.
 	 */
 	xfs_buf_hold(bp);
-	if (!(bp->b_flags & XBF_ASYNC))
-		xfs_buf_hold(bp);
-
 
 	/*
 	 * Set the count to 1 initially, this will stop an I/O completion

@@ -1340,14 +1321,13 @@ xfs_buf_iorequest(
 	_xfs_buf_ioapply(bp);
 
 	/*
-	 * If _xfs_buf_ioapply failed or we are doing synchronous IO that
-	 * completes extremely quickly, we can get back here with only the IO
+	 * If _xfs_buf_ioapply failed, 
+	 * we can get back here with only the IO
 	 * reference we took above.  _xfs_buf_ioend will drop it to zero, so
 	 * we'd better run completion processing synchronously so that the we
-	 * don't return to the caller with completion still pending. In the
-	 * error case, this allows the caller to check b_error safely without
-	 * waiting, and in the synchronous IO case it avoids unnecessary context
-	 * switches an latency for high-peformance devices.
+	 * don't return to the caller with completion still pending.
+	 * this allows the caller to check b_error safely without
+	 * waiting
 	 */
 	if (atomic_dec_and_test(&bp->b_io_remaining) == 1) {
 		if (bp->b_error || !(bp->b_flags & XBF_ASYNC))

@@ -1357,25 +1337,70 @@ xfs_buf_iorequest(
 	}
 
 	xfs_buf_rele(bp);
+	/* Note: it is not safe to reference bp now we've dropped our ref */
 }
 
 /*
- * Waits for I/O to complete on the buffer supplied.  It returns immediately if
- * no I/O is pending or there is already a pending error on the buffer, in which
- * case nothing will ever complete.  It returns the I/O error code, if any, or
- * 0 if there was no error.
+ * Synchronous buffer IO submission path, read or write.
  */
 int
-xfs_buf_iowait(
-	xfs_buf_t		*bp)
+xfs_buf_submit_wait(
+	struct xfs_buf	*bp)
 {
-	trace_xfs_buf_iowait(bp, _RET_IP_);
+	int		error;
+
+	trace_xfs_buf_iorequest(bp, _RET_IP_);
 
-	if (!bp->b_error)
-		wait_for_completion(&bp->b_iowait);
+	ASSERT(!(bp->b_flags & (_XBF_DELWRI_Q | XBF_ASYNC)));
+
+	if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
+		xfs_buf_ioerror(bp, -EIO);
+		xfs_buf_stale(bp);
+		bp->b_flags &= ~XBF_DONE;
+		return -EIO;
+	}
 
+	if (bp->b_flags & XBF_WRITE)
+		xfs_buf_wait_unpin(bp);
+
+	/* clear the internal error state to avoid spurious errors */
+	bp->b_io_error = 0;
+
+	/*
+	 * For synchronous IO, the IO does not inherit the submitters reference
+	 * count, nor the buffer lock. Hence we cannot release the reference we
+	 * are about to take until we've waited for all IO completion to occur,
+	 * including any xfs_buf_ioend_async() work that may be pending.
+	 */
+	xfs_buf_hold(bp);
+
+	/*
+	 * Set the count to 1 initially, this will stop an I/O completion
+	 * callout which happens before we have started all the I/O from calling
+	 * xfs_buf_ioend too early.
+	 */
+	atomic_set(&bp->b_io_remaining, 1);
+	_xfs_buf_ioapply(bp);
+
+	/*
+	 * make sure we run completion synchronously if it raced with us and is
+	 * already complete.
+	 */
+	if (atomic_dec_and_test(&bp->b_io_remaining) == 1)
+		xfs_buf_ioend(bp);
+
+	/* wait for completion before gathering the error from the buffer */
+	trace_xfs_buf_iowait(bp, _RET_IP_);
+	wait_for_completion(&bp->b_iowait);
 	trace_xfs_buf_iowait_done(bp, _RET_IP_);
-	return bp->b_error;
+	error = bp->b_error;
+
+	/*
+	 * all done now, we can release the hold that keeps the buffer
+	 * referenced for the entire IO.
+	 */
+	xfs_buf_rele(bp);
+	return error;
 }
 
 xfs_caddr_t

@@ -1769,29 +1794,19 @@ __xfs_buf_delwri_submit(
 	blk_start_plug(&plug);
 	list_for_each_entry_safe(bp, n, io_list, b_list) {
 		bp->b_flags &= ~(_XBF_DELWRI_Q | XBF_ASYNC | XBF_WRITE_FAIL);
-		bp->b_flags |= XBF_WRITE;
+		bp->b_flags |= XBF_WRITE | XBF_ASYNC;
 
-		if (!wait) {
-			bp->b_flags |= XBF_ASYNC;
+		/*
+		 * we do all Io submission async. This means if we need to wait
+		 * for IO completion we need to take an extra reference so the
+		 * buffer is still valid on the other side.
+		 */
+		if (!wait)
 			list_del_init(&bp->b_list);
-		}
-
-		if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
-			trace_xfs_bdstrat_shut(bp, _RET_IP_);
-
-			xfs_buf_ioerror(bp, -EIO);
-			xfs_buf_stale(bp);
+		else
+			xfs_buf_hold(bp);
 
-			/*
-			 * if sync IO, xfs_buf_ioend is going to remove a
-			 * ref here. We need to add that so we can collect
-			 * all the buffer errors in the wait loop.
-			 */
-			if (wait)
-				xfs_buf_hold(bp);
-			xfs_buf_ioend(bp);
-		} else
-			xfs_buf_iorequest(bp);
+		xfs_buf_submit(bp);
 	}
 	blk_finish_plug(&plug);
 

@@ -1838,7 +1853,10 @@ xfs_buf_delwri_submit(
 		bp = list_first_entry(&io_list, struct xfs_buf, b_list);
 
 		list_del_init(&bp->b_list);
-		error2 = xfs_buf_iowait(bp);
+
+		/* locking the buffer will wait for async IO completion. */
+		xfs_buf_lock(bp);
+		error2 = bp->b_error;
 		xfs_buf_relse(bp);
 		if (!error)
 			error = error2;

fs/xfs/xfs_buf.h +2 -2

index d8f57f6..0acfc30 100644
@@ -290,8 +290,8 @@ extern int xfs_bwrite(struct xfs_buf *bp);
 extern void xfs_buf_ioend(struct xfs_buf *bp);
 extern void xfs_buf_ioerror(xfs_buf_t *, int);
 extern void xfs_buf_ioerror_alert(struct xfs_buf *, const char *func);
-extern void xfs_buf_iorequest(xfs_buf_t *);
-extern int xfs_buf_iowait(xfs_buf_t *);
+extern void xfs_buf_submit(struct xfs_buf *bp);
+extern int xfs_buf_submit_wait(struct xfs_buf *bp);
 extern void xfs_buf_iomove(xfs_buf_t *, size_t, size_t, void *,
 				xfs_buf_rw_t);
 #define xfs_buf_zero(bp, off, len) \

fs/xfs/xfs_buf_item.c +2 -2

index 4fd41b5..cbea909 100644
@@ -1081,7 +1081,7 @@ xfs_buf_iodone_callbacks(
 	 * a way to shut the filesystem down if the writes keep failing.
 	 *
 	 * In practice we'll shut the filesystem down soon as non-transient
-	 * erorrs tend to affect the whole device and a failing log write
+	 * errors tend to affect the whole device and a failing log write
 	 * will make us give up.  But we really ought to do better here.
 	 */
 	if (XFS_BUF_ISASYNC(bp)) {

@@ -1094,7 +1094,7 @@ xfs_buf_iodone_callbacks(
 		if (!(bp->b_flags & (XBF_STALE|XBF_WRITE_FAIL))) {
 			bp->b_flags |= XBF_WRITE | XBF_ASYNC |
 				       XBF_DONE | XBF_WRITE_FAIL;
-			xfs_buf_iorequest(bp);
+			xfs_buf_submit(bp);
 		} else {
 			xfs_buf_relse(bp);
 		}

fs/xfs/xfs_log.c +1 -1

index e4665db..c4d7f79 100644
@@ -1699,7 +1699,7 @@ xlog_bdstrat(
 		return 0;
 	}
 
-	xfs_buf_iorequest(bp);
+	xfs_buf_submit(bp);
 	return 0;
 }
 

fs/xfs/xfs_log_recover.c +7 -15

index 4ba19bf..1e14452 100644
@@ -193,12 +193,8 @@ xlog_bread_noalign(
 	bp->b_io_length = nbblks;
 	bp->b_error = 0;
 
-	if (XFS_FORCED_SHUTDOWN(log->l_mp))
-		return -EIO;
-
-	xfs_buf_iorequest(bp);
-	error = xfs_buf_iowait(bp);
-	if (error)
+	error = xfs_buf_submit_wait(bp);
+	if (error && !XFS_FORCED_SHUTDOWN(log->l_mp))
 		xfs_buf_ioerror_alert(bp, __func__);
 	return error;
 }

@@ -4427,16 +4423,12 @@ xlog_do_recover(
 	XFS_BUF_UNASYNC(bp);
 	bp->b_ops = &xfs_sb_buf_ops;
 
-	if (XFS_FORCED_SHUTDOWN(log->l_mp)) {
-		xfs_buf_relse(bp);
-		return -EIO;
-	}
-
-	xfs_buf_iorequest(bp);
-	error = xfs_buf_iowait(bp);
+	error = xfs_buf_submit_wait(bp);
 	if (error) {
-		xfs_buf_ioerror_alert(bp, __func__);
-		ASSERT(0);
+		if (XFS_FORCED_SHUTDOWN(log->l_mp)) {
+			xfs_buf_ioerror_alert(bp, __func__);
+			ASSERT(0);
+		}
 		xfs_buf_relse(bp);
 		return error;
 	}

fs/xfs/xfs_trans_buf.c +6 -11

index 9c3e610..4bdf02c 100644
@@ -286,18 +286,13 @@ xfs_trans_read_buf_map(
 			bp->b_flags |= XBF_READ;
 			bp->b_ops = ops;
 
-			if (XFS_FORCED_SHUTDOWN(mp)) {
-				trace_xfs_bdstrat_shut(bp, _RET_IP_);
-				error = -EIO;
-				goto out_stale;
-			}
-
-			xfs_buf_iorequest(bp);
-			error = xfs_buf_iowait(bp);
+			error = xfs_buf_submit_wait(bp);
 			if (error) {
-				xfs_buf_ioerror_alert(bp, __func__);
-				goto out_do_shutdown;
-
+				if (!XFS_FORCED_SHUTDOWN(mp)) {
+					xfs_buf_ioerror_alert(bp, __func__);
+					goto out_do_shutdown;
+				}
+				goto out_stale;
 			}
 		}
 

Signature · 6 lines

2.0.0

_______________________________________________
xfs mailing list
[email protected]
http://oss.sgi.com/mailman/listinfo/xfs

Quoted 47 lines · Dave Chinner

From: Dave Chinner <[email protected]>

There is a lot of cookie-cutter code that looks like:

if (shutdown)
handle buffer error
xfs_buf_iorequest(bp)
error = xfs_buf_iowait(bp)
if (error)
handle buffer error

spread through XFS. There's significant complexity now in xfs_buf_iorequest() to specifically handle this sort of synchronous IO pattern, but there's all sorts of nasty surprises in different error handling code dependent on who owns the buffer references and the locks.

Pull this pattern into a single helper, where we can hide all the synchronous IO warts and hence make the error handling for all the callers much saner. This removes the need for a special extra reference to protect IO completion processing, as we can now hold a single reference across dispatch and waiting, simplifying the sync IO smeantics and error handling.

In doing this, also rename xfs_buf_iorequest to xfs_buf_submit and make it explicitly handle on asynchronous IO. This forces all users to be switched specifically to one interface or the other and removes any ambiguity between how the interfaces are to be used. It also means that xfs_buf_iowait() goes away.

For the special case of delwri buffer submission and waiting, we don't need to issue IO synchronously at all. The second pass to cal xfs_buf_iowait() can now just block on xfs_buf_lock() - the buffer will be unlocked when the async IO is complete. This formalises a sane the method of waiting for async IO - take an extra reference, submit the IO, call xfs_buf_lock() when you want to wait for IO completion. i.e.:

	bp = xfs_buf_find();
	xfs_buf_hold(bp);
	bp->b_flags |= XBF_ASYNC;
	xfs_buf_iosubmit(bp);
	xfs_buf_lock(bp)
	error = bp->b_error;
	....
	xfs_buf_relse(bp);

Quoted 2 lines

Signed-off-by Dave Chinner <[email protected]>

---

On a quick look at submit_wait this looks pretty good. It actually implements the general model I've been looking for for sync I/O. E.g., send the I/O, wait on synchronization, then check for errors. In other words, a pure synchronous mechanism. The refactoring and new helpers and whatnot are additional bonus and abstract it nicely.

I still have to take a closer look to review the actual code, but since we go and remove the additional sync I/O reference counting business, why do we even add that stuff early on? Can't we get from where the current code is to here in a more direct manner?

Quoted 57 lines · Brian Foster

Quoted 47 lines · Dave Chinner

From: Dave Chinner <[email protected]>

There is a lot of cookie-cutter code that looks like:

if (shutdown)
handle buffer error
xfs_buf_iorequest(bp)
error = xfs_buf_iowait(bp)
if (error)
handle buffer error

spread through XFS. There's significant complexity now in xfs_buf_iorequest() to specifically handle this sort of synchronous IO pattern, but there's all sorts of nasty surprises in different error handling code dependent on who owns the buffer references and the locks.

Pull this pattern into a single helper, where we can hide all the synchronous IO warts and hence make the error handling for all the callers much saner. This removes the need for a special extra reference to protect IO completion processing, as we can now hold a single reference across dispatch and waiting, simplifying the sync IO smeantics and error handling.

In doing this, also rename xfs_buf_iorequest to xfs_buf_submit and make it explicitly handle on asynchronous IO. This forces all users to be switched specifically to one interface or the other and removes any ambiguity between how the interfaces are to be used. It also means that xfs_buf_iowait() goes away.

For the special case of delwri buffer submission and waiting, we don't need to issue IO synchronously at all. The second pass to cal xfs_buf_iowait() can now just block on xfs_buf_lock() - the buffer will be unlocked when the async IO is complete. This formalises a sane the method of waiting for async IO - take an extra reference, submit the IO, call xfs_buf_lock() when you want to wait for IO completion. i.e.:

	bp = xfs_buf_find();
	xfs_buf_hold(bp);
	bp->b_flags |= XBF_ASYNC;
	xfs_buf_iosubmit(bp);
	xfs_buf_lock(bp)
	error = bp->b_error;
	....
	xfs_buf_relse(bp);

Quoted 2 lines

Signed-off-by Dave Chinner <[email protected]>

---

On a quick look at submit_wait this looks pretty good. It actually implements the general model I've been looking for for sync I/O. E.g., send the I/O, wait on synchronization, then check for errors. In other words, a pure synchronous mechanism. The refactoring and new helpers and whatnot are additional bonus and abstract it nicely.

Quoted 4 lines

I still have to take a closer look to review the actual code, but since we go and remove the additional sync I/O reference counting business, why do we even add that stuff early on? Can't we get from where the current code is to here in a more direct manner?

Simply because we need a fix that we can backport, and that fix is a simple addition that does not significantly affect the rest of the patchset...

Cheers,

Dave.

Signature · 7 lines

Dave Chinner
[email protected]

_______________________________________________
xfs mailing list
[email protected]
http://oss.sgi.com/mailman/listinfo/xfs

From: Dave Chinner <[email protected]>

There is a lot of cookie-cutter code that looks like:

if (shutdown)
handle buffer error
xfs_buf_iorequest(bp)
error = xfs_buf_iowait(bp)
if (error)
handle buffer error

spread through XFS. There's significant complexity now in xfs_buf_iorequest() to specifically handle this sort of synchronous IO pattern, but there's all sorts of nasty surprises in different error handling code dependent on who owns the buffer references and the locks.

Pull this pattern into a single helper, where we can hide all the synchronous IO warts and hence make the error handling for all the callers much saner. This removes the need for a special extra reference to protect IO completion processing, as we can now hold a single reference across dispatch and waiting, simplifying the sync IO smeantics and error handling.

In doing this, also rename xfs_buf_iorequest to xfs_buf_submit and make it explicitly handle on asynchronous IO. This forces all users to be switched specifically to one interface or the other and removes any ambiguity between how the interfaces are to be used. It also means that xfs_buf_iowait() goes away.

For the special case of delwri buffer submission and waiting, we don't need to issue IO synchronously at all. The second pass to cal xfs_buf_iowait() can now just block on xfs_buf_lock() - the buffer will be unlocked when the async IO is complete. This formalises a sane the method of waiting for async IO - take an extra reference, submit the IO, call xfs_buf_lock() when you want to wait for IO completion. i.e.:

	bp = xfs_buf_find();
	xfs_buf_hold(bp);
	bp->b_flags |= XBF_ASYNC;
	xfs_buf_iosubmit(bp);
	xfs_buf_lock(bp)
	error = bp->b_error;
	....
	xfs_buf_relse(bp);

Signed-off-by Dave Chinner <[email protected]>

---
 fs/xfs/xfs_bmap_util.c   |  14 +---
 fs/xfs/xfs_buf.c         | 186 ++++++++++++++++++++++++++---------------------
 fs/xfs/xfs_buf.h         |   4 +-
 fs/xfs/xfs_buf_item.c    |   4 +-
 fs/xfs/xfs_log.c         |   2 +-
 fs/xfs/xfs_log_recover.c |  22 ++----
 fs/xfs/xfs_trans_buf.c   |  17 ++---
 7 files changed, 122 insertions(+), 127 deletions(-)

fs/xfs/xfs_bmap_util.c +2 -12

index 2f1e30d..c53cc03 100644
@@ -1157,12 +1157,7 @@ xfs_zero_remaining_bytes(
 		XFS_BUF_READ(bp);
 		XFS_BUF_SET_ADDR(bp, xfs_fsb_to_db(ip, imap.br_startblock));
 
-		if (XFS_FORCED_SHUTDOWN(mp)) {
-			error = -EIO;
-			break;
-		}
-		xfs_buf_iorequest(bp);
-		error = xfs_buf_iowait(bp);
+		error = xfs_buf_submit_wait(bp);
 		if (error) {
 			xfs_buf_ioerror_alert(bp,
 					"xfs_zero_remaining_bytes(read)");

@@ -1175,12 +1170,7 @@ xfs_zero_remaining_bytes(
 		XFS_BUF_UNREAD(bp);
 		XFS_BUF_WRITE(bp);
 
-		if (XFS_FORCED_SHUTDOWN(mp)) {
-			error = -EIO;
-			break;
-		}
-		xfs_buf_iorequest(bp);
-		error = xfs_buf_iowait(bp);
+		error = xfs_buf_submit_wait(bp);
 		if (error) {
 			xfs_buf_ioerror_alert(bp,
 					"xfs_zero_remaining_bytes(write)");

fs/xfs/xfs_buf.c +26 -30

index 352e9219..a2599f9 100644
@@ -623,10 +623,11 @@ _xfs_buf_read(
 	bp->b_flags &= ~(XBF_WRITE | XBF_ASYNC | XBF_READ_AHEAD);
 	bp->b_flags |= flags & (XBF_READ | XBF_ASYNC | XBF_READ_AHEAD);
 
-	xfs_buf_iorequest(bp);
-	if (flags & XBF_ASYNC)
+	if (flags & XBF_ASYNC) {
+		xfs_buf_submit(bp);
 		return 0;
-	return xfs_buf_iowait(bp);
+	}
+	return xfs_buf_submit_wait(bp);
 }
 
 xfs_buf_t *

@@ -708,12 +709,7 @@ xfs_buf_read_uncached(
 	bp->b_flags |= XBF_READ;
 	bp->b_ops = ops;
 
-	if (XFS_FORCED_SHUTDOWN(target->bt_mount)) {
-		xfs_buf_relse(bp);
-		return NULL;
-	}
-	xfs_buf_iorequest(bp);
-	xfs_buf_iowait(bp);
+	xfs_buf_submit_wait(bp);
 	return bp;
 }
 

@@ -1027,10 +1023,8 @@ xfs_buf_ioend(
 		(*(bp->b_iodone))(bp);
 	else if (bp->b_flags & XBF_ASYNC)
 		xfs_buf_relse(bp);
-	else {
+	else
 		complete(&bp->b_iowait);
-		xfs_buf_rele(bp);
-	}
 }
 
 static void

@@ -1083,21 +1077,7 @@ xfs_bwrite(
 	bp->b_flags &= ~(XBF_ASYNC | XBF_READ | _XBF_DELWRI_Q |
 			 XBF_WRITE_FAIL | XBF_DONE);
 
-	if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
-		trace_xfs_bdstrat_shut(bp, _RET_IP_);
-
-		xfs_buf_ioerror(bp, -EIO);
-		xfs_buf_stale(bp);
-
-		/* sync IO, xfs_buf_ioend is going to remove a ref here */
-		xfs_buf_hold(bp);
-		xfs_buf_ioend(bp);
-		return -EIO;
-	}
-
-	xfs_buf_iorequest(bp);
-
-	error = xfs_buf_iowait(bp);
+	error = xfs_buf_submit_wait(bp);
 	if (error) {
 		xfs_force_shutdown(bp->b_target->bt_mount,
 				   SHUTDOWN_META_IO_ERROR);

@@ -1206,7 +1186,7 @@ next_chunk:
 	} else {
 		/*
 		 * This is guaranteed not to be the last io reference count
-		 * because the caller (xfs_buf_iorequest) holds a count itself.
+		 * because the caller (xfs_buf_submit) holds a count itself.
 		 */
 		atomic_dec(&bp->b_io_remaining);
 		xfs_buf_ioerror(bp, -EIO);

@@ -1296,13 +1276,29 @@ _xfs_buf_ioapply(
 	blk_finish_plug(&plug);
 }
 
+/*
+ * Asynchronous IO submission path. This transfers the buffer lock ownership and
+ * the current reference to the IO. It is not safe to reference the buffer after
+ * a call to this function unless the caller holds an additional reference
+ * itself.
+ */
 void
-xfs_buf_iorequest(
-	xfs_buf_t		*bp)
+xfs_buf_submit(
+	struct xfs_buf	*bp)
 {
 	trace_xfs_buf_iorequest(bp, _RET_IP_);
 
 	ASSERT(!(bp->b_flags & _XBF_DELWRI_Q));
+	ASSERT(bp->b_flags & XBF_ASYNC);
+
+	/* on shutdown we stale and complete the buffer immediately */
+	if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
+		xfs_buf_ioerror(bp, -EIO);
+		bp->b_flags &= ~XBF_DONE;
+		xfs_buf_stale(bp);
+		xfs_buf_ioend(bp);
+		return;
+	}
 
 	if (bp->b_flags & XBF_WRITE)
 		xfs_buf_wait_unpin(bp);

fs/xfs/xfs_buf.c +0 -7

 	/*
-	 * Take references to the buffer. For XBF_ASYNC buffers, holding a
-	 * reference for as long as submission takes is all that is necessary
-	 * here. The IO inherits the lock and hold count from the submitter,
-	 * and these are release during IO completion processing. Taking a hold
-	 * over submission ensures that the buffer is not freed until we have
-	 * completed all processing, regardless of when IO errors occur or are
-	 * reported.

fs/xfs/xfs_buf.c +24 -18

 	xfs_buf_hold(bp);
-	if (!(bp->b_flags & XBF_ASYNC))
-		xfs_buf_hold(bp);
-
 
 	/*
 	 * Set the count to 1 initially, this will stop an I/O completion

@@ -1340,14 +1321,13 @@ xfs_buf_iorequest(
 	_xfs_buf_ioapply(bp);
 
 	/*
-	 * If _xfs_buf_ioapply failed or we are doing synchronous IO that
-	 * completes extremely quickly, we can get back here with only the IO
+	 * If _xfs_buf_ioapply failed, 
+	 * we can get back here with only the IO
 	 * reference we took above.  _xfs_buf_ioend will drop it to zero, so
 	 * we'd better run completion processing synchronously so that the we
-	 * don't return to the caller with completion still pending. In the
-	 * error case, this allows the caller to check b_error safely without
-	 * waiting, and in the synchronous IO case it avoids unnecessary context
-	 * switches an latency for high-peformance devices.
+	 * don't return to the caller with completion still pending.
+	 * this allows the caller to check b_error safely without
+	 * waiting
 	 */
 	if (atomic_dec_and_test(&bp->b_io_remaining) == 1) {
 		if (bp->b_error || !(bp->b_flags & XBF_ASYNC))

@@ -1357,25 +1337,70 @@ xfs_buf_iorequest(
 	}
 
 	xfs_buf_rele(bp);
+	/* Note: it is not safe to reference bp now we've dropped our ref */
 }
 
 /*
- * Waits for I/O to complete on the buffer supplied.  It returns immediately if
- * no I/O is pending or there is already a pending error on the buffer, in which
- * case nothing will ever complete.  It returns the I/O error code, if any, or
- * 0 if there was no error.
+ * Synchronous buffer IO submission path, read or write.
  */
 int
-xfs_buf_iowait(
-	xfs_buf_t		*bp)
+xfs_buf_submit_wait(
+	struct xfs_buf	*bp)
 {
-	trace_xfs_buf_iowait(bp, _RET_IP_);
+	int		error;
+
+	trace_xfs_buf_iorequest(bp, _RET_IP_);
 
-	if (!bp->b_error)
-		wait_for_completion(&bp->b_iowait);
+	ASSERT(!(bp->b_flags & (_XBF_DELWRI_Q | XBF_ASYNC)));
+
+	if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
+		xfs_buf_ioerror(bp, -EIO);
+		xfs_buf_stale(bp);
+		bp->b_flags &= ~XBF_DONE;
+		return -EIO;
+	}
 
+	if (bp->b_flags & XBF_WRITE)
+		xfs_buf_wait_unpin(bp);
+
+	/* clear the internal error state to avoid spurious errors */

fs/xfs/xfs_buf.c +36 -21

+	/*
+	 * Set the count to 1 initially, this will stop an I/O completion
+	 * callout which happens before we have started all the I/O from calling
+	 * xfs_buf_ioend too early.
+	 */
+	atomic_set(&bp->b_io_remaining, 1);
+	_xfs_buf_ioapply(bp);
+
+	/*
+	 * make sure we run completion synchronously if it raced with us and is
+	 * already complete.
+	 */
+	if (atomic_dec_and_test(&bp->b_io_remaining) == 1)
+		xfs_buf_ioend(bp);
+
+	/* wait for completion before gathering the error from the buffer */
+	trace_xfs_buf_iowait(bp, _RET_IP_);
+	wait_for_completion(&bp->b_iowait);
 	trace_xfs_buf_iowait_done(bp, _RET_IP_);
-	return bp->b_error;
+	error = bp->b_error;
+
+	/*
+	 * all done now, we can release the hold that keeps the buffer
+	 * referenced for the entire IO.
+	 */
+	xfs_buf_rele(bp);
+	return error;
 }
 
 xfs_caddr_t

@@ -1769,29 +1794,19 @@ __xfs_buf_delwri_submit(
 	blk_start_plug(&plug);
 	list_for_each_entry_safe(bp, n, io_list, b_list) {
 		bp->b_flags &= ~(_XBF_DELWRI_Q | XBF_ASYNC | XBF_WRITE_FAIL);
-		bp->b_flags |= XBF_WRITE;
+		bp->b_flags |= XBF_WRITE | XBF_ASYNC;
 
-		if (!wait) {
-			bp->b_flags |= XBF_ASYNC;
+		/*
+		 * we do all Io submission async. This means if we need to wait
+		 * for IO completion we need to take an extra reference so the
+		 * buffer is still valid on the other side.
+		 */
+		if (!wait)
 			list_del_init(&bp->b_list);
-		}
-
-		if (XFS_FORCED_SHUTDOWN(bp->b_target->bt_mount)) {
-			trace_xfs_bdstrat_shut(bp, _RET_IP_);
-
-			xfs_buf_ioerror(bp, -EIO);
-			xfs_buf_stale(bp);
+		else
+			xfs_buf_hold(bp);
 
-			/*
-			 * if sync IO, xfs_buf_ioend is going to remove a
-			 * ref here. We need to add that so we can collect
-			 * all the buffer errors in the wait loop.
-			 */
-			if (wait)
-				xfs_buf_hold(bp);
-			xfs_buf_ioend(bp);
-		} else
-			xfs_buf_iorequest(bp);
+		xfs_buf_submit(bp);
 	}
 	blk_finish_plug(&plug);
 

fs/xfs/xfs_buf.c +0 -0

 		xfs_buf_relse(bp);
 		if (!error)
 			error = error2;

fs/xfs/xfs_buf.h +2 -2

index d8f57f6..0acfc30 100644
@@ -290,8 +290,8 @@ extern int xfs_bwrite(struct xfs_buf *bp);
 extern void xfs_buf_ioend(struct xfs_buf *bp);
 extern void xfs_buf_ioerror(xfs_buf_t *, int);
 extern void xfs_buf_ioerror_alert(struct xfs_buf *, const char *func);
-extern void xfs_buf_iorequest(xfs_buf_t *);
-extern int xfs_buf_iowait(xfs_buf_t *);
+extern void xfs_buf_submit(struct xfs_buf *bp);
+extern int xfs_buf_submit_wait(struct xfs_buf *bp);
 extern void xfs_buf_iomove(xfs_buf_t *, size_t, size_t, void *,
 				xfs_buf_rw_t);
 #define xfs_buf_zero(bp, off, len) \

fs/xfs/xfs_buf_item.c +2 -2

index 4fd41b5..cbea909 100644
@@ -1081,7 +1081,7 @@ xfs_buf_iodone_callbacks(
 	 * a way to shut the filesystem down if the writes keep failing.
 	 *
 	 * In practice we'll shut the filesystem down soon as non-transient
-	 * erorrs tend to affect the whole device and a failing log write
+	 * errors tend to affect the whole device and a failing log write
 	 * will make us give up.  But we really ought to do better here.
 	 */
 	if (XFS_BUF_ISASYNC(bp)) {

@@ -1094,7 +1094,7 @@ xfs_buf_iodone_callbacks(
 		if (!(bp->b_flags & (XBF_STALE|XBF_WRITE_FAIL))) {
 			bp->b_flags |= XBF_WRITE | XBF_ASYNC |
 				       XBF_DONE | XBF_WRITE_FAIL;
-			xfs_buf_iorequest(bp);
+			xfs_buf_submit(bp);
 		} else {
 			xfs_buf_relse(bp);
 		}

fs/xfs/xfs_log.c +1 -1

index e4665db..c4d7f79 100644
@@ -1699,7 +1699,7 @@ xlog_bdstrat(
 		return 0;
 	}
 
-	xfs_buf_iorequest(bp);
+	xfs_buf_submit(bp);
 	return 0;
 }
 

fs/xfs/xfs_log_recover.c +2 -8

index 4ba19bf..1e14452 100644
@@ -193,12 +193,8 @@ xlog_bread_noalign(
 	bp->b_io_length = nbblks;
 	bp->b_error = 0;
 
-	if (XFS_FORCED_SHUTDOWN(log->l_mp))
-		return -EIO;
-
-	xfs_buf_iorequest(bp);
-	error = xfs_buf_iowait(bp);
-	if (error)
+	error = xfs_buf_submit_wait(bp);
+	if (error && !XFS_FORCED_SHUTDOWN(log->l_mp))
 		xfs_buf_ioerror_alert(bp, __func__);
 	return error;
 }

@@ -4427,16 +4423,12 @@ xlog_do_recover(
 	XFS_BUF_UNASYNC(bp);
 	bp->b_ops = &xfs_sb_buf_ops;
 
-	if (XFS_FORCED_SHUTDOWN(log->l_mp)) {
-		xfs_buf_relse(bp);