rbd: consolidate osd request setup

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()

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)

---
 drivers/block/rbd.c |   17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

index f0b30b2..4ce9981 100644
@@ -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

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.

---
 drivers/block/rbd.c |   10 +++-------
 1 file changed, 3 insertions(+), 7 deletions(-)

index 4ce9981..49e500f 100644
@@ -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

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.

---
 drivers/block/rbd.c |   36 ++++++++++++++++--------------------
 1 file changed, 16 insertions(+), 20 deletions(-)

index 49e500f..685d049 100644
@@ -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

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.

---
 drivers/block/rbd.c |   52
+++++++++++++++++++++++++++++++++++++++------------
 1 file changed, 40 insertions(+), 12 deletions(-)

index 685d049..6fa6ba7 100644
@@ -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

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

---
 drivers/block/rbd.c |   48 ++++++++++++++++++++++++++++++------------------
 1 file changed, 30 insertions(+), 18 deletions(-)

index 6fa6ba7..9f41c32 100644
@@ -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

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().

---
 drivers/block/rbd.c |   68
++++++++++++++++++++-------------------------------
 1 file changed, 27 insertions(+), 41 deletions(-)

index 9f41c32..21fbf82 100644
@@ -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

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

. . .

I've created this to track getting this issue resolved:

    http://tracker.newdream.net/issues/3847

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

-Alex

. . .

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.

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.

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

-Alex