[PATCH REPOST 0/6] rbd: consolidate osd request setup

[PATCH REPOST 0/6] rbd: consolidate osd request setup

[Alex Elder]

This series consolidates and encapsulates the setup of all
osd requests into a single function which takes variable
arguments appropriate for the type of request.  The result
groups together common code idioms and I think makes the
spots that build these messages a little easier to read.

					-Alex

[PATCH REPOST 1/6] rbd: don't assign extent info in rbd_do_request()
[PATCH REPOST 2/6] rbd: don't assign extent info in rbd_req_sync_op()
[PATCH REPOST 3/6] rbd: initialize off and len in rbd_create_rw_op()
[PATCH REPOST 4/6] rbd: define generalized osd request op routines
[PATCH REPOST 5/6] rbd: move call osd op setup into rbd_osd_req_op_create()
[PATCH REPOST 6/6] rbd: move remaining osd op setup into
rbd_osd_req_op_create()

[PATCH REPOST 1/6] rbd: don't assign extent info in rbd_do_request()

[Alex Elder]

In rbd_do_request() there's a sort of last-minute assignment of the
extent offset and length and payload length for read and write
operations.  Move those assignments into the caller (in those spots
that might initiate read or write operations)

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index f0b30b2..4ce9981 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -1156,13 +1156,6 @@ static int rbd_do_request(struct request *rq,
 	osd_req->r_oid_len = strlen(osd_req->r_oid);

 	osd_req->r_file_layout = rbd_dev->layout;	/* struct */
-
-	if (op->op == CEPH_OSD_OP_READ || op->op == CEPH_OSD_OP_WRITE) {
-		op->extent.offset = ofs;
-		op->extent.length = len;
-		if (op->op == CEPH_OSD_OP_WRITE)
-			op->payload_len = len;
-	}
 	osd_req->r_num_pages = calc_pages_for(ofs, len);
 	osd_req->r_page_alignment = ofs & ~PAGE_MASK;

@@ -1269,6 +1262,13 @@ static int rbd_req_sync_op(struct rbd_device
*rbd_dev,
 	if (IS_ERR(pages))
 		return PTR_ERR(pages);

+	if (op->op == CEPH_OSD_OP_READ || op->op == CEPH_OSD_OP_WRITE) {
+		op->extent.offset = ofs;
+		op->extent.length = inbound_size;
+		if (op->op == CEPH_OSD_OP_WRITE)
+			op->payload_len = inbound_size;
+	}
+
 	ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,
 			  object_name, ofs, inbound_size, NULL,
 			  pages, num_pages,
@@ -1332,6 +1332,9 @@ static int rbd_do_op(struct request *rq,
 	op = rbd_create_rw_op(opcode, payload_len);
 	if (!op)
 		goto done;
+	op->extent.offset = seg_ofs;
+	op->extent.length = seg_len;
+	op->payload_len = payload_len;

 	/* we've taken care of segment sizes earlier when we
 	   cloned the bios. We should never have a segment
-- 
1.7.9.5

[PATCH REPOST 2/6] rbd: don't assign extent info in rbd_req_sync_op()

[Alex Elder]

Move the assignment of the extent offset and length and payload
length out of rbd_req_sync_op() and into its caller in the one spot
where a read (and note--no write) operation might be initiated.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   10 +++-------
 1 file changed, 3 insertions(+), 7 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 4ce9981..49e500f 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -1262,13 +1262,6 @@ static int rbd_req_sync_op(struct rbd_device
*rbd_dev,
 	if (IS_ERR(pages))
 		return PTR_ERR(pages);

-	if (op->op == CEPH_OSD_OP_READ || op->op == CEPH_OSD_OP_WRITE) {
-		op->extent.offset = ofs;
-		op->extent.length = inbound_size;
-		if (op->op == CEPH_OSD_OP_WRITE)
-			op->payload_len = inbound_size;
-	}
-
 	ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,
 			  object_name, ofs, inbound_size, NULL,
 			  pages, num_pages,
@@ -1373,6 +1366,9 @@ static int rbd_req_sync_read(struct rbd_device
*rbd_dev,
 	op = rbd_create_rw_op(CEPH_OSD_OP_READ, 0);
 	if (!op)
 		return -ENOMEM;
+	op->extent.offset = ofs;
+	op->extent.length = len;
+	op->payload_len = 0;

 	ret = rbd_req_sync_op(rbd_dev, CEPH_OSD_FLAG_READ,
 			       op, object_name, ofs, len, buf, NULL, ver);
-- 
1.7.9.5

[PATCH REPOST 3/6] rbd: initialize off and len in rbd_create_rw_op()

[Alex Elder]

Move the initialization of a read or write operation's offset,
length, and payload length fields into rbd_create_rw_op().

This will actually get removed in the next patch, but it finishes
the consolidation of setting these fields at osd op creation time.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   36 ++++++++++++++++--------------------
 1 file changed, 16 insertions(+), 20 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 49e500f..685d049 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -1025,19 +1025,23 @@ out_err:
 	return NULL;
 }

-static struct ceph_osd_req_op *rbd_create_rw_op(int opcode, u32
payload_len)
+static struct ceph_osd_req_op *rbd_create_rw_op(int opcode, u64 ofs,
u64 len)
 {
 	struct ceph_osd_req_op *op;

 	op = kzalloc(sizeof (*op), GFP_NOIO);
 	if (!op)
 		return NULL;
-	/*
-	 * op extent offset and length will be set later on
-	 * after ceph_calc_file_object_mapping().
-	 */
+
 	op->op = opcode;
-	op->payload_len = payload_len;
+	if (opcode == CEPH_OSD_OP_READ || opcode == CEPH_OSD_OP_WRITE) {
+		op->extent.offset = ofs;
+		op->extent.length = len;
+		if (opcode == CEPH_OSD_OP_WRITE) {
+			rbd_assert(len <= (u64) U32_MAX);
+			op->payload_len = len;
+		}
+	}

 	return op;
 }
@@ -1297,7 +1301,6 @@ static int rbd_do_op(struct request *rq,
 	u64 seg_len;
 	int ret;
 	struct ceph_osd_req_op *op;
-	u32 payload_len;
 	int opcode;
 	int flags;
 	u64 snapid;
@@ -1312,22 +1315,17 @@ static int rbd_do_op(struct request *rq,
 		opcode = CEPH_OSD_OP_WRITE;
 		flags = CEPH_OSD_FLAG_WRITE|CEPH_OSD_FLAG_ONDISK;
 		snapid = CEPH_NOSNAP;
-		payload_len = seg_len;
 	} else {
 		opcode = CEPH_OSD_OP_READ;
 		flags = CEPH_OSD_FLAG_READ;
 		rbd_assert(!snapc);
 		snapid = rbd_dev->spec->snap_id;
-		payload_len = 0;
 	}

 	ret = -ENOMEM;
-	op = rbd_create_rw_op(opcode, payload_len);
+	op = rbd_create_rw_op(opcode, seg_ofs, seg_len);
 	if (!op)
 		goto done;
-	op->extent.offset = seg_ofs;
-	op->extent.length = seg_len;
-	op->payload_len = payload_len;

 	/* we've taken care of segment sizes earlier when we
 	   cloned the bios. We should never have a segment
@@ -1363,12 +1361,9 @@ static int rbd_req_sync_read(struct rbd_device
*rbd_dev,
 	struct ceph_osd_req_op *op;
 	int ret;

-	op = rbd_create_rw_op(CEPH_OSD_OP_READ, 0);
+	op = rbd_create_rw_op(CEPH_OSD_OP_READ, ofs, len);
 	if (!op)
 		return -ENOMEM;
-	op->extent.offset = ofs;
-	op->extent.length = len;
-	op->payload_len = 0;

 	ret = rbd_req_sync_op(rbd_dev, CEPH_OSD_FLAG_READ,
 			       op, object_name, ofs, len, buf, NULL, ver);
@@ -1387,7 +1382,7 @@ static int rbd_req_sync_notify_ack(struct
rbd_device *rbd_dev,
 	struct ceph_osd_req_op *op;
 	int ret;

-	op = rbd_create_rw_op(CEPH_OSD_OP_NOTIFY_ACK, 0);
+	op = rbd_create_rw_op(CEPH_OSD_OP_NOTIFY_ACK, 0, 0);
 	if (!op)
 		return -ENOMEM;

@@ -1438,7 +1433,7 @@ static int rbd_req_sync_watch(struct rbd_device
*rbd_dev, int start)
 	__le64 version = 0;
 	int ret;

-	op = rbd_create_rw_op(CEPH_OSD_OP_WATCH, 0);
+	op = rbd_create_rw_op(CEPH_OSD_OP_WATCH, 0, 0);
 	if (!op)
 		return -ENOMEM;

@@ -1501,9 +1496,10 @@ static int rbd_req_sync_exec(struct rbd_device
*rbd_dev,
 	 * operation.
 	 */
 	payload_size = class_name_len + method_name_len + outbound_size;
-	op = rbd_create_rw_op(CEPH_OSD_OP_CALL, payload_size);
+	op = rbd_create_rw_op(CEPH_OSD_OP_CALL, 0, 0);
 	if (!op)
 		return -ENOMEM;
+	op->payload_len = payload_size;

 	op->cls.class_name = class_name;
 	op->cls.class_len = (__u8) class_name_len;
-- 
1.7.9.5

[PATCH REPOST 4/6] rbd: define generalized osd request op routines

[Alex Elder]

Create a baseline function to encapsulate the creation of osd
requests, along with a matching function to destroy them.  For now
this just duplicates what rbd_create_rw_op() does for read and write
operations, but the next patches will expand on this.

Since rbd_create_rw_op() is no longer used for read or write
requests, the special handling in that function for those types can
be removed.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   52
+++++++++++++++++++++++++++++++++++++++------------
 1 file changed, 40 insertions(+), 12 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 685d049..6fa6ba7 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -1034,14 +1034,6 @@ static struct ceph_osd_req_op
*rbd_create_rw_op(int opcode, u64 ofs, u64 len)
 		return NULL;

 	op->op = opcode;
-	if (opcode == CEPH_OSD_OP_READ || opcode == CEPH_OSD_OP_WRITE) {
-		op->extent.offset = ofs;
-		op->extent.length = len;
-		if (opcode == CEPH_OSD_OP_WRITE) {
-			rbd_assert(len <= (u64) U32_MAX);
-			op->payload_len = len;
-		}
-	}

 	return op;
 }
@@ -1051,6 +1043,42 @@ static void rbd_destroy_op(struct ceph_osd_req_op
*op)
 	kfree(op);
 }

+struct ceph_osd_req_op *rbd_osd_req_op_create(u16 opcode, ...)
+{
+	struct ceph_osd_req_op *op;
+	va_list args;
+
+	op = kzalloc(sizeof (*op), GFP_NOIO);
+	if (!op)
+		return NULL;
+	op->op = opcode;
+	va_start(args, opcode);
+	switch (opcode) {
+	case CEPH_OSD_OP_READ:
+	case CEPH_OSD_OP_WRITE:
+		/* rbd_osd_req_op_create(READ, offset, length) */
+		/* rbd_osd_req_op_create(WRITE, offset, length) */
+		op->extent.offset = va_arg(args, u64);
+		op->extent.length = va_arg(args, u64);
+		if (opcode == CEPH_OSD_OP_WRITE)
+			op->payload_len = op->extent.length;
+		break;
+	default:
+		rbd_warn(NULL, "unsupported opcode %hu\n", opcode);
+		kfree(op);
+		op = NULL;
+		break;
+	}
+	va_end(args);
+
+	return op;
+}
+
+static void rbd_osd_req_op_destroy(struct ceph_osd_req_op *op)
+{
+	kfree(op);
+}
+
 static void rbd_coll_end_req_index(struct request *rq,
 				   struct rbd_req_coll *coll,
 				   int index,
@@ -1323,7 +1351,7 @@ static int rbd_do_op(struct request *rq,
 	}

 	ret = -ENOMEM;
-	op = rbd_create_rw_op(opcode, seg_ofs, seg_len);
+	op = rbd_osd_req_op_create(opcode, seg_ofs, seg_len);
 	if (!op)
 		goto done;

@@ -1343,7 +1371,7 @@ static int rbd_do_op(struct request *rq,
 	if (ret < 0)
 		rbd_coll_end_req_index(rq, coll, coll_index,
 					(s32) ret, seg_len);
-	rbd_destroy_op(op);
+	rbd_osd_req_op_destroy(op);
 done:
 	kfree(seg_name);
 	return ret;
@@ -1361,13 +1389,13 @@ static int rbd_req_sync_read(struct rbd_device
*rbd_dev,
 	struct ceph_osd_req_op *op;
 	int ret;

-	op = rbd_create_rw_op(CEPH_OSD_OP_READ, ofs, len);
+	op = rbd_osd_req_op_create(CEPH_OSD_OP_READ, ofs, len);
 	if (!op)
 		return -ENOMEM;

 	ret = rbd_req_sync_op(rbd_dev, CEPH_OSD_FLAG_READ,
 			       op, object_name, ofs, len, buf, NULL, ver);
-	rbd_destroy_op(op);
+	rbd_osd_req_op_destroy(op);

 	return ret;
 }
-- 
1.7.9.5

[PATCH REPOST 5/6] rbd: move call osd op setup into rbd_osd_req_op_create()

[Alex Elder]

Move the initialization of the CEPH_OSD_OP_CALL operation into
rbd_osd_req_op_create().

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   48 ++++++++++++++++++++++++++++++------------------
 1 file changed, 30 insertions(+), 18 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 6fa6ba7..9f41c32 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -52,10 +52,12 @@
 #define	SECTOR_SHIFT	9
 #define	SECTOR_SIZE	(1ULL << SECTOR_SHIFT)

-/* It might be useful to have this defined elsewhere too */
+/* It might be useful to have these defined elsewhere */

-#define	U32_MAX	((u32) (~0U))
-#define	U64_MAX	((u64) (~0ULL))
+#define	U8_MAX	((u8)	(~0U))
+#define	U16_MAX	((u16)	(~0U))
+#define	U32_MAX	((u32)	(~0U))
+#define	U64_MAX	((u64)	(~0ULL))

 #define RBD_DRV_NAME "rbd"
 #define RBD_DRV_NAME_LONG "rbd (rados block device)"
@@ -1047,6 +1049,7 @@ struct ceph_osd_req_op *rbd_osd_req_op_create(u16
opcode, ...)
 {
 	struct ceph_osd_req_op *op;
 	va_list args;
+	size_t size;

 	op = kzalloc(sizeof (*op), GFP_NOIO);
 	if (!op)
@@ -1063,6 +1066,27 @@ struct ceph_osd_req_op *rbd_osd_req_op_create(u16
opcode, ...)
 		if (opcode == CEPH_OSD_OP_WRITE)
 			op->payload_len = op->extent.length;
 		break;
+	case CEPH_OSD_OP_CALL:
+		/* rbd_osd_req_op_create(CALL, class, method, data, datalen) */
+		op->cls.class_name = va_arg(args, char *);
+		size = strlen(op->cls.class_name);
+		rbd_assert(size <= (size_t) U8_MAX);
+		op->cls.class_len = size;
+		op->payload_len = size;
+
+		op->cls.method_name = va_arg(args, char *);
+		size = strlen(op->cls.method_name);
+		rbd_assert(size <= (size_t) U8_MAX);
+		op->cls.method_len = size;
+		op->payload_len += size;
+
+		op->cls.argc = 0;
+		op->cls.indata = va_arg(args, void *);
+		size = va_arg(args, size_t);
+		rbd_assert(size <= (size_t) U32_MAX);
+		op->cls.indata_len = (u32) size;
+		op->payload_len += size;
+		break;
 	default:
 		rbd_warn(NULL, "unsupported opcode %hu\n", opcode);
 		kfree(op);
@@ -1510,9 +1534,6 @@ static int rbd_req_sync_exec(struct rbd_device
*rbd_dev,
 			     u64 *ver)
 {
 	struct ceph_osd_req_op *op;
-	int class_name_len = strlen(class_name);
-	int method_name_len = strlen(method_name);
-	int payload_size;
 	int ret;

 	/*
@@ -1523,25 +1544,16 @@ static int rbd_req_sync_exec(struct rbd_device
*rbd_dev,
 	 * the perspective of the server side) in the OSD request
 	 * operation.
 	 */
-	payload_size = class_name_len + method_name_len + outbound_size;
-	op = rbd_create_rw_op(CEPH_OSD_OP_CALL, 0, 0);
+	op = rbd_osd_req_op_create(CEPH_OSD_OP_CALL, class_name,
+					method_name, outbound, outbound_size);
 	if (!op)
 		return -ENOMEM;
-	op->payload_len = payload_size;
-
-	op->cls.class_name = class_name;
-	op->cls.class_len = (__u8) class_name_len;
-	op->cls.method_name = method_name;
-	op->cls.method_len = (__u8) method_name_len;
-	op->cls.argc = 0;
-	op->cls.indata = outbound;
-	op->cls.indata_len = outbound_size;

 	ret = rbd_req_sync_op(rbd_dev, CEPH_OSD_FLAG_READ, op,
 			       object_name, 0, inbound_size, inbound,
 			       NULL, ver);

-	rbd_destroy_op(op);
+	rbd_osd_req_op_destroy(op);

 	dout("cls_exec returned %d\n", ret);
 	return ret;
-- 
1.7.9.5

[PATCH REPOST 6/6] rbd: move remaining osd op setup into rbd_osd_req_op_create()

[Alex Elder]

The two remaining osd ops used by rbd are CEPH_OSD_OP_WATCH and
CEPH_OSD_OP_NOTIFY_ACK.  Move the setup of those operations into
rbd_osd_req_op_create(), and get rid of rbd_create_rw_op() and
rbd_destroy_op().

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   68
++++++++++++++++++++-------------------------------
 1 file changed, 27 insertions(+), 41 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 9f41c32..21fbf82 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -1027,24 +1027,6 @@ out_err:
 	return NULL;
 }

-static struct ceph_osd_req_op *rbd_create_rw_op(int opcode, u64 ofs,
u64 len)
-{
-	struct ceph_osd_req_op *op;
-
-	op = kzalloc(sizeof (*op), GFP_NOIO);
-	if (!op)
-		return NULL;
-
-	op->op = opcode;
-
-	return op;
-}
-
-static void rbd_destroy_op(struct ceph_osd_req_op *op)
-{
-	kfree(op);
-}
-
 struct ceph_osd_req_op *rbd_osd_req_op_create(u16 opcode, ...)
 {
 	struct ceph_osd_req_op *op;
@@ -1087,6 +1069,16 @@ struct ceph_osd_req_op *rbd_osd_req_op_create(u16
opcode, ...)
 		op->cls.indata_len = (u32) size;
 		op->payload_len += size;
 		break;
+	case CEPH_OSD_OP_NOTIFY_ACK:
+	case CEPH_OSD_OP_WATCH:
+		/* rbd_osd_req_op_create(NOTIFY_ACK, cookie, version) */
+		/* rbd_osd_req_op_create(WATCH, cookie, version, flag) */
+		op->watch.cookie = va_arg(args, u64);
+		op->watch.ver = va_arg(args, u64);
+		op->watch.ver = cpu_to_le64(op->watch.ver);	/* XXX */
+		if (opcode == CEPH_OSD_OP_WATCH && va_arg(args, int))
+			op->watch.flag = (u8) 1;
+		break;
 	default:
 		rbd_warn(NULL, "unsupported opcode %hu\n", opcode);
 		kfree(op);
@@ -1434,14 +1426,10 @@ static int rbd_req_sync_notify_ack(struct
rbd_device *rbd_dev,
 	struct ceph_osd_req_op *op;
 	int ret;

-	op = rbd_create_rw_op(CEPH_OSD_OP_NOTIFY_ACK, 0, 0);
+	op = rbd_osd_req_op_create(CEPH_OSD_OP_NOTIFY_ACK, notify_id, ver);
 	if (!op)
 		return -ENOMEM;

-	op->watch.ver = cpu_to_le64(ver);
-	op->watch.cookie = notify_id;
-	op->watch.flag = 0;
-
 	ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,
 			  rbd_dev->header_name, 0, 0, NULL,
 			  NULL, 0,
@@ -1450,7 +1438,8 @@ static int rbd_req_sync_notify_ack(struct
rbd_device *rbd_dev,
 			  NULL, 0,
 			  rbd_simple_req_cb, 0, NULL);

-	rbd_destroy_op(op);
+	rbd_osd_req_op_destroy(op);
+
 	return ret;
 }

@@ -1480,14 +1469,9 @@ static void rbd_watch_cb(u64 ver, u64 notify_id,
u8 opcode, void *data)
  */
 static int rbd_req_sync_watch(struct rbd_device *rbd_dev, int start)
 {
-	struct ceph_osd_req_op *op;
 	struct ceph_osd_request **linger_req = NULL;
-	__le64 version = 0;
-	int ret;
-
-	op = rbd_create_rw_op(CEPH_OSD_OP_WATCH, 0, 0);
-	if (!op)
-		return -ENOMEM;
+	struct ceph_osd_req_op *op;
+	int ret = 0;

 	if (start) {
 		struct ceph_osd_client *osdc;
@@ -1496,26 +1480,28 @@ static int rbd_req_sync_watch(struct rbd_device
*rbd_dev, int start)
 		ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0, rbd_dev,
 						&rbd_dev->watch_event);
 		if (ret < 0)
-			goto done;
-		version = cpu_to_le64(rbd_dev->header.obj_version);
+			return ret;
 		linger_req = &rbd_dev->watch_request;
+	} else {
+		rbd_assert(rbd_dev->watch_request != NULL);
 	}

-	op->watch.ver = version;
-	op->watch.cookie = cpu_to_le64(rbd_dev->watch_event->cookie);
-	op->watch.flag = (u8) start ? 1 : 0;
-
-	ret = rbd_req_sync_op(rbd_dev,
+	op = rbd_osd_req_op_create(CEPH_OSD_OP_WATCH,
+				rbd_dev->watch_event->cookie,
+				rbd_dev->header.obj_version, start);
+	if (op)
+		ret = rbd_req_sync_op(rbd_dev,
 			      CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
 			      op, rbd_dev->header_name,
 			      0, 0, NULL, linger_req, NULL);

-	if (!start || ret < 0) {
+	/* Cancel the event if we're tearing down, or on error */
+
+	if (!start || !op || ret < 0) {
 		ceph_osdc_cancel_event(rbd_dev->watch_event);
 		rbd_dev->watch_event = NULL;
 	}
-done:
-	rbd_destroy_op(op);
+	rbd_osd_req_op_destroy(op);

 	return ret;
 }
-- 
1.7.9.5

Re: [PATCH REPOST 6/6] rbd: move remaining osd op setup into rbd_osd_req_op_create()

[Josh Durgin]

On 01/04/2013 07:07 AM, Alex Elder wrote:
> The two remaining osd ops used by rbd are CEPH_OSD_OP_WATCH and
> CEPH_OSD_OP_NOTIFY_ACK.  Move the setup of those operations into
> rbd_osd_req_op_create(), and get rid of rbd_create_rw_op() and
> rbd_destroy_op().
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |   68
> ++++++++++++++++++++-------------------------------
>   1 file changed, 27 insertions(+), 41 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 9f41c32..21fbf82 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -1027,24 +1027,6 @@ out_err:
>   	return NULL;
>   }
>
> -static struct ceph_osd_req_op *rbd_create_rw_op(int opcode, u64 ofs,
> u64 len)
> -{
> -	struct ceph_osd_req_op *op;
> -
> -	op = kzalloc(sizeof (*op), GFP_NOIO);
> -	if (!op)
> -		return NULL;
> -
> -	op->op = opcode;
> -
> -	return op;
> -}
> -
> -static void rbd_destroy_op(struct ceph_osd_req_op *op)
> -{
> -	kfree(op);
> -}
> -
>   struct ceph_osd_req_op *rbd_osd_req_op_create(u16 opcode, ...)
>   {
>   	struct ceph_osd_req_op *op;
> @@ -1087,6 +1069,16 @@ struct ceph_osd_req_op *rbd_osd_req_op_create(u16
> opcode, ...)
>   		op->cls.indata_len = (u32) size;
>   		op->payload_len += size;
>   		break;
> +	case CEPH_OSD_OP_NOTIFY_ACK:
> +	case CEPH_OSD_OP_WATCH:
> +		/* rbd_osd_req_op_create(NOTIFY_ACK, cookie, version) */
> +		/* rbd_osd_req_op_create(WATCH, cookie, version, flag) */
> +		op->watch.cookie = va_arg(args, u64);
> +		op->watch.ver = va_arg(args, u64);
> +		op->watch.ver = cpu_to_le64(op->watch.ver);	/* XXX */

why the /* XXX */ comment?

> +		if (opcode == CEPH_OSD_OP_WATCH && va_arg(args, int))
> +			op->watch.flag = (u8) 1;
> +		break;
>   	default:
>   		rbd_warn(NULL, "unsupported opcode %hu\n", opcode);
>   		kfree(op);
> @@ -1434,14 +1426,10 @@ static int rbd_req_sync_notify_ack(struct
> rbd_device *rbd_dev,
>   	struct ceph_osd_req_op *op;
>   	int ret;
>
> -	op = rbd_create_rw_op(CEPH_OSD_OP_NOTIFY_ACK, 0, 0);
> +	op = rbd_osd_req_op_create(CEPH_OSD_OP_NOTIFY_ACK, notify_id, ver);
>   	if (!op)
>   		return -ENOMEM;
>
> -	op->watch.ver = cpu_to_le64(ver);
> -	op->watch.cookie = notify_id;
> -	op->watch.flag = 0;
> -
>   	ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,
>   			  rbd_dev->header_name, 0, 0, NULL,
>   			  NULL, 0,
> @@ -1450,7 +1438,8 @@ static int rbd_req_sync_notify_ack(struct
> rbd_device *rbd_dev,
>   			  NULL, 0,
>   			  rbd_simple_req_cb, 0, NULL);
>
> -	rbd_destroy_op(op);
> +	rbd_osd_req_op_destroy(op);
> +
>   	return ret;
>   }
>
> @@ -1480,14 +1469,9 @@ static void rbd_watch_cb(u64 ver, u64 notify_id,
> u8 opcode, void *data)
>    */
>   static int rbd_req_sync_watch(struct rbd_device *rbd_dev, int start)
>   {
> -	struct ceph_osd_req_op *op;
>   	struct ceph_osd_request **linger_req = NULL;
> -	__le64 version = 0;
> -	int ret;
> -
> -	op = rbd_create_rw_op(CEPH_OSD_OP_WATCH, 0, 0);
> -	if (!op)
> -		return -ENOMEM;
> +	struct ceph_osd_req_op *op;
> +	int ret = 0;
>
>   	if (start) {
>   		struct ceph_osd_client *osdc;
> @@ -1496,26 +1480,28 @@ static int rbd_req_sync_watch(struct rbd_device
> *rbd_dev, int start)
>   		ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0, rbd_dev,
>   						&rbd_dev->watch_event);
>   		if (ret < 0)
> -			goto done;
> -		version = cpu_to_le64(rbd_dev->header.obj_version);
> +			return ret;
>   		linger_req = &rbd_dev->watch_request;
> +	} else {
> +		rbd_assert(rbd_dev->watch_request != NULL);
>   	}
>
> -	op->watch.ver = version;
> -	op->watch.cookie = cpu_to_le64(rbd_dev->watch_event->cookie);
> -	op->watch.flag = (u8) start ? 1 : 0;
> -
> -	ret = rbd_req_sync_op(rbd_dev,
> +	op = rbd_osd_req_op_create(CEPH_OSD_OP_WATCH,
> +				rbd_dev->watch_event->cookie,
> +				rbd_dev->header.obj_version, start);
> +	if (op)
> +		ret = rbd_req_sync_op(rbd_dev,
>   			      CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
>   			      op, rbd_dev->header_name,
>   			      0, 0, NULL, linger_req, NULL);
>
> -	if (!start || ret < 0) {
> +	/* Cancel the event if we're tearing down, or on error */
> +
> +	if (!start || !op || ret < 0) {
>   		ceph_osdc_cancel_event(rbd_dev->watch_event);
>   		rbd_dev->watch_event = NULL;
>   	}
> -done:
> -	rbd_destroy_op(op);
> +	rbd_osd_req_op_destroy(op);
>
>   	return ret;
>   }
>

Re: [PATCH REPOST 0/6] rbd: consolidate osd request setup

[Josh Durgin]

On 01/04/2013 07:03 AM, Alex Elder wrote:
> This series consolidates and encapsulates the setup of all
> osd requests into a single function which takes variable
> arguments appropriate for the type of request.  The result
> groups together common code idioms and I think makes the
> spots that build these messages a little easier to read.
>
> 					-Alex
>
> [PATCH REPOST 1/6] rbd: don't assign extent info in rbd_do_request()
> [PATCH REPOST 2/6] rbd: don't assign extent info in rbd_req_sync_op()
> [PATCH REPOST 3/6] rbd: initialize off and len in rbd_create_rw_op()
> [PATCH REPOST 4/6] rbd: define generalized osd request op routines
> [PATCH REPOST 5/6] rbd: move call osd op setup into rbd_osd_req_op_create()
> [PATCH REPOST 6/6] rbd: move remaining osd op setup into
> rbd_osd_req_op_create()

I'm not sure about the varargs approach. It makes it easy to
accidentally use the wrong parameters. What do you think about
replacing calls to rbd_osd_req_create_op() with helpers for the various
kinds of requests that just call rbd_osd_req_create_op() themselves, so
that the arguments can be checked at compile time? This will probably
be more of an issue with multi-op osd requests in the future.

Eventually I think all this osd-request-related stuff should go into
libceph, but that's a cleanup for another day.

In any case, the new structure looks good to me.

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

Re: [PATCH REPOST 6/6] rbd: move remaining osd op setup into rbd_osd_req_op_create()

[Alex Elder]

On 01/16/2013 10:23 PM, Josh Durgin wrote:
> On 01/04/2013 07:07 AM, Alex Elder wrote:
>> The two remaining osd ops used by rbd are CEPH_OSD_OP_WATCH and
>> CEPH_OSD_OP_NOTIFY_ACK.  Move the setup of those operations into
>> rbd_osd_req_op_create(), and get rid of rbd_create_rw_op() and
>> rbd_destroy_op().
>>
>> Signed-off-by: Alex Elder <elder@inktank.com>
>> ---
>>   drivers/block/rbd.c |   68
>> ++++++++++++++++++++-------------------------------
>>   1 file changed, 27 insertions(+), 41 deletions(-)
>>
>> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
>> index 9f41c32..21fbf82 100644
>> --- a/drivers/block/rbd.c
>> +++ b/drivers/block/rbd.c
>> @@ -1027,24 +1027,6 @@ out_err:
>>       return NULL;
>>   }
>>
>> -static struct ceph_osd_req_op *rbd_create_rw_op(int opcode, u64 ofs,
>> u64 len)
>> -{
>> -    struct ceph_osd_req_op *op;
>> -
>> -    op = kzalloc(sizeof (*op), GFP_NOIO);
>> -    if (!op)
>> -        return NULL;
>> -
>> -    op->op = opcode;
>> -
>> -    return op;
>> -}
>> -
>> -static void rbd_destroy_op(struct ceph_osd_req_op *op)
>> -{
>> -    kfree(op);
>> -}
>> -
>>   struct ceph_osd_req_op *rbd_osd_req_op_create(u16 opcode, ...)
>>   {
>>       struct ceph_osd_req_op *op;
>> @@ -1087,6 +1069,16 @@ struct ceph_osd_req_op *rbd_osd_req_op_create(u16
>> opcode, ...)
>>           op->cls.indata_len = (u32) size;
>>           op->payload_len += size;
>>           break;
>> +    case CEPH_OSD_OP_NOTIFY_ACK:
>> +    case CEPH_OSD_OP_WATCH:
>> +        /* rbd_osd_req_op_create(NOTIFY_ACK, cookie, version) */
>> +        /* rbd_osd_req_op_create(WATCH, cookie, version, flag) */
>> +        op->watch.cookie = va_arg(args, u64);
>> +        op->watch.ver = va_arg(args, u64);
>> +        op->watch.ver = cpu_to_le64(op->watch.ver);    /* XXX */
> 
> why the /* XXX */ comment?

Because it's the only value here that is converted
from cpu byte order.  It was added in this commit:

    a71b891bc7d77a070e723c8c53d1dd73cf931555
    rbd: send header version when notifying

And I think was done without full understanding that it
was being done different from all the others.  I think
it may be wrong but I haven't really looked at it yet.
Pulling them all into this function made this difference
more obvious.

It was a "note to self" that I wanted to fix that.

I normally try to resolve anything like that before I
post for review but I guess I forgot.  There may be
others.

					-Alex

>> +        if (opcode == CEPH_OSD_OP_WATCH && va_arg(args, int))
>> +            op->watch.flag = (u8) 1;
>> +        break;
>>       default:
>>           rbd_warn(NULL, "unsupported opcode %hu\n", opcode);
>>           kfree(op);
>> @@ -1434,14 +1426,10 @@ static int rbd_req_sync_notify_ack(struct
>> rbd_device *rbd_dev,
>>       struct ceph_osd_req_op *op;
>>       int ret;
>>
>> -    op = rbd_create_rw_op(CEPH_OSD_OP_NOTIFY_ACK, 0, 0);
>> +    op = rbd_osd_req_op_create(CEPH_OSD_OP_NOTIFY_ACK, notify_id, ver);
>>       if (!op)
>>           return -ENOMEM;
>>
>> -    op->watch.ver = cpu_to_le64(ver);
>> -    op->watch.cookie = notify_id;
>> -    op->watch.flag = 0;
>> -
>>       ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,
>>                 rbd_dev->header_name, 0, 0, NULL,
>>                 NULL, 0,
>> @@ -1450,7 +1438,8 @@ static int rbd_req_sync_notify_ack(struct
>> rbd_device *rbd_dev,
>>                 NULL, 0,
>>                 rbd_simple_req_cb, 0, NULL);
>>
>> -    rbd_destroy_op(op);
>> +    rbd_osd_req_op_destroy(op);
>> +
>>       return ret;
>>   }
>>
>> @@ -1480,14 +1469,9 @@ static void rbd_watch_cb(u64 ver, u64 notify_id,
>> u8 opcode, void *data)
>>    */
>>   static int rbd_req_sync_watch(struct rbd_device *rbd_dev, int start)
>>   {
>> -    struct ceph_osd_req_op *op;
>>       struct ceph_osd_request **linger_req = NULL;
>> -    __le64 version = 0;
>> -    int ret;
>> -
>> -    op = rbd_create_rw_op(CEPH_OSD_OP_WATCH, 0, 0);
>> -    if (!op)
>> -        return -ENOMEM;
>> +    struct ceph_osd_req_op *op;
>> +    int ret = 0;
>>
>>       if (start) {
>>           struct ceph_osd_client *osdc;
>> @@ -1496,26 +1480,28 @@ static int rbd_req_sync_watch(struct rbd_device
>> *rbd_dev, int start)
>>           ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0, rbd_dev,
>>                           &rbd_dev->watch_event);
>>           if (ret < 0)
>> -            goto done;
>> -        version = cpu_to_le64(rbd_dev->header.obj_version);
>> +            return ret;
>>           linger_req = &rbd_dev->watch_request;
>> +    } else {
>> +        rbd_assert(rbd_dev->watch_request != NULL);
>>       }
>>
>> -    op->watch.ver = version;
>> -    op->watch.cookie = cpu_to_le64(rbd_dev->watch_event->cookie);
>> -    op->watch.flag = (u8) start ? 1 : 0;
>> -
>> -    ret = rbd_req_sync_op(rbd_dev,
>> +    op = rbd_osd_req_op_create(CEPH_OSD_OP_WATCH,
>> +                rbd_dev->watch_event->cookie,
>> +                rbd_dev->header.obj_version, start);
>> +    if (op)
>> +        ret = rbd_req_sync_op(rbd_dev,
>>                     CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK,
>>                     op, rbd_dev->header_name,
>>                     0, 0, NULL, linger_req, NULL);
>>
>> -    if (!start || ret < 0) {
>> +    /* Cancel the event if we're tearing down, or on error */
>> +
>> +    if (!start || !op || ret < 0) {
>>           ceph_osdc_cancel_event(rbd_dev->watch_event);
>>           rbd_dev->watch_event = NULL;
>>       }
>> -done:
>> -    rbd_destroy_op(op);
>> +    rbd_osd_req_op_destroy(op);
>>
>>       return ret;
>>   }
>>
> 

Re: [PATCH REPOST 0/6] rbd: consolidate osd request setup

[Alex Elder]

On 01/16/2013 10:25 PM, Josh Durgin wrote:
> On 01/04/2013 07:03 AM, Alex Elder wrote:
>> This series consolidates and encapsulates the setup of all
>> osd requests into a single function which takes variable
>> arguments appropriate for the type of request.  The result
>> groups together common code idioms and I think makes the
>> spots that build these messages a little easier to read.
>>
>>                     -Alex
>>
>> [PATCH REPOST 1/6] rbd: don't assign extent info in rbd_do_request()
>> [PATCH REPOST 2/6] rbd: don't assign extent info in rbd_req_sync_op()
>> [PATCH REPOST 3/6] rbd: initialize off and len in rbd_create_rw_op()
>> [PATCH REPOST 4/6] rbd: define generalized osd request op routines
>> [PATCH REPOST 5/6] rbd: move call osd op setup into
>> rbd_osd_req_op_create()
>> [PATCH REPOST 6/6] rbd: move remaining osd op setup into
>> rbd_osd_req_op_create()
> 
> I'm not sure about the varargs approach. It makes it easy to
> accidentally use the wrong parameters. What do you think about
> replacing calls to rbd_osd_req_create_op() with helpers for the various
> kinds of requests that just call rbd_osd_req_create_op() themselves, so
> that the arguments can be checked at compile time? This will probably
> be more of an issue with multi-op osd requests in the future.

I'm OK with splitting it up.

However, at this point I'm afraid it would be quite a pain to
do so in this original set of patches because of the rework
required to layer the subsequent patches on top of it.

I'm going to commit these as-is, and we can talk about
splitting up the big varargs function into pieces later
(maybe soon) as some follow-on work.

> Eventually I think all this osd-request-related stuff should go into
> libceph, but that's a cleanup for another day.

That might make sense.  I have been pretty focused on the
rbd side of things only, with less concern about the bigger
picture.

> In any case, the new structure looks good to me.
> 
> Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

					-Alex

Re: [PATCH REPOST 6/6] rbd: move remaining osd op setup into rbd_osd_req_op_create()

[Alex Elder]

On 01/16/2013 10:37 PM, Alex Elder wrote:
> On 01/16/2013 10:23 PM, Josh Durgin wrote:
>> On 01/04/2013 07:07 AM, Alex Elder wrote:
>>> The two remaining osd ops used by rbd are CEPH_OSD_OP_WATCH and
>>> CEPH_OSD_OP_NOTIFY_ACK.  Move the setup of those operations into
>>> rbd_osd_req_op_create(), and get rid of rbd_create_rw_op() and
>>> rbd_destroy_op().
>>>
>>> Signed-off-by: Alex Elder <elder@inktank.com>

. . .

>>> +    case CEPH_OSD_OP_NOTIFY_ACK:
>>> +    case CEPH_OSD_OP_WATCH:
>>> +        /* rbd_osd_req_op_create(NOTIFY_ACK, cookie, version) */
>>> +        /* rbd_osd_req_op_create(WATCH, cookie, version, flag) */
>>> +        op->watch.cookie = va_arg(args, u64);
>>> +        op->watch.ver = va_arg(args, u64);
>>> +        op->watch.ver = cpu_to_le64(op->watch.ver);    /* XXX */
>>
>> why the /* XXX */ comment?
> 
> Because it's the only value here that is converted
> from cpu byte order.  It was added in this commit:
> 
>     a71b891bc7d77a070e723c8c53d1dd73cf931555
>     rbd: send header version when notifying
> 
> And I think was done without full understanding that it
> was being done different from all the others.  I think
> it may be wrong but I haven't really looked at it yet.
> Pulling them all into this function made this difference
> more obvious.

I've created this to track getting this issue resolved:
    http://tracker.newdream.net/issues/3847

> It was a "note to self" that I wanted to fix that.
> 
> I normally try to resolve anything like that before I
> post for review but I guess I forgot.  There may be
> others.

I'm deleting the "XXX" comment before I commit this change.

					-Alex

. . .