libceph: keep source rather than message osd op array
An osd request keeps a pointer to the osd operations (ops) array
that it builds in its request message.
In order to allow each op in the array to have its own distinct
data, we will need to keep track of each op's data, and that
information does not go over the wire.
As long as we're tracking the data we might as well just track the
entire (source) op definition for each of the ops. And if we're
doing that, we'll have no more need to keep a pointer to the
wire-encoded version.
This patch makes the array of source ops be kept with the osd
request structure, and uses that instead of the version encoded in
the message in places where that was previously used. The array
will be embedded in the request structure, and the maximum number of
ops we ever actually use is currently 2. So reduce CEPH_OSD_MAX_OP
to 2 to reduce the size of the structure.
The result of doing this sort of ripples back up, and as a result
various function parameters and local variables become unnecessary.
Make r_num_ops be unsigned, and move the definition of struct
ceph_osd_req_op earlier to ensure it's defined where needed.
It does not yet add per-op data, that's coming soon.
This resolves:
http://tracker.ceph.com/issues/4656
Signed-off-by: Alex Elder <elder@inktank.com>
Reviewed-by: Josh Durgin <josh.durgin@inktank.com>
diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 4a4be14..c12b555 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -1285,7 +1285,7 @@
*/
obj_request->xferred = osd_req->r_reply_op_len[0];
rbd_assert(obj_request->xferred < (u64) UINT_MAX);
- opcode = osd_req->r_request_ops[0].op;
+ opcode = osd_req->r_ops[0].op;
switch (opcode) {
case CEPH_OSD_OP_READ:
rbd_osd_read_callback(obj_request);
@@ -1312,8 +1312,7 @@
}
static void rbd_osd_req_format_op(struct rbd_obj_request *obj_request,
- bool write_request,
- struct ceph_osd_req_op *op)
+ bool write_request)
{
struct rbd_img_request *img_request = obj_request->img_request;
struct ceph_snap_context *snapc = NULL;
@@ -1333,7 +1332,7 @@
}
ceph_osdc_build_request(obj_request->osd_req, obj_request->offset,
- 1, op, snapc, snap_id, mtime);
+ snapc, snap_id, mtime);
}
static struct ceph_osd_request *rbd_osd_req_create(
@@ -1562,7 +1561,7 @@
while (resid) {
const char *object_name;
unsigned int clone_size;
- struct ceph_osd_req_op op;
+ struct ceph_osd_req_op *op;
u64 offset;
u64 length;
@@ -1591,8 +1590,9 @@
if (!obj_request->osd_req)
goto out_partial;
- osd_req_op_extent_init(&op, opcode, offset, length, 0, 0);
- rbd_osd_req_format_op(obj_request, write_request, &op);
+ op = &obj_request->osd_req->r_ops[0];
+ osd_req_op_extent_init(op, opcode, offset, length, 0, 0);
+ rbd_osd_req_format_op(obj_request, write_request);
/* status and version are initially zero-filled */
@@ -1694,7 +1694,7 @@
u64 ver, u64 notify_id)
{
struct rbd_obj_request *obj_request;
- struct ceph_osd_req_op op;
+ struct ceph_osd_req_op *op;
struct ceph_osd_client *osdc;
int ret;
@@ -1708,8 +1708,9 @@
if (!obj_request->osd_req)
goto out;
- osd_req_op_watch_init(&op, CEPH_OSD_OP_NOTIFY_ACK, notify_id, ver, 0);
- rbd_osd_req_format_op(obj_request, false, &op);
+ op = &obj_request->osd_req->r_ops[0];
+ osd_req_op_watch_init(op, CEPH_OSD_OP_NOTIFY_ACK, notify_id, ver, 0);
+ rbd_osd_req_format_op(obj_request, false);
osdc = &rbd_dev->rbd_client->client->osdc;
obj_request->callback = rbd_obj_request_put;
@@ -1749,7 +1750,7 @@
{
struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
struct rbd_obj_request *obj_request;
- struct ceph_osd_req_op op;
+ struct ceph_osd_req_op *op;
int ret;
rbd_assert(start ^ !!rbd_dev->watch_event);
@@ -1773,10 +1774,11 @@
if (!obj_request->osd_req)
goto out_cancel;
- osd_req_op_watch_init(&op, CEPH_OSD_OP_WATCH,
+ op = &obj_request->osd_req->r_ops[0];
+ osd_req_op_watch_init(op, CEPH_OSD_OP_WATCH,
rbd_dev->watch_event->cookie,
rbd_dev->header.obj_version, start);
- rbd_osd_req_format_op(obj_request, true, &op);
+ rbd_osd_req_format_op(obj_request, true);
if (start)
ceph_osdc_set_request_linger(osdc, obj_request->osd_req);
@@ -1836,7 +1838,7 @@
{
struct rbd_obj_request *obj_request;
struct ceph_osd_client *osdc;
- struct ceph_osd_req_op op;
+ struct ceph_osd_req_op *op;
struct page **pages;
u32 page_count;
int ret;
@@ -1866,9 +1868,10 @@
if (!obj_request->osd_req)
goto out;
- osd_req_op_cls_init(&op, CEPH_OSD_OP_CALL, class_name, method_name,
+ op = &obj_request->osd_req->r_ops[0];
+ osd_req_op_cls_init(op, CEPH_OSD_OP_CALL, class_name, method_name,
outbound, outbound_size);
- rbd_osd_req_format_op(obj_request, false, &op);
+ rbd_osd_req_format_op(obj_request, false);
osdc = &rbd_dev->rbd_client->client->osdc;
ret = rbd_obj_request_submit(osdc, obj_request);
@@ -2046,8 +2049,8 @@
char *buf, u64 *version)
{
- struct ceph_osd_req_op op;
struct rbd_obj_request *obj_request;
+ struct ceph_osd_req_op *op;
struct ceph_osd_client *osdc;
struct page **pages = NULL;
u32 page_count;
@@ -2072,8 +2075,9 @@
if (!obj_request->osd_req)
goto out;
- osd_req_op_extent_init(&op, CEPH_OSD_OP_READ, offset, length, 0, 0);
- rbd_osd_req_format_op(obj_request, false, &op);
+ op = &obj_request->osd_req->r_ops[0];
+ osd_req_op_extent_init(op, CEPH_OSD_OP_READ, offset, length, 0, 0);
+ rbd_osd_req_format_op(obj_request, false);
osdc = &rbd_dev->rbd_client->client->osdc;
ret = rbd_obj_request_submit(osdc, obj_request);