[PATCH V2 0/6] scattered page writeback

[PATCH V2 0/6] scattered page writeback

[Yan, Zheng]

This series adds scattered page writeback, which uses one OSD request
to writeback nonconsecutive dirty page within single strip. Scattered
page writeback can increase performance of buffered random writes.

Yan, Zheng (6):
  libceph: enlarge max number of operations in OSD request
  libceph: move r_reply_op_{len,result} into struct ceph_osd_req_op
  libceph: allow reserving operations in OSD request
  libceph: add helper that extends numner of operations in OSD request
  libceph: add helper that duplicates last extent operation
  ceph: scattered page writeback

 drivers/block/rbd.c             |   2 +-
 fs/ceph/addr.c                  | 221 +++++++++++++++++++++++++---------------
 include/linux/ceph/osd_client.h |  17 +++-
 net/ceph/osd_client.c           | 134 ++++++++++++++++++++----
 4 files changed, 268 insertions(+), 106 deletions(-)

Changes since V1
 . Estimate number of discontiguous dirty pages in ceph_writepages_start()
 . Use kzalloc() to allocate r_ops array
 . Calculate reply message size according to max_ops

-- 
2.5.0

[PATCH V2 1/6] libceph: enlarge max number of operations in OSD request

[Yan, Zheng]

Each operation requires a 'ceph_osd_req_op' structure. To avoid
increasing memory usage of 'struct ceph_osd_request' in ordinary
cases, we dynamically allocate 'ceph_osd_req_op' structures when
number of operations in OSD request are larger than 3

Signed-off-by: Yan, Zheng <zyan@redhat.com>
---
 include/linux/ceph/osd_client.h |  6 ++++--
 net/ceph/osd_client.c           | 46 +++++++++++++++++++++++++++++------------
 2 files changed, 37 insertions(+), 15 deletions(-)

diff --git a/include/linux/ceph/osd_client.h b/include/linux/ceph/osd_client.h
index 7506b48..5bf428a 100644
--- a/include/linux/ceph/osd_client.h
+++ b/include/linux/ceph/osd_client.h
@@ -43,7 +43,8 @@ struct ceph_osd {
 };
 
 
-#define CEPH_OSD_MAX_OP	3
+#define CEPH_OSD_MAX_OP		16
+#define CEPH_OSD_INITIAL_OP	3
 
 enum ceph_osd_data_type {
 	CEPH_OSD_DATA_TYPE_NONE = 0,
@@ -136,7 +137,8 @@ struct ceph_osd_request {
 
 	/* request osd ops array  */
 	unsigned int		r_num_ops;
-	struct ceph_osd_req_op	r_ops[CEPH_OSD_MAX_OP];
+	struct ceph_osd_req_op	*r_ops;
+	struct ceph_osd_req_op	r_inline_ops[CEPH_OSD_INITIAL_OP];
 
 	/* these are updated on each send */
 	__le32           *r_request_osdmap_epoch;
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index f8f2359..b93752e 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -335,12 +335,14 @@ static void ceph_osdc_release_request(struct kref *kref)
 	for (which = 0; which < req->r_num_ops; which++)
 		osd_req_op_data_release(req, which);
 
+	if (req->r_ops != req->r_inline_ops)
+		kfree(req->r_ops);
+
 	ceph_put_snap_context(req->r_snapc);
 	if (req->r_mempool)
 		mempool_free(req, req->r_osdc->req_mempool);
 	else
 		kmem_cache_free(ceph_osd_request_cache, req);
-
 }
 
 void ceph_osdc_get_request(struct ceph_osd_request *req)
@@ -372,16 +374,6 @@ struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
 	BUILD_BUG_ON(CEPH_OSD_MAX_OP > U16_MAX);
 	BUG_ON(num_ops > CEPH_OSD_MAX_OP);
 
-	msg_size = 4 + 4 + 8 + 8 + 4+8;
-	msg_size += 2 + 4 + 8 + 4 + 4; /* oloc */
-	msg_size += 1 + 8 + 4 + 4;     /* pg_t */
-	msg_size += 4 + CEPH_MAX_OID_NAME_LEN; /* oid */
-	msg_size += 2 + num_ops*sizeof(struct ceph_osd_op);
-	msg_size += 8;  /* snapid */
-	msg_size += 8;  /* snap_seq */
-	msg_size += 8 * (snapc ? snapc->num_snaps : 0);  /* snaps */
-	msg_size += 4;
-
 	if (use_mempool) {
 		req = mempool_alloc(osdc->req_mempool, gfp_flags);
 		memset(req, 0, sizeof(*req));
@@ -395,6 +387,17 @@ struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
 	req->r_mempool = use_mempool;
 	req->r_num_ops = num_ops;
 
+	if (num_ops <= CEPH_OSD_INITIAL_OP) {
+		req->r_ops = req->r_inline_ops;
+	} else {
+		BUG_ON(use_mempool);
+		req->r_ops = kzalloc(sizeof(*req->r_ops) * num_ops, gfp_flags);
+		if (!req->r_ops) {
+			ceph_osdc_put_request(req);
+			return NULL;
+		}
+	}
+
 	kref_init(&req->r_kref);
 	init_completion(&req->r_completion);
 	init_completion(&req->r_safe_completion);
@@ -409,11 +412,18 @@ struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
 	req->r_target_oloc.pool = -1;
 
 	/* create reply message */
+	msg_size = OSD_OPREPLY_FRONT_LEN;
+	if (num_ops > CEPH_OSD_INITIAL_OP) {
+		/* ceph_osd_op and op_result */
+		msg_size += (num_ops - CEPH_OSD_INITIAL_OP) *
+			    (sizeof(struct ceph_osd_op) + 4);
+	}
+
 	if (use_mempool)
 		msg = ceph_msgpool_get(&osdc->msgpool_op_reply, 0);
 	else
-		msg = ceph_msg_new(CEPH_MSG_OSD_OPREPLY,
-				   OSD_OPREPLY_FRONT_LEN, gfp_flags, true);
+		msg = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, msg_size,
+				   gfp_flags, true);
 	if (!msg) {
 		ceph_osdc_put_request(req);
 		return NULL;
@@ -421,6 +431,16 @@ struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
 	req->r_reply = msg;
 
 	/* create request message; allow space for oid */
+	msg_size = 4 + 4 + 8 + 8 + 4 + 8;
+	msg_size += 2 + 4 + 8 + 4 + 4; /* oloc */
+	msg_size += 1 + 8 + 4 + 4;     /* pg_t */
+	msg_size += 4 + CEPH_MAX_OID_NAME_LEN; /* oid */
+	msg_size += 2 + num_ops * sizeof(struct ceph_osd_op);
+	msg_size += 8;  /* snapid */
+	msg_size += 8;  /* snap_seq */
+	msg_size += 8 * (snapc ? snapc->num_snaps : 0);  /* snaps */
+	msg_size += 4;
+
 	if (use_mempool)
 		msg = ceph_msgpool_get(&osdc->msgpool_op, 0);
 	else
-- 
2.5.0

[PATCH V2 2/6] libceph: move r_reply_op_{len,result} into struct ceph_osd_req_op

[Yan, Zheng]

This avoids defining large array of r_reply_op_{len,result} in
in 'struct ceph_osd_request'

Signed-off-by: Yan, Zheng <zyan@redhat.com>
---
 drivers/block/rbd.c             | 2 +-
 include/linux/ceph/osd_client.h | 5 +++--
 net/ceph/osd_client.c           | 4 ++--
 3 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 4a87678..22cfc01 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -1854,7 +1854,7 @@ static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
 	 * passed to the block layer, which just supports a 32-bit
 	 * length field.
 	 */
-	obj_request->xferred = osd_req->r_reply_op_len[0];
+	obj_request->xferred = osd_req->r_ops[0].reply_op_len;
 	rbd_assert(obj_request->xferred < (u64)UINT_MAX);
 
 	opcode = osd_req->r_ops[0].op;
diff --git a/include/linux/ceph/osd_client.h b/include/linux/ceph/osd_client.h
index 5bf428a..fda3a05 100644
--- a/include/linux/ceph/osd_client.h
+++ b/include/linux/ceph/osd_client.h
@@ -116,6 +116,9 @@ struct ceph_osd_req_op {
 			u64 expected_write_size;
 		} alloc_hint;
 	};
+	/* reply */
+	int               reply_op_len;
+	s32               reply_op_result;
 };
 
 /* an in-flight request */
@@ -150,8 +153,6 @@ struct ceph_osd_request {
 	struct ceph_eversion *r_request_reassert_version;
 
 	int               r_result;
-	int               r_reply_op_len[CEPH_OSD_MAX_OP];
-	s32               r_reply_op_result[CEPH_OSD_MAX_OP];
 	int               r_got_reply;
 	int		  r_linger;
 
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index b93752e..3852006 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -1840,7 +1840,7 @@ static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg)
 		int len;
 
 		len = le32_to_cpu(op->payload_len);
-		req->r_reply_op_len[i] = len;
+		req->r_ops[i].reply_op_len = len;
 		dout(" op %d has %d bytes\n", i, len);
 		payload_len += len;
 		p += sizeof(*op);
@@ -1855,7 +1855,7 @@ static void handle_reply(struct ceph_osd_client *osdc, struct ceph_msg *msg)
 	ceph_decode_need(&p, end, 4 + numops * 4, bad_put);
 	retry_attempt = ceph_decode_32(&p);
 	for (i = 0; i < numops; i++)
-		req->r_reply_op_result[i] = ceph_decode_32(&p);
+		req->r_ops[i].reply_op_result = ceph_decode_32(&p);
 
 	if (le16_to_cpu(msg->hdr.version) >= 6) {
 		p += 8 + 4; /* skip replay_version */
-- 
2.5.0

[PATCH V2 3/6] libceph: allow reserving operations in OSD request

[Yan, Zheng]

This allows us to reserve some operations for furture use. we do
not need to use all reserved operations.

Signed-off-by: Yan, Zheng <zyan@redhat.com>
---
 include/linux/ceph/osd_client.h | 1 +
 net/ceph/osd_client.c           | 7 +++++--
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/include/linux/ceph/osd_client.h b/include/linux/ceph/osd_client.h
index fda3a05..9802df1 100644
--- a/include/linux/ceph/osd_client.h
+++ b/include/linux/ceph/osd_client.h
@@ -139,6 +139,7 @@ struct ceph_osd_request {
 	u32               r_sent;      /* >0 if r_request is sending/sent */
 
 	/* request osd ops array  */
+	unsigned int		r_max_ops;
 	unsigned int		r_num_ops;
 	struct ceph_osd_req_op	*r_ops;
 	struct ceph_osd_req_op	r_inline_ops[CEPH_OSD_INITIAL_OP];
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index 3852006..9c0cf2e 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -385,7 +385,8 @@ struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
 
 	req->r_osdc = osdc;
 	req->r_mempool = use_mempool;
-	req->r_num_ops = num_ops;
+	req->r_num_ops = 0;
+	req->r_max_ops = num_ops;
 
 	if (num_ops <= CEPH_OSD_INITIAL_OP) {
 		req->r_ops = req->r_inline_ops;
@@ -480,8 +481,10 @@ _osd_req_op_init(struct ceph_osd_request *osd_req, unsigned int which,
 {
 	struct ceph_osd_req_op *op;
 
-	BUG_ON(which >= osd_req->r_num_ops);
+	BUG_ON(which >= osd_req->r_max_ops);
 	BUG_ON(!osd_req_opcode_valid(opcode));
+	if (which >= osd_req->r_num_ops)
+		osd_req->r_num_ops = which + 1;
 
 	op = &osd_req->r_ops[which];
 	memset(op, 0, sizeof (*op));
-- 
2.5.0

[PATCH V2 4/6] libceph: add helper that extends numner of operations in OSD request

[Yan, Zheng]

The helper function resizes the r_ops array and re-allocates request
messages.

Signed-off-by: Yan, Zheng <zyan@redhat.com>
---
 include/linux/ceph/osd_client.h |  3 +++
 net/ceph/osd_client.c           | 56 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 59 insertions(+)

diff --git a/include/linux/ceph/osd_client.h b/include/linux/ceph/osd_client.h
index 9802df1..6430766 100644
--- a/include/linux/ceph/osd_client.h
+++ b/include/linux/ceph/osd_client.h
@@ -322,6 +322,9 @@ extern struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *
 					       unsigned int num_ops,
 					       bool use_mempool,
 					       gfp_t gfp_flags);
+extern int ceph_osdc_extend_request(struct ceph_osd_request *req,
+				    unsigned int num_ops, gfp_t gfp_flags);
+
 
 extern void ceph_osdc_build_request(struct ceph_osd_request *req, u64 off,
 				    struct ceph_snap_context *snapc,
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index 9c0cf2e..c9df912 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -361,6 +361,62 @@ void ceph_osdc_put_request(struct ceph_osd_request *req)
 }
 EXPORT_SYMBOL(ceph_osdc_put_request);
 
+int ceph_osdc_extend_request(struct ceph_osd_request *req,
+			     unsigned int num_ops, gfp_t gfp_flags)
+{
+	struct ceph_osd_req_op *new_ops;
+	struct ceph_msg *request_msg;
+	struct ceph_msg *reply_msg = NULL;
+	size_t msg_size;
+
+	if (num_ops > CEPH_OSD_MAX_OP)
+		return -EINVAL;
+
+	if (num_ops <= req->r_max_ops)
+		return 0;
+
+	msg_size = req->r_request->front_alloc_len +
+		   (num_ops - req->r_max_ops) *
+		   sizeof(struct ceph_osd_op);
+	request_msg = ceph_msg_new(CEPH_MSG_OSD_OP, msg_size, gfp_flags, true);
+	if (!request_msg)
+		return -ENOMEM;
+
+	if (num_ops > CEPH_OSD_INITIAL_OP) {
+		msg_size = req->r_reply->front_alloc_len +
+			   (num_ops - req->r_max_ops) *
+			   (sizeof(struct ceph_osd_op) + 4);
+		reply_msg = ceph_msg_new(CEPH_MSG_OSD_OPREPLY, msg_size,
+					 gfp_flags, true);
+		if (!reply_msg)
+			goto out_enomem;
+
+		new_ops = kzalloc(sizeof(*req->r_ops) * num_ops, gfp_flags);
+		if (!new_ops)
+			goto out_enomem;
+
+		memcpy(new_ops, req->r_ops,
+		       sizeof(*req->r_ops) * req->r_num_ops);
+		if (req->r_ops != req->r_inline_ops)
+			kfree(req->r_ops);
+		req->r_ops = new_ops;
+
+		ceph_msg_put(req->r_reply);
+		req->r_reply = reply_msg;
+	}
+
+	ceph_msg_put(req->r_request);
+	req->r_request = request_msg;
+	req->r_max_ops = num_ops;
+	return 0;
+out_enomem:
+	ceph_msg_put(request_msg);
+	if (reply_msg)
+		ceph_msg_put(reply_msg);
+	return -ENOMEM;
+}
+EXPORT_SYMBOL(ceph_osdc_extend_request);
+
 struct ceph_osd_request *ceph_osdc_alloc_request(struct ceph_osd_client *osdc,
 					       struct ceph_snap_context *snapc,
 					       unsigned int num_ops,
-- 
2.5.0

[PATCH V2 5/6] libceph: add helper that duplicates last extent operation

[Yan, Zheng]

This helper duplicates last extent operation in OSD request, then
adjusts the new extent operation's offset and length. The helper
is for scatterd page writeback, which adds nonconsecutive dirty
pages to single OSD request.

Signed-off-by: Yan, Zheng <zyan@redhat.com>
---
 include/linux/ceph/osd_client.h |  2 ++
 net/ceph/osd_client.c           | 21 +++++++++++++++++++++
 2 files changed, 23 insertions(+)

diff --git a/include/linux/ceph/osd_client.h b/include/linux/ceph/osd_client.h
index 6430766..aa07591 100644
--- a/include/linux/ceph/osd_client.h
+++ b/include/linux/ceph/osd_client.h
@@ -267,6 +267,8 @@ extern void osd_req_op_extent_init(struct ceph_osd_request *osd_req,
 					u64 truncate_size, u32 truncate_seq);
 extern void osd_req_op_extent_update(struct ceph_osd_request *osd_req,
 					unsigned int which, u64 length);
+extern void osd_req_op_extent_dup_last(struct ceph_osd_request *osd_req,
+				       u64 offset_inc);
 
 extern struct ceph_osd_data *osd_req_op_extent_osd_data(
 					struct ceph_osd_request *osd_req,
diff --git a/net/ceph/osd_client.c b/net/ceph/osd_client.c
index c9df912..c0a5665 100644
--- a/net/ceph/osd_client.c
+++ b/net/ceph/osd_client.c
@@ -600,6 +600,27 @@ void osd_req_op_extent_update(struct ceph_osd_request *osd_req,
 }
 EXPORT_SYMBOL(osd_req_op_extent_update);
 
+void osd_req_op_extent_dup_last(struct ceph_osd_request *osd_req,
+				u64 offset_inc)
+{
+	struct ceph_osd_req_op *op, *prev_op;
+	int which;
+
+	BUG_ON(osd_req->r_num_ops == 0);
+	which = osd_req->r_num_ops;
+
+	prev_op = &osd_req->r_ops[which - 1];
+	op = _osd_req_op_init(osd_req, which, prev_op->op, prev_op->flags);
+	/* dup previous one */
+	op->payload_len = prev_op->payload_len;
+	op->extent = prev_op->extent;
+	/* adjust offset */
+	op->extent.offset += offset_inc;
+	op->extent.length -= offset_inc;
+	op->payload_len -= offset_inc;
+}
+EXPORT_SYMBOL(osd_req_op_extent_dup_last);
+
 void osd_req_op_cls_init(struct ceph_osd_request *osd_req, unsigned int which,
 			u16 opcode, const char *class, const char *method)
 {
-- 
2.5.0

[PATCH V2 6/6] ceph: scattered page writeback

[Yan, Zheng]

This patch makes ceph_writepages_start() try using single OSD request
to write all dirty pages within a strip. When a nonconsecutive dirty
page is found, ceph_writepages_start() tries starting a new write
operation to existing OSD request. If it succeeds, it uses the new
operation to writeback the dirty page.

Signed-off-by: Yan, Zheng <zyan@redhat.com>
---
 fs/ceph/addr.c | 221 +++++++++++++++++++++++++++++++++++----------------------
 1 file changed, 137 insertions(+), 84 deletions(-)

diff --git a/fs/ceph/addr.c b/fs/ceph/addr.c
index c222137..a423f02 100644
--- a/fs/ceph/addr.c
+++ b/fs/ceph/addr.c
@@ -606,71 +606,71 @@ static void writepages_finish(struct ceph_osd_request *req,
 	struct inode *inode = req->r_inode;
 	struct ceph_inode_info *ci = ceph_inode(inode);
 	struct ceph_osd_data *osd_data;
-	unsigned wrote;
 	struct page *page;
-	int num_pages;
-	int i;
+	int num_pages, total_pages = 0;
+	int i, j;
+	int rc = req->r_result;
 	struct ceph_snap_context *snapc = req->r_snapc;
 	struct address_space *mapping = inode->i_mapping;
-	int rc = req->r_result;
-	u64 bytes = req->r_ops[0].extent.length;
 	struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
-	long writeback_stat;
-	unsigned issued = ceph_caps_issued(ci);
+	bool remove_page;
 
-	osd_data = osd_req_op_extent_osd_data(req, 0);
-	BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
-	num_pages = calc_pages_for((u64)osd_data->alignment,
-					(u64)osd_data->length);
-	if (rc >= 0) {
-		/*
-		 * Assume we wrote the pages we originally sent.  The
-		 * osd might reply with fewer pages if our writeback
-		 * raced with a truncation and was adjusted at the osd,
-		 * so don't believe the reply.
-		 */
-		wrote = num_pages;
-	} else {
-		wrote = 0;
+
+	dout("writepages_finish %p rc %d\n", inode, rc);
+	if (rc < 0)
 		mapping_set_error(mapping, rc);
-	}
-	dout("writepages_finish %p rc %d bytes %llu wrote %d (pages)\n",
-	     inode, rc, bytes, wrote);
 
-	/* clean all pages */
-	for (i = 0; i < num_pages; i++) {
-		page = osd_data->pages[i];
-		BUG_ON(!page);
-		WARN_ON(!PageUptodate(page));
+	/*
+	 * We lost the cache cap, need to truncate the page before
+	 * it is unlocked, otherwise we'd truncate it later in the
+	 * page truncation thread, possibly losing some data that
+	 * raced its way in
+	 */
+	remove_page = !(ceph_caps_issued(ci) &
+			(CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO));
 
-		writeback_stat =
-			atomic_long_dec_return(&fsc->writeback_count);
-		if (writeback_stat <
-		    CONGESTION_OFF_THRESH(fsc->mount_options->congestion_kb))
-			clear_bdi_congested(&fsc->backing_dev_info,
-					    BLK_RW_ASYNC);
+	/* clean all pages */
+	for (i = 0; i < req->r_num_ops; i++) {
+		if (req->r_ops[i].op != CEPH_OSD_OP_WRITE)
+			break;
 
-		ceph_put_snap_context(page_snap_context(page));
-		page->private = 0;
-		ClearPagePrivate(page);
-		dout("unlocking %d %p\n", i, page);
-		end_page_writeback(page);
+		osd_data = osd_req_op_extent_osd_data(req, i);
+		BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
+		num_pages = calc_pages_for((u64)osd_data->alignment,
+					   (u64)osd_data->length);
+		total_pages += num_pages;
+		for (j = 0; j < num_pages; j++) {
+			page = osd_data->pages[j];
+			BUG_ON(!page);
+			WARN_ON(!PageUptodate(page));
+
+			if (atomic_long_dec_return(&fsc->writeback_count) <
+			     CONGESTION_OFF_THRESH(
+					fsc->mount_options->congestion_kb))
+				clear_bdi_congested(&fsc->backing_dev_info,
+						    BLK_RW_ASYNC);
+
+			ceph_put_snap_context(page_snap_context(page));
+			page->private = 0;
+			ClearPagePrivate(page);
+			dout("unlocking %p\n", page);
+			end_page_writeback(page);
+
+			if (remove_page)
+				generic_error_remove_page(inode->i_mapping,
+							  page);
 
-		/*
-		 * We lost the cache cap, need to truncate the page before
-		 * it is unlocked, otherwise we'd truncate it later in the
-		 * page truncation thread, possibly losing some data that
-		 * raced its way in
-		 */
-		if ((issued & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0)
-			generic_error_remove_page(inode->i_mapping, page);
+			unlock_page(page);
+		}
+		dout("writepages_finish %p wrote %llu bytes cleaned %d pages\n",
+		     inode, osd_data->length, rc >= 0 ? num_pages : 0);
 
-		unlock_page(page);
+		ceph_release_pages(osd_data->pages, num_pages);
 	}
-	dout("%p wrote+cleaned %d pages\n", inode, wrote);
-	ceph_put_wrbuffer_cap_refs(ci, num_pages, snapc);
 
-	ceph_release_pages(osd_data->pages, num_pages);
+	ceph_put_wrbuffer_cap_refs(ci, total_pages, snapc);
+
+	osd_data = osd_req_op_extent_osd_data(req, 0);
 	if (osd_data->pages_from_pool)
 		mempool_free(osd_data->pages,
 			     ceph_sb_to_client(inode->i_sb)->wb_pagevec_pool);
@@ -778,17 +778,15 @@ retry:
 	while (!done && index <= end) {
 		unsigned i;
 		int first;
-		pgoff_t next;
-		int pvec_pages, locked_pages;
-		struct page **pages = NULL;
+		pgoff_t strip_end = 0;
+		int num_ops = 0, max_ops = 0;
+		int pvec_pages, locked_pages = 0;
+		struct page **pages = NULL, **data_pages = NULL;
 		mempool_t *pool = NULL;	/* Becomes non-null if mempool used */
 		struct page *page;
 		int want;
-		u64 offset, len;
-		long writeback_stat;
+		u64 offset = 0, len = 0;
 
-		next = 0;
-		locked_pages = 0;
 		max_pages = max_pages_ever;
 
 get_more_pages:
@@ -824,8 +822,8 @@ get_more_pages:
 				unlock_page(page);
 				break;
 			}
-			if (next && (page->index != next)) {
-				dout("not consecutive %p\n", page);
+			if (strip_end && (page->index > strip_end)) {
+				dout("end of strip %p\n", page);
 				unlock_page(page);
 				break;
 			}
@@ -875,10 +873,11 @@ get_more_pages:
 				/* prepare async write request */
 				offset = (u64)page_offset(page);
 				len = wsize;
+				max_ops = CEPH_OSD_INITIAL_OP;
 				req = ceph_osdc_new_request(&fsc->client->osdc,
 							&ci->i_layout, vino,
-							offset, &len, 0,
-							do_sync ? 2 : 1,
+							offset, &len,
+							0, max_ops,
 							CEPH_OSD_OP_WRITE,
 							CEPH_OSD_FLAG_WRITE |
 							CEPH_OSD_FLAG_ONDISK,
@@ -890,9 +889,8 @@ get_more_pages:
 					break;
 				}
 
-				if (do_sync)
-					osd_req_op_init(req, 1,
-							CEPH_OSD_OP_STARTSYNC, 0);
+				strip_end = page->index +
+					    ((len - 1) >> PAGE_CACHE_SHIFT);
 
 				req->r_callback = writepages_finish;
 				req->r_inode = inode;
@@ -905,6 +903,60 @@ get_more_pages:
 					pages = mempool_alloc(pool, GFP_NOFS);
 					BUG_ON(!pages);
 				}
+
+				num_ops = 1;
+				data_pages = pages;
+				len = 0;
+			} else if (page->index !=
+				   (offset + len) >> PAGE_CACHE_SHIFT) {
+				u64 offset_inc;
+				bool new_op = true;
+
+				if (num_ops + do_sync == CEPH_OSD_MAX_OP) {
+					new_op = false;
+				} else if (num_ops + do_sync == max_ops) {
+					int new_max = max_ops;
+					int j;
+					for (j = i + 1; j < pvec_pages; j++) {
+						if (pvec.pages[j]->index >
+						    strip_end)
+							break;
+						if (pvec.pages[j]->index - 1 !=
+						    pvec.pages[j - 1]->index)
+							new_max++;
+					}
+					new_max++;
+					new_max = min(new_max, CEPH_OSD_MAX_OP);
+					if (ceph_osdc_extend_request(req,
+								new_max,
+								GFP_NOFS) >= 0)
+						max_ops = new_max;
+					else
+						new_op = false;
+				}
+				if (!new_op) {
+					redirty_page_for_writepage(wbc, page);
+					unlock_page(page);
+					break;
+				}
+
+				offset_inc = (u64)page_offset(page) - offset;
+				osd_req_op_extent_dup_last(req, offset_inc);
+
+				dout("writepages got pages at %llu~%llu\n",
+				      offset, len);
+
+				osd_req_op_extent_update(req, num_ops - 1, len);
+				osd_req_op_extent_osd_data_pages(req,
+								num_ops - 1,
+								data_pages,
+								len, 0,
+								!!pool, false);
+
+				num_ops++;
+				data_pages = pages + locked_pages;
+				offset = (u64)page_offset(page);
+				len = 0;
 			}
 
 			/* note position of first page in pvec */
@@ -913,9 +965,8 @@ get_more_pages:
 			dout("%p will write page %p idx %lu\n",
 			     inode, page, page->index);
 
-			writeback_stat =
-			       atomic_long_inc_return(&fsc->writeback_count);
-			if (writeback_stat > CONGESTION_ON_THRESH(
+			if (atomic_long_inc_return(&fsc->writeback_count) >
+			    CONGESTION_ON_THRESH(
 				    fsc->mount_options->congestion_kb)) {
 				set_bdi_congested(&fsc->backing_dev_info,
 						  BLK_RW_ASYNC);
@@ -924,7 +975,7 @@ get_more_pages:
 			set_page_writeback(page);
 			pages[locked_pages] = page;
 			locked_pages++;
-			next = page->index + 1;
+			len += PAGE_CACHE_SIZE;
 		}
 
 		/* did we get anything? */
@@ -952,31 +1003,34 @@ get_more_pages:
 		}
 
 		/* Format the osd request message and submit the write */
-		offset = page_offset(pages[0]);
-		len = (u64)locked_pages << PAGE_CACHE_SHIFT;
 		if (snap_size == -1) {
+			/* writepages_finish() clears writeback pages
+			 * according to the data length, so make sure
+			 * data length covers all locked pages */
+			u64 min_len = len + 1 - PAGE_CACHE_SIZE;
 			len = min(len, (u64)i_size_read(inode) - offset);
-			 /* writepages_finish() clears writeback pages
-			  * according to the data length, so make sure
-			  * data length covers all locked pages */
-			len = max(len, 1 +
-				((u64)(locked_pages - 1) << PAGE_CACHE_SHIFT));
+			len = max(len, min_len);
 		} else {
 			len = min(len, snap_size - offset);
 		}
-		dout("writepages got %d pages at %llu~%llu\n",
-		     locked_pages, offset, len);
+		dout("writepages got pages at %llu~%llu\n", offset, len);
 
-		osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0,
-							!!pool, false);
+		osd_req_op_extent_osd_data_pages(req, num_ops - 1, data_pages,
+						 len, 0, !!pool, false);
+		osd_req_op_extent_update(req, num_ops - 1, len);
 
+		if (do_sync) {
+			osd_req_op_init(req, num_ops, CEPH_OSD_OP_STARTSYNC, 0);
+			num_ops++;
+		}
+		BUG_ON(num_ops > max_ops);
+
+		index = pages[locked_pages - 1]->index + 1;
 		pages = NULL;	/* request message now owns the pages array */
 		pool = NULL;
 
 		/* Update the write op length in case we changed it */
 
-		osd_req_op_extent_update(req, 0, len);
-
 		vino = ceph_vino(inode);
 		ceph_osdc_build_request(req, offset, snapc, vino.snap,
 					&inode->i_mtime);
@@ -986,7 +1040,6 @@ get_more_pages:
 		req = NULL;
 
 		/* continue? */
-		index = next;
 		wbc->nr_to_write -= locked_pages;
 		if (wbc->nr_to_write <= 0)
 			done = 1;
-- 
2.5.0