[PATCH 0/6] rbd: consolidate request operation creation

[PATCH 0/6] rbd: consolidate request operation creation

[Alex Elder]

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

[PATCH 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 223a401..37b39c1 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 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 37b39c1..b2e819c 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 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 b2e819c..54d11c9 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 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 54d11c9..17a7781 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 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 17a7781..97f356f 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 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 97f356f..0647637 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