[XFS] Initial pass at going directly-to-bio on the buffered IO path. This

allows us to submit much larger I/Os instead of sending down lots of small
buffer_heads.  To do this we need to have a rather complicated I/O
submission and completion tracking infrastructure.  Part of the latter has
been merged already a long time ago for direct I/O support. Part of the
problem is that we need to track sub-pagesize regions and for that we
still need buffer_heads for the time beeing.  Long-term I hope we can move
to better data strucutures and/or maybe move this to fs/mpage.c instead of
having it in XFS.  Original patch from Nathan Scott with various updates
from David Chinner and Christoph Hellwig.

SGI-PV: 947118
SGI-Modid: xfs-linux-melb:xfs-kern:203822a

Signed-off-by: Christoph Hellwig <hch@sgi.com>
Signed-off-by: Nathan Scott <nathans@sgi.com>

1 files changed, 421 insertions, 355 deletions

diff --git a/fs/xfs/linux-2.6/xfs_aops.c b/fs/xfs/linux-2.6/xfs_aops.c
index 3f6b9e29850..e99d04d3fe8 100644
--- a/fs/xfs/linux-2.6/xfs_aops.c
+++ b/fs/xfs/linux-2.6/xfs_aops.c
@@ -43,8 +43,6 @@
 #include <linux/writeback.h>
 
 STATIC void xfs_count_page_state(struct page *, int *, int *, int *);
-STATIC void xfs_convert_page(struct inode *, struct page *, xfs_iomap_t *,
-		struct writeback_control *wbc, void *, int, int);
 
 #if defined(XFS_RW_TRACE)
 void
@@ -58,7 +56,7 @@ xfs_page_trace(
 	bhv_desc_t	*bdp;
 	vnode_t		*vp = LINVFS_GET_VP(inode);
 	loff_t		isize = i_size_read(inode);
-	loff_t		offset = (loff_t)page->index << PAGE_CACHE_SHIFT;
+	loff_t		offset = page_offset(page);
 	int		delalloc = -1, unmapped = -1, unwritten = -1;
 
 	if (page_has_buffers(page))
@@ -103,15 +101,56 @@ xfs_finish_ioend(
 		queue_work(xfsdatad_workqueue, &ioend->io_work);
 }
 
+/*
+ * We're now finished for good with this ioend structure.
+ * Update the page state via the associated buffer_heads,
+ * release holds on the inode and bio, and finally free
+ * up memory.  Do not use the ioend after this.
+ */
 STATIC void
 xfs_destroy_ioend(
 	xfs_ioend_t		*ioend)
 {
+	struct buffer_head	*bh, *next;
+
+	for (bh = ioend->io_buffer_head; bh; bh = next) {
+		next = bh->b_private;
+		bh->b_end_io(bh, ioend->io_uptodate);
+	}
+
 	vn_iowake(ioend->io_vnode);
 	mempool_free(ioend, xfs_ioend_pool);
 }
 
 /*
+ * Buffered IO write completion for delayed allocate extents.
+ * TODO: Update ondisk isize now that we know the file data
+ * has been flushed (i.e. the notorious "NULL file" problem).
+ */
+STATIC void
+xfs_end_bio_delalloc(
+	void			*data)
+{
+	xfs_ioend_t		*ioend = data;
+
+	xfs_destroy_ioend(ioend);
+}
+
+/*
+ * Buffered IO write completion for regular, written extents.
+ */
+STATIC void
+xfs_end_bio_written(
+	void			*data)
+{
+	xfs_ioend_t		*ioend = data;
+
+	xfs_destroy_ioend(ioend);
+}
+
+/*
+ * IO write completion for unwritten extents.
+ *
  * Issue transactions to convert a buffer range from unwritten
  * to written extents.
  */
@@ -123,21 +162,10 @@ xfs_end_bio_unwritten(
 	vnode_t			*vp = ioend->io_vnode;
 	xfs_off_t		offset = ioend->io_offset;
 	size_t			size = ioend->io_size;
-	struct buffer_head	*bh, *next;
 	int			error;
 
 	if (ioend->io_uptodate)
 		VOP_BMAP(vp, offset, size, BMAPI_UNWRITTEN, NULL, NULL, error);
-
-	/* ioend->io_buffer_head is only non-NULL for buffered I/O */
-	for (bh = ioend->io_buffer_head; bh; bh = next) {
-		next = bh->b_private;
-
-		bh->b_end_io = NULL;
-		clear_buffer_unwritten(bh);
-		end_buffer_async_write(bh, ioend->io_uptodate);
-	}
-
 	xfs_destroy_ioend(ioend);
 }
 
@@ -149,7 +177,8 @@ xfs_end_bio_unwritten(
  */
 STATIC xfs_ioend_t *
 xfs_alloc_ioend(
-	struct inode		*inode)
+	struct inode		*inode,
+	unsigned int		type)
 {
 	xfs_ioend_t		*ioend;
 
@@ -162,45 +191,25 @@ xfs_alloc_ioend(
 	 */
 	atomic_set(&ioend->io_remaining, 1);
 	ioend->io_uptodate = 1; /* cleared if any I/O fails */
+	ioend->io_list = NULL;
+	ioend->io_type = type;
 	ioend->io_vnode = LINVFS_GET_VP(inode);
 	ioend->io_buffer_head = NULL;
+	ioend->io_buffer_tail = NULL;
 	atomic_inc(&ioend->io_vnode->v_iocount);
 	ioend->io_offset = 0;
 	ioend->io_size = 0;
 
-	INIT_WORK(&ioend->io_work, xfs_end_bio_unwritten, ioend);
+	if (type == IOMAP_UNWRITTEN)
+		INIT_WORK(&ioend->io_work, xfs_end_bio_unwritten, ioend);
+	else if (type == IOMAP_DELAY)
+		INIT_WORK(&ioend->io_work, xfs_end_bio_delalloc, ioend);
+	else
+		INIT_WORK(&ioend->io_work, xfs_end_bio_written, ioend);
 
 	return ioend;
 }
 
-void
-linvfs_unwritten_done(
-	struct buffer_head	*bh,
-	int			uptodate)
-{
-	xfs_ioend_t		*ioend = bh->b_private;
-	static spinlock_t	unwritten_done_lock = SPIN_LOCK_UNLOCKED;
-	unsigned long		flags;
-
-	ASSERT(buffer_unwritten(bh));
-	bh->b_end_io = NULL;
-
-	if (!uptodate)
-		ioend->io_uptodate = 0;
-
-	/*
-	 * Deep magic here.  We reuse b_private in the buffer_heads to build
-	 * a chain for completing the I/O from user context after we've issued
-	 * a transaction to convert the unwritten extent.
-	 */
-	spin_lock_irqsave(&unwritten_done_lock, flags);
-	bh->b_private = ioend->io_buffer_head;
-	ioend->io_buffer_head = bh;
-	spin_unlock_irqrestore(&unwritten_done_lock, flags);
-
-	xfs_finish_ioend(ioend);
-}
-
 STATIC int
 xfs_map_blocks(
 	struct inode		*inode,
@@ -228,7 +237,7 @@ xfs_offset_to_map(
 	xfs_iomap_t		*iomapp,
 	unsigned long		offset)
 {
-	loff_t			full_offset;	/* offset from start of file */
+	xfs_off_t		full_offset;	/* offset from start of file */
 
 	ASSERT(offset < PAGE_CACHE_SIZE);
 
@@ -243,16 +252,223 @@ xfs_offset_to_map(
 	return NULL;
 }
 
+/*
+ * BIO completion handler for buffered IO.
+ */
+STATIC int
+xfs_end_bio(
+	struct bio		*bio,
+	unsigned int		bytes_done,
+	int			error)
+{
+	xfs_ioend_t		*ioend = bio->bi_private;
+
+	if (bio->bi_size)
+		return 1;
+
+	ASSERT(ioend);
+	ASSERT(atomic_read(&bio->bi_cnt) >= 1);
+
+	/* Toss bio and pass work off to an xfsdatad thread */
+	if (!test_bit(BIO_UPTODATE, &bio->bi_flags))
+		ioend->io_uptodate = 0;
+	bio->bi_private = NULL;
+	bio->bi_end_io = NULL;
+
+	bio_put(bio);
+	xfs_finish_ioend(ioend);
+	return 0;
+}
+
+STATIC void
+xfs_submit_ioend_bio(
+	xfs_ioend_t	*ioend,
+	struct bio	*bio)
+{
+	atomic_inc(&ioend->io_remaining);
+
+	bio->bi_private = ioend;
+	bio->bi_end_io = xfs_end_bio;
+
+	submit_bio(WRITE, bio);
+	ASSERT(!bio_flagged(bio, BIO_EOPNOTSUPP));
+	bio_put(bio);
+}
+
+STATIC struct bio *
+xfs_alloc_ioend_bio(
+	struct buffer_head	*bh)
+{
+	struct bio		*bio;
+	int			nvecs = bio_get_nr_vecs(bh->b_bdev);
+
+	do {
+		bio = bio_alloc(GFP_NOIO, nvecs);
+		nvecs >>= 1;
+	} while (!bio);
+
+	ASSERT(bio->bi_private == NULL);
+	bio->bi_sector = bh->b_blocknr * (bh->b_size >> 9);
+	bio->bi_bdev = bh->b_bdev;
+	bio_get(bio);
+	return bio;
+}
+
+STATIC void
+xfs_start_buffer_writeback(
+	struct buffer_head	*bh)
+{
+	ASSERT(buffer_mapped(bh));
+	ASSERT(buffer_locked(bh));
+	ASSERT(!buffer_delay(bh));
+	ASSERT(!buffer_unwritten(bh));
+
+	mark_buffer_async_write(bh);
+	set_buffer_uptodate(bh);
+	clear_buffer_dirty(bh);
+}
+
+STATIC void
+xfs_start_page_writeback(
+	struct page		*page,
+	struct writeback_control *wbc,
+	int			clear_dirty,
+	int			buffers)
+{
+	ASSERT(PageLocked(page));
+	ASSERT(!PageWriteback(page));
+	set_page_writeback(page);
+	if (clear_dirty)
+		clear_page_dirty(page);
+	unlock_page(page);
+	if (!buffers) {
+		end_page_writeback(page);
+		wbc->pages_skipped++;	/* We didn't write this page */
+	}
+}
+
+static inline int bio_add_buffer(struct bio *bio, struct buffer_head *bh)
+{
+	return bio_add_page(bio, bh->b_page, bh->b_size, bh_offset(bh));
+}
+
+/*
+ * Submit all of the bios for all of the ioends we have saved up,
+ * covering the initial writepage page and also any probed pages.
+ */
+STATIC void
+xfs_submit_ioend(
+	xfs_ioend_t		*ioend)
+{
+	xfs_ioend_t		*next;
+	struct buffer_head	*bh;
+	struct bio		*bio;
+	sector_t		lastblock = 0;
+
+	do {
+		next = ioend->io_list;
+		bio = NULL;
+
+		for (bh = ioend->io_buffer_head; bh; bh = bh->b_private) {
+			xfs_start_buffer_writeback(bh);
+
+			if (!bio) {
+ retry:
+				bio = xfs_alloc_ioend_bio(bh);
+			} else if (bh->b_blocknr != lastblock + 1) {
+				xfs_submit_ioend_bio(ioend, bio);
+				goto retry;
+			}
+
+			if (bio_add_buffer(bio, bh) != bh->b_size) {
+				xfs_submit_ioend_bio(ioend, bio);
+				goto retry;
+			}
+
+			lastblock = bh->b_blocknr;
+		}
+		if (bio)
+			xfs_submit_ioend_bio(ioend, bio);
+		xfs_finish_ioend(ioend);
+	} while ((ioend = next) != NULL);
+}
+
+/*
+ * Cancel submission of all buffer_heads so far in this endio.
+ * Toss the endio too.  Only ever called for the initial page
+ * in a writepage request, so only ever one page.
+ */
+STATIC void
+xfs_cancel_ioend(
+	xfs_ioend_t		*ioend)
+{
+	xfs_ioend_t		*next;
+	struct buffer_head	*bh, *next_bh;
+
+	do {
+		next = ioend->io_list;
+		bh = ioend->io_buffer_head;
+		do {
+			next_bh = bh->b_private;
+			clear_buffer_async_write(bh);
+			unlock_buffer(bh);
+		} while ((bh = next_bh) != NULL);
+
+		vn_iowake(ioend->io_vnode);
+		mempool_free(ioend, xfs_ioend_pool);
+	} while ((ioend = next) != NULL);
+}
+
+/*
+ * Test to see if we've been building up a completion structure for
+ * earlier buffers -- if so, we try to append to this ioend if we
+ * can, otherwise we finish off any current ioend and start another.
+ * Return true if we've finished the given ioend.
+ */
+STATIC void
+xfs_add_to_ioend(
+	struct inode		*inode,
+	struct buffer_head	*bh,
+	unsigned int		p_offset,
+	unsigned int		type,
+	xfs_ioend_t		**result,
+	int			need_ioend)
+{
+	xfs_ioend_t		*ioend = *result;
+
+	if (!ioend || need_ioend || type != ioend->io_type) {
+		xfs_ioend_t	*previous = *result;
+		xfs_off_t	offset;
+
+		offset = (xfs_off_t)bh->b_page->index << PAGE_CACHE_SHIFT;
+		offset += p_offset;
+		ioend = xfs_alloc_ioend(inode, type);
+		ioend->io_offset = offset;
+		ioend->io_buffer_head = bh;
+		ioend->io_buffer_tail = bh;
+		if (previous)
+			previous->io_list = ioend;
+		*result = ioend;
+	} else {
+		ioend->io_buffer_tail->b_private = bh;
+		ioend->io_buffer_tail = bh;
+	}
+
+	bh->b_private = NULL;
+	ioend->io_size += bh->b_size;
+}
+
 STATIC void
 xfs_map_at_offset(
 	struct page		*page,
 	struct buffer_head	*bh,
 	unsigned long		offset,
 	int			block_bits,
-	xfs_iomap_t		*iomapp)
+	xfs_iomap_t		*iomapp,
+	xfs_ioend_t		*ioend)
 {
 	xfs_daddr_t		bn;
-	loff_t			delta;
+	xfs_off_t		delta;
 	int			sector_shift;
 
 	ASSERT(!(iomapp->iomap_flags & IOMAP_HOLE));
@@ -276,60 +492,7 @@ xfs_map_at_offset(
 	bh->b_bdev = iomapp->iomap_target->bt_bdev;
 	set_buffer_mapped(bh);
 	clear_buffer_delay(bh);
-}
-
-/*
- * Look for a page at index which is unlocked and contains our
- * unwritten extent flagged buffers at its head.  Returns page
- * locked and with an extra reference count, and length of the
- * unwritten extent component on this page that we can write,
- * in units of filesystem blocks.
- */
-STATIC struct page *
-xfs_probe_unwritten_page(
-	struct address_space	*mapping,
-	pgoff_t			index,
-	xfs_iomap_t		*iomapp,
-	xfs_ioend_t		*ioend,
-	unsigned long		max_offset,
-	unsigned long		*fsbs,
-	unsigned int            bbits)
-{
-	struct page		*page;
-
-	page = find_trylock_page(mapping, index);
-	if (!page)
-		return NULL;
-	if (PageWriteback(page))
-		goto out;
-
-	if (page->mapping && page_has_buffers(page)) {
-		struct buffer_head	*bh, *head;
-		unsigned long		p_offset = 0;
-
-		*fsbs = 0;
-		bh = head = page_buffers(page);
-		do {
-			if (!buffer_unwritten(bh) || !buffer_uptodate(bh))
-				break;
-			if (!xfs_offset_to_map(page, iomapp, p_offset))
-				break;
-			if (p_offset >= max_offset)
-				break;
-			xfs_map_at_offset(page, bh, p_offset, bbits, iomapp);
-			set_buffer_unwritten_io(bh);
-			bh->b_private = ioend;
-			p_offset += bh->b_size;
-			(*fsbs)++;
-		} while ((bh = bh->b_this_page) != head);
-
-		if (p_offset)
-			return page;
-	}
-
-out:
-	unlock_page(page);
-	return NULL;
+	clear_buffer_unwritten(bh);
 }
 
 /*
@@ -372,15 +535,16 @@ out:
 	return ret;
 }
 
-STATIC unsigned int
+STATIC size_t
 xfs_probe_unmapped_cluster(
 	struct inode		*inode,
 	struct page		*startpage,
 	struct buffer_head	*bh,
 	struct buffer_head	*head)
 {
+	size_t			len, total = 0;
 	pgoff_t			tindex, tlast, tloff;
-	unsigned int		pg_offset, len, total = 0;
+	unsigned int		pg_offset;
 	struct address_space	*mapping = inode->i_mapping;
 
 	/* First sum forwards in this page */
@@ -414,14 +578,15 @@ xfs_probe_unmapped_cluster(
 }
 
 /*
- * Probe for a given page (index) in the inode and test if it is delayed
- * and without unwritten buffers.  Returns page locked and with an extra
- * reference count.
+ * Probe for a given page (index) in the inode and test if it is suitable
+ * for writing as part of an unwritten or delayed allocate extent.
+ * Returns page locked and with an extra reference count if so, else NULL.
  */
 STATIC struct page *
-xfs_probe_delalloc_page(
+xfs_probe_delayed_page(
 	struct inode		*inode,
-	pgoff_t			index)
+	pgoff_t			index,
+	unsigned int		type)
 {
 	struct page		*page;
 
@@ -437,12 +602,12 @@ xfs_probe_delalloc_page(
 
 		bh = head = page_buffers(page);
 		do {
-			if (buffer_unwritten(bh)) {
-				acceptable = 0;
+			if (buffer_unwritten(bh))
+				acceptable = (type == IOMAP_UNWRITTEN);
+			else if (buffer_delay(bh))
+				acceptable = (type == IOMAP_DELAY);
+			else
 				break;
-			} else if (buffer_delay(bh)) {
-				acceptable = 1;
-			}
 		} while ((bh = bh->b_this_page) != head);
 
 		if (acceptable)
@@ -454,161 +619,30 @@ out:
 	return NULL;
 }
 
-STATIC int
-xfs_map_unwritten(
-	struct inode		*inode,
-	struct page		*start_page,
-	struct buffer_head	*head,
-	struct buffer_head	*curr,
-	unsigned long		p_offset,
-	int			block_bits,
-	xfs_iomap_t		*iomapp,
-	struct writeback_control *wbc,
-	int			startio,
-	int			all_bh)
-{
-	struct buffer_head	*bh = curr;
-	xfs_iomap_t		*tmp;
-	xfs_ioend_t		*ioend;
-	loff_t			offset;
-	unsigned long		nblocks = 0;
-
-	offset = start_page->index;
-	offset <<= PAGE_CACHE_SHIFT;
-	offset += p_offset;
-
-	ioend = xfs_alloc_ioend(inode);
-
-	/* First map forwards in the page consecutive buffers
-	 * covering this unwritten extent
-	 */
-	do {
-		if (!buffer_unwritten(bh))
-			break;
-		tmp = xfs_offset_to_map(start_page, iomapp, p_offset);
-		if (!tmp)
-			break;
-		xfs_map_at_offset(start_page, bh, p_offset, block_bits, iomapp);
-		set_buffer_unwritten_io(bh);
-		bh->b_private = ioend;
-		p_offset += bh->b_size;
-		nblocks++;
-	} while ((bh = bh->b_this_page) != head);
-
-	atomic_add(nblocks, &ioend->io_remaining);
-
-	/* If we reached the end of the page, map forwards in any
-	 * following pages which are also covered by this extent.
-	 */
-	if (bh == head) {
-		struct address_space	*mapping = inode->i_mapping;
-		pgoff_t			tindex, tloff, tlast;
-		unsigned long		bs;
-		unsigned int		pg_offset, bbits = inode->i_blkbits;
-		struct page		*page;
-
-		tlast = i_size_read(inode) >> PAGE_CACHE_SHIFT;
-		tloff = (iomapp->iomap_offset + iomapp->iomap_bsize) >> PAGE_CACHE_SHIFT;
-		tloff = min(tlast, tloff);
-		for (tindex = start_page->index + 1; tindex < tloff; tindex++) {
-			page = xfs_probe_unwritten_page(mapping,
-						tindex, iomapp, ioend,
-						PAGE_CACHE_SIZE, &bs, bbits);
-			if (!page)
-				break;
-			nblocks += bs;
-			atomic_add(bs, &ioend->io_remaining);
-			xfs_convert_page(inode, page, iomapp, wbc, ioend,
-							startio, all_bh);
-			/* stop if converting the next page might add
-			 * enough blocks that the corresponding byte
-			 * count won't fit in our ulong page buf length */
-			if (nblocks >= ((ULONG_MAX - PAGE_SIZE) >> block_bits))
-				goto enough;
-		}
-
-		if (tindex == tlast &&
-		    (pg_offset = (i_size_read(inode) & (PAGE_CACHE_SIZE - 1)))) {
-			page = xfs_probe_unwritten_page(mapping,
-							tindex, iomapp, ioend,
-							pg_offset, &bs, bbits);
-			if (page) {
-				nblocks += bs;
-				atomic_add(bs, &ioend->io_remaining);
-				xfs_convert_page(inode, page, iomapp, wbc, ioend,
-							startio, all_bh);
-				if (nblocks >= ((ULONG_MAX - PAGE_SIZE) >> block_bits))
-					goto enough;
-			}
-		}
-	}
-
-enough:
-	ioend->io_size = (xfs_off_t)nblocks << block_bits;
-	ioend->io_offset = offset;
-	xfs_finish_ioend(ioend);
-	return 0;
-}
-
-STATIC void
-xfs_submit_page(
-	struct page		*page,
-	struct writeback_control *wbc,
-	struct buffer_head	*bh_arr[],
-	int			bh_count,
-	int			probed_page,
-	int			clear_dirty)
-{
-	struct buffer_head	*bh;
-	int			i;
-
-	BUG_ON(PageWriteback(page));
-	if (bh_count)
-		set_page_writeback(page);
-	if (clear_dirty)
-		clear_page_dirty(page);
-	unlock_page(page);
-
-	if (bh_count) {
-		for (i = 0; i < bh_count; i++) {
-			bh = bh_arr[i];
-			mark_buffer_async_write(bh);
-			if (buffer_unwritten(bh))
-				set_buffer_unwritten_io(bh);
-			set_buffer_uptodate(bh);
-			clear_buffer_dirty(bh);
-		}
-
-		for (i = 0; i < bh_count; i++)
-			submit_bh(WRITE, bh_arr[i]);
-
-		if (probed_page && clear_dirty)
-			wbc->nr_to_write--;	/* Wrote an "extra" page */
-	}
-}
-
 /*
  * Allocate & map buffers for page given the extent map. Write it out.
  * except for the original page of a writepage, this is called on
  * delalloc/unwritten pages only, for the original page it is possible
  * that the page has no mapping at all.
  */
-STATIC void
+STATIC int
 xfs_convert_page(
 	struct inode		*inode,
 	struct page		*page,
 	xfs_iomap_t		*iomapp,
+	xfs_ioend_t		**ioendp,
 	struct writeback_control *wbc,
 	void			*private,
 	int			startio,
 	int			all_bh)
 {
-	struct buffer_head	*bh_arr[MAX_BUF_PER_PAGE], *bh, *head;
+	struct buffer_head	*bh, *head;
 	xfs_iomap_t		*mp = iomapp, *tmp;
-	unsigned long		offset, end_offset;
-	int			index = 0;
+	unsigned long		p_offset, end_offset;
+	unsigned int		type;
 	int			bbits = inode->i_blkbits;
 	int			len, page_dirty;
+	int			count = 0, done = 0, uptodate = 1;
 
 	end_offset = (i_size_read(inode) & (PAGE_CACHE_SIZE - 1));
 
@@ -621,59 +655,66 @@ xfs_convert_page(
 	end_offset = roundup(end_offset, len);
 	page_dirty = end_offset / len;
 
-	offset = 0;
+	p_offset = 0;
 	bh = head = page_buffers(page);
 	do {
-		if (offset >= end_offset)
+		if (p_offset >= end_offset)
 			break;
-		if (!(PageUptodate(page) || buffer_uptodate(bh)))
+		if (!buffer_uptodate(bh))
+			uptodate = 0;
+		if (!(PageUptodate(page) || buffer_uptodate(bh))) {
+			done = 1;
 			continue;
-		if (buffer_mapped(bh) && all_bh &&
-		    !(buffer_unwritten(bh) || buffer_delay(bh))) {
-			if (startio) {
+		}
+
+		if (buffer_unwritten(bh))
+			type = IOMAP_UNWRITTEN;
+		else if (buffer_delay(bh))
+			type = IOMAP_DELAY;
+		else {
+			type = 0;
+			if (!(buffer_mapped(bh) && all_bh && startio)) {
+				done = 1;
+			} else if (startio) {
 				lock_buffer(bh);
-				bh_arr[index++] = bh;
+				xfs_add_to_ioend(inode, bh, p_offset,
+						type, ioendp, done);
+				count++;
 				page_dirty--;
 			}
 			continue;
 		}
-		tmp = xfs_offset_to_map(page, mp, offset);
-		if (!tmp)
+		tmp = xfs_offset_to_map(page, mp, p_offset);
+		if (!tmp) {
+			done = 1;
 			continue;
+		}
 		ASSERT(!(tmp->iomap_flags & IOMAP_HOLE));
 		ASSERT(!(tmp->iomap_flags & IOMAP_DELAY));
 
-		/* If this is a new unwritten extent buffer (i.e. one
-		 * that we haven't passed in private data for, we must
-		 * now map this buffer too.
-		 */
-		if (buffer_unwritten(bh) && !bh->b_end_io) {
-			ASSERT(tmp->iomap_flags & IOMAP_UNWRITTEN);
-			xfs_map_unwritten(inode, page, head, bh, offset,
-					bbits, tmp, wbc, startio, all_bh);
-		} else if (! (buffer_unwritten(bh) && buffer_locked(bh))) {
-			xfs_map_at_offset(page, bh, offset, bbits, tmp);
-			if (buffer_unwritten(bh)) {
-				set_buffer_unwritten_io(bh);
-				bh->b_private = private;
-				ASSERT(private);
-			}
-		}
+		xfs_map_at_offset(page, bh, p_offset, bbits, tmp, *ioendp);
 		if (startio) {
-			bh_arr[index++] = bh;
+			xfs_add_to_ioend(inode, bh, p_offset,
+					type, ioendp, done);
+			count++;
 		} else {
 			set_buffer_dirty(bh);
 			unlock_buffer(bh);
 			mark_buffer_dirty(bh);
 		}
 		page_dirty--;
-	} while (offset += len, (bh = bh->b_this_page) != head);
+	} while (p_offset += len, (bh = bh->b_this_page) != head);
 
-	if (startio && index) {
-		xfs_submit_page(page, wbc, bh_arr, index, 1, !page_dirty);
-	} else {
-		unlock_page(page);
+	if (uptodate && bh == head)
+		SetPageUptodate(page);
+
+	if (startio) {
+		if (count)
+			wbc->nr_to_write--;
+		xfs_start_page_writeback(page, wbc, !page_dirty, count);
 	}
+
+	return done;
 }
 
 /*
@@ -685,19 +726,22 @@ xfs_cluster_write(
 	struct inode		*inode,
 	pgoff_t			tindex,
 	xfs_iomap_t		*iomapp,
+	xfs_ioend_t		**ioendp,
 	struct writeback_control *wbc,
 	int			startio,
 	int			all_bh,
 	pgoff_t			tlast)
 {
 	struct page		*page;
+	unsigned int		type = (*ioendp)->io_type;
+	int			done;
 
-	for (; tindex <= tlast; tindex++) {
-		page = xfs_probe_delalloc_page(inode, tindex);
+	for (done = 0; tindex <= tlast && !done; tindex++) {
+		page = xfs_probe_delayed_page(inode, tindex, type);
 		if (!page)
 			break;
-		xfs_convert_page(inode, page, iomapp, wbc, NULL,
-				startio, all_bh);
+		done = xfs_convert_page(inode, page, iomapp, ioendp,
+						wbc, NULL, startio, all_bh);
 	}
 }
 
@@ -728,18 +772,21 @@ xfs_page_state_convert(
 	int		startio,
 	int		unmapped) /* also implies page uptodate */
 {
-	struct buffer_head	*bh_arr[MAX_BUF_PER_PAGE], *bh, *head;
+	struct buffer_head	*bh, *head;
 	xfs_iomap_t		*iomp, iomap;
+	xfs_ioend_t		*ioend = NULL, *iohead = NULL;
 	loff_t			offset;
 	unsigned long           p_offset = 0;
+	unsigned int		type;
 	__uint64_t              end_offset;
 	pgoff_t                 end_index, last_index, tlast;
-	int			len, err, i, cnt = 0, uptodate = 1;
-	int			flags;
-	int			page_dirty;
+	int			flags, len, err, done = 1;
+	int			uptodate = 1;
+	int			page_dirty, count = 0, trylock_flag = 0;
 
 	/* wait for other IO threads? */
-	flags = (startio && wbc->sync_mode != WB_SYNC_NONE) ? 0 : BMAPI_TRYLOCK;
+	if (startio && wbc->sync_mode != WB_SYNC_NONE)
+		trylock_flag |= BMAPI_TRYLOCK;
 
 	/* Is this page beyond the end of the file? */
 	offset = i_size_read(inode);
@@ -754,98 +801,98 @@ xfs_page_state_convert(
 		}
 	}
 
-	end_offset = min_t(unsigned long long,
-			(loff_t)(page->index + 1) << PAGE_CACHE_SHIFT, offset);
-	offset = (loff_t)page->index << PAGE_CACHE_SHIFT;
-
 	/*
 	 * page_dirty is initially a count of buffers on the page before
 	 * EOF and is decrememted as we move each into a cleanable state.
-	 */
+	 *
+	 * Derivation:
+	 *
+	 * End offset is the highest offset that this page should represent.
+	 * If we are on the last page, (end_offset & (PAGE_CACHE_SIZE - 1))
+	 * will evaluate non-zero and be less than PAGE_CACHE_SIZE and
+	 * hence give us the correct page_dirty count. On any other page,
+	 * it will be zero and in that case we need page_dirty to be the
+	 * count of buffers on the page.
+ 	 */
+	end_offset = min_t(unsigned long long,
+			(xfs_off_t)(page->index + 1) << PAGE_CACHE_SHIFT, offset);
 	len = 1 << inode->i_blkbits;
-	p_offset = max(p_offset, PAGE_CACHE_SIZE);
-	p_offset = roundup(p_offset, len);
+	p_offset = min_t(unsigned long, end_offset & (PAGE_CACHE_SIZE - 1),
+					PAGE_CACHE_SIZE);
+	p_offset = p_offset ? roundup(p_offset, len) : PAGE_CACHE_SIZE;
 	page_dirty = p_offset / len;
 
 	iomp = NULL;
-	p_offset = 0;
 	bh = head = page_buffers(page);
+	offset = page_offset(page);
+
+	/* TODO: fix up "done" variable and iomap pointer (boolean) */
+	/* TODO: cleanup count and page_dirty */
 
 	do {
 		if (offset >= end_offset)
 			break;
 		if (!buffer_uptodate(bh))
 			uptodate = 0;
-		if (!(PageUptodate(page) || buffer_uptodate(bh)) && !startio)
+		if (!(PageUptodate(page) || buffer_uptodate(bh)) && !startio) {
+			done = 1;
 			continue;
+		}
 
 		if (iomp) {
 			iomp = xfs_offset_to_map(page, &iomap, p_offset);
+			done = (iomp == NULL);
 		}
 
 		/*
 		 * First case, map an unwritten extent and prepare for
 		 * extent state conversion transaction on completion.
-		 */
-		if (buffer_unwritten(bh)) {
-			if (!startio)
-				continue;
-			if (!iomp) {
-				err = xfs_map_blocks(inode, offset, len, &iomap,
-						BMAPI_WRITE|BMAPI_IGNSTATE);
-				if (err) {
-					goto error;
-				}
-				iomp = xfs_offset_to_map(page, &iomap,
-								p_offset);
-			}
-			if (iomp) {
-				if (!bh->b_end_io) {
-					err = xfs_map_unwritten(inode, page,
-							head, bh, p_offset,
-							inode->i_blkbits, iomp,
-							wbc, startio, unmapped);
-					if (err) {
-						goto error;
-					}
-				} else {
-					set_bit(BH_Lock, &bh->b_state);
-				}
-				BUG_ON(!buffer_locked(bh));
-				bh_arr[cnt++] = bh;
-				page_dirty--;
-			}
-		/*
+		 *
 		 * Second case, allocate space for a delalloc buffer.
 		 * We can return EAGAIN here in the release page case.
 		 */
-		} else if (buffer_delay(bh)) {
+		if (buffer_unwritten(bh) || buffer_delay(bh)) {
+			if (buffer_unwritten(bh)) {
+				type = IOMAP_UNWRITTEN;
+				flags = BMAPI_WRITE|BMAPI_IGNSTATE;
+			} else {
+				type = IOMAP_DELAY;
+				flags = BMAPI_ALLOCATE;
+				if (!startio)
+					flags |= trylock_flag;
+			}
+
 			if (!iomp) {
+				done = 1;
 				err = xfs_map_blocks(inode, offset, len, &iomap,
-						BMAPI_ALLOCATE | flags);
-				if (err) {
+						flags);
+				if (err)
 					goto error;
-				}
 				iomp = xfs_offset_to_map(page, &iomap,
 								p_offset);
+				done = (iomp == NULL);
 			}
 			if (iomp) {
 				xfs_map_at_offset(page, bh, p_offset,
-						inode->i_blkbits, iomp);
+						inode->i_blkbits, iomp, ioend);
 				if (startio) {
-					bh_arr[cnt++] = bh;
+					xfs_add_to_ioend(inode, bh, p_offset,
+						type, &ioend, done);
 				} else {
 					set_buffer_dirty(bh);
 					unlock_buffer(bh);
 					mark_buffer_dirty(bh);
 				}
 				page_dirty--;
+				count++;
+			} else {
+				done = 1;
 			}
 		} else if ((buffer_uptodate(bh) || PageUptodate(page)) &&
 			   (unmapped || startio)) {
 
+			type = 0;
 			if (!buffer_mapped(bh)) {
-				int	size;
 
 				/*
 				 * Getting here implies an unmapped buffer
@@ -853,6 +900,8 @@ xfs_page_state_convert(
 				 * need to write the whole page out.
 				 */
 				if (!iomp) {
+					int	size;
+
 					size = xfs_probe_unmapped_cluster(
 							inode, page, bh, head);
 					err = xfs_map_blocks(inode, offset,
@@ -863,52 +912,70 @@ xfs_page_state_convert(
 					}
 					iomp = xfs_offset_to_map(page, &iomap,
 								     p_offset);
+					done = (iomp == NULL);
 				}
 				if (iomp) {
-					xfs_map_at_offset(page,
-							bh, p_offset,
-							inode->i_blkbits, iomp);
+					xfs_map_at_offset(page, bh, p_offset,
+							inode->i_blkbits, iomp,
+							ioend);
 					if (startio) {
-						bh_arr[cnt++] = bh;
+						xfs_add_to_ioend(inode,
+							bh, p_offset, type,
+							&ioend, done);
 					} else {
 						set_buffer_dirty(bh);
 						unlock_buffer(bh);
 						mark_buffer_dirty(bh);
 					}
 					page_dirty--;
+					count++;
+				} else {
+					done = 1;
 				}
 			} else if (startio) {
 				if (buffer_uptodate(bh) &&
 				    !test_and_set_bit(BH_Lock, &bh->b_state)) {
-					bh_arr[cnt++] = bh;
+					ASSERT(buffer_mapped(bh));
+					xfs_add_to_ioend(inode,
+							bh, p_offset, type,
+							&ioend, done);
 					page_dirty--;
+					count++;
+				} else {
+					done = 1;
 				}
+			} else {
+				done = 1;
 			}
 		}
-	} while (offset += len, p_offset += len,
-		((bh = bh->b_this_page) != head));
+
+		if (!iohead)
+			iohead = ioend;
+
+	} while (offset += len, ((bh = bh->b_this_page) != head));
 
 	if (uptodate && bh == head)
 		SetPageUptodate(page);
 
-	if (startio) {
-		xfs_submit_page(page, wbc, bh_arr, cnt, 0, !page_dirty);
-	}
+	if (startio)
+		xfs_start_page_writeback(page, wbc, 1, count);
 
-	if (iomp) {
+	if (ioend && iomp && !done) {
 		offset = (iomp->iomap_offset + iomp->iomap_bsize - 1) >>
 					PAGE_CACHE_SHIFT;
 		tlast = min_t(pgoff_t, offset, last_index);
-		xfs_cluster_write(inode, page->index + 1, iomp, wbc,
-					startio, unmapped, tlast);
+		xfs_cluster_write(inode, page->index + 1, iomp, &ioend,
+					wbc, startio, unmapped, tlast);
 	}
 
+	if (iohead)
+		xfs_submit_ioend(iohead);
+
 	return page_dirty;
 
 error:
-	for (i = 0; i < cnt; i++) {
-		unlock_buffer(bh_arr[i]);
-	}
+	if (iohead)
+		xfs_cancel_ioend(iohead);
 
 	/*
 	 * If it's delalloc and we have nowhere to put it,
@@ -916,9 +983,8 @@ error:
 	 * us to try again.
 	 */
 	if (err != -EAGAIN) {
-		if (!unmapped) {
+		if (!unmapped)
 			block_invalidatepage(page, 0);
-		}
 		ClearPageUptodate(page);
 	}
 	return err;
@@ -1094,7 +1160,7 @@ linvfs_direct_IO(
 	if (error)
 		return -error;
 
-	iocb->private = xfs_alloc_ioend(inode);
+	iocb->private = xfs_alloc_ioend(inode, IOMAP_UNWRITTEN);
 
 	ret = blockdev_direct_IO_own_locking(rw, iocb, inode,
 		iomap.iomap_target->bt_bdev,