rbd: consolidate request operation creation

I'm working on getting data structures for every rbd image request initialized in one place, and similarly getting all osd object requests for rbd images initialized at once at well. This series addresses an issue I bumped into while trying to do this, and then goes a little further to implement the centralized op initialization.

For read or write osd operations, their offset and length fields as well as the operation's payload length field are assigned fairly late, as the osd request is getting built up in rbd_do_request(). This is sort of strange; all of the other request types get all their fields assigned right after their osd op structure gets created.

This series moves those late assignments for read and write ops earlier, ultimately into the existing "create op" function. It then creates a new function that fully initializes the op with provided arguments for all op types.

We end up with a single function that fully initializes osd ops of any type, as desired.

-Alex

[PATCH 1/6] rbd: don't assign extent info in rbd_do_request() [PATCH 2/6] rbd: don't assign extent info in rbd_req_sync_op() [PATCH 3/6] rbd: initialize off and len in rbd_create_rw_op() [PATCH 4/6] rbd: define generalized osd request op routines [PATCH 5/6] rbd: move call osd op setup into rbd_osd_req_op_create() [PATCH 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 223a401..37b39c1 100644
@@ -1158,13 +1158,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 37b39c1..b2e819c 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 b2e819c..54d11c9 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 54d11c9..17a7781 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 17a7781..97f356f 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 97f356f..0647637 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