rbd: new request tracking code

This series of patches reimplements the way rbd requests are managed.

The motivation for doing this is that the existing code assumes in places that every I/O request was associated with a Linux request initiated from the block layer. In order to support layered rbd images, we need to initiate rbd image requests within rbd. In addition, the size of an rbd object in a parent image could be smaller than the object size in the child. That means satisfying a request to a single object in the child image may require issuing requests to multiple objects in the parent. This too was not easily supported by the existing code.

In this new code, we distinguish between an "image request" and an "object request." An image request is an I/O operation defined in terms of an entire rbd image--its offset and length are relative to the image as a whole. An object request is a request targeted at a single rbd object. An image request is implemented using one or more object requests, but individual object requests can also be used for doing things like reading or operating on the rbd image header object.

The new code also more clearly separates itself from the osd client, especially with respect to request completion. Each object request supplies a common callback function with its osd client request. This callback handles certain rbd-specific tasks, like zero-filling portions of a read buffer that were not supplied by the. A simple wait routine is now supplied to allow synchronous completion of an object request.

This set of changes is available in the ceph-client git repository branch "wip-rbd-review". That branch also begins with three unrelated commits (which I recently posted for review) and ends with two others (which are temporary and there only to test with version 2 image support enabled).

-Alex

[PATCH 01/12] rbd: new request tracking code
[PATCH 02/12] rbd: kill rbd_rq_fn() and all other related code [PATCH 03/12] rbd: kill rbd_req_coll and rbd_request [PATCH 04/12] rbd: implement sync object read with new code [PATCH 05/12] rbd: get rid of rbd_req_sync_read() [PATCH 06/12] rbd: implement watch/unwatch with new code [PATCH 07/12] rbd: get rid of rbd_req_sync_watch() [PATCH 08/12] rbd: use new code for notify ack [PATCH 09/12] rbd: get rid of rbd_req_sync_notify_ack() [PATCH 10/12] rbd: send notify ack asynchronously [PATCH 11/12] rbd: implement sync method with new code [PATCH 12/12] rbd: get rid of rbd_req_sync_exec()

This patch fully implements the new request tracking code for rbd I/O requests.

Each I/O request to an rbd image will get an rbd_image_request structure allocated to track it. This provides access to all information about the original request, as well as access to the set of one or more object requests that are initiated as a result of the image request.

An rbd_obj_request structure defines a request sent to a single osd object (possibly) as part of an rbd image request. An rbd object request refers to a ceph_osd_request structure built up to represent the request; for now it will contain a single osd operation. It also provides space to hold the result status and the version of the object when the osd request completes.

An rbd_obj_request structure can also stand on its own. This will be used for reading the version 1 header object, for issuing acknowledgements to event notifications, and for making object method calls.

All rbd object requests now complete asynchronously with respect to the osd client--they supply a common callback routine.

This resolves:

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

---
 drivers/block/rbd.c |  596
++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 594 insertions(+), 2 deletions(-)

index 6689363..485fa70 100644
@@ -181,6 +181,55 @@ struct rbd_req_coll {
 	struct rbd_req_status	status[0];
 };
 
+struct rbd_image_request;
+
+enum obj_req_type { obj_req_bio };	/* More types to come */
+
+struct rbd_obj_request;
+typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);
+
+struct rbd_obj_request {
+	const char		*object_name;
+	u64			offset;		/* object start byte */
+	u64			length;		/* bytes from offset */
+
+	struct rbd_image_request *image_request;
+	u32			which;		/* posn in image req array */
+
+	enum obj_req_type	type;
+	struct bio		*bio_list;
+
+	struct ceph_osd_request	*osd_req;
+
+	u64			xferred;	/* bytes transferred */
+	u64			version;
+	s32			result;
+	atomic_t		done;
+
+	rbd_obj_callback_t	callback;
+
+	struct kref		kref;
+};
+
+struct rbd_image_request {
+	struct request		*rq;
+	struct rbd_device	*rbd_dev;
+	union {
+		struct ceph_snap_context *snapc;	/* for writes */
+		u64		snap_id;		/* for reads */
+	};
+	u64			offset;	/* starting image byte offset */
+	u64			length;	/* byte count from offset */
+	spinlock_t		completion_lock;
+	u32			next_completion;
+	bool			write_request;	/* false for read */
+	u32			obj_req_count;
+
+	struct kref		kref;
+
+	struct rbd_obj_request	*obj_requests[0];
+};
+
 /*
  * a single io request
  */

@@ -1031,6 +1080,40 @@ out_err:
 	return NULL;
 }
 
+static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
+{
+	kref_get(&obj_request->kref);
+}
+
+static void rbd_obj_request_destroy(struct kref *kref);
+static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
+{
+	rbd_assert(obj_request != NULL);
+	kref_put(&obj_request->kref, rbd_obj_request_destroy);
+}
+
+static void rbd_image_request_get(struct rbd_image_request *image_request)
+{
+	kref_get(&image_request->kref);
+}
+
+static void rbd_image_request_destroy(struct kref *kref);
+static void rbd_image_request_put(struct rbd_image_request *image_request)
+{
+	rbd_assert(image_request != NULL);
+	kref_put(&image_request->kref, rbd_image_request_destroy);
+}
+
+static bool obj_req_type_valid(enum obj_req_type type)
+{
+	switch (type) {
+	case obj_req_bio:
+		return true;
+	default:
+		return false;
+	}
+}
+
 struct ceph_osd_req_op *rbd_osd_req_op_create(u16 opcode, ...)
 {
 	struct ceph_osd_req_op *op;

@@ -1395,6 +1478,19 @@ done:
 	return ret;
 }
 
+static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
+				struct rbd_obj_request *obj_request)
+{
+	int ret;
+
+	rbd_obj_request_get(obj_request);
+	ret = ceph_osdc_start_request(osdc, obj_request->osd_req, false);
+	if (ret)
+		rbd_obj_request_put(obj_request);
+
+	return ret;
+}
+
 /*
  * Request sync osd read
  */

@@ -1618,6 +1714,502 @@ static int rbd_dev_do_request(struct request *rq,
 	return 0;
 }
 
+/* Returns true if this call completed the last object request */
+
+static bool rbd_block_request_complete(struct rbd_image_request

*image_request,
+ u32 which)
+
+{
+ struct request *rq = image_request->rq;
+ bool more = true;
+
+ rbd_assert(rq != NULL);
+ rbd_assert(which < image_request->obj_req_count);
+ rbd_assert(which >= image_request->next_completion);
+
+ rbd_image_request_put(image_request);
+
+ spin_lock(&image_request->completion_lock);
+ if (which != image_request->next_completion)
+ goto out;
+ do {
+ struct rbd_obj_request *obj_request;
+ unsigned int xferred;
+ int result;
+
+ rbd_assert(more);
+ obj_request = image_request->obj_requests[which];
+ rbd_assert(obj_request != NULL);
+ if (!atomic_read(&obj_request->done))
+ break;
+ rbd_assert(obj_request->xferred <= (u64) UINT_MAX);
+ xferred = (unsigned int) obj_request->xferred;
+ result = (int) obj_request->result;
+
+ more = blk_end_request(rq, result, xferred);
+ } while (++which < image_request->obj_req_count);
+ rbd_assert(more ^ (which == image_request->obj_req_count));
+ image_request->next_completion = which;
+out:
+ spin_unlock(&image_request->completion_lock);
+
+ return !more;
+}
+
+static void rbd_osd_read_callback(struct rbd_obj_request *obj_request,
+ struct ceph_osd_op *op)
+{
+ u64 xferred;
+
+ /*
+ * We support a 64-bit length, but ultimately it has to be
+ * passed to blk_end_request(), which takes an unsigned int.
+ */
+ xferred = le64_to_cpu(op->extent.length);
+ rbd_assert(xferred < (u64) UINT_MAX);
+ if (obj_request->result == (s32) -ENOENT) {
+ zero_bio_chain(obj_request->bio_list, 0);
+ obj_request->result = 0;
+ } else if (xferred < obj_request->length && !obj_request->result) {
+ zero_bio_chain(obj_request->bio_list, xferred);
+ xferred = obj_request->length;
+ }
+ obj_request->xferred = xferred;
+ atomic_set(&obj_request->done, 1);
+}
+
+static void rbd_osd_write_callback(struct rbd_obj_request *obj_request,
+ struct ceph_osd_op *op)
+{
+ obj_request->xferred = le64_to_cpu(op->extent.length);
+ atomic_set(&obj_request->done, 1);
+}
+
+static void rbd_obj_request_complete(struct rbd_obj_request *obj_request) +{
+ if (obj_request->callback)
+ obj_request->callback(obj_request);
+}
+
+static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
+ struct ceph_msg *msg)
+{
+ struct rbd_obj_request *obj_request = osd_req->r_priv;
+ struct ceph_osd_reply_head *reply_head;
+ struct ceph_osd_op *op;
+ u32 num_ops;
+ u16 opcode;
+
+ rbd_assert(osd_req == obj_request->osd_req);
+ rbd_obj_request_put(obj_request);
+ obj_request->xferred = le32_to_cpu(msg->hdr.data_len);
+ reply_head = msg->front.iov_base;
+ obj_request->result = (s32) le32_to_cpu(reply_head->result);
+ obj_request->version = le64_to_cpu(osd_req->r_reassert_version.version);
+
+ num_ops = le32_to_cpu(reply_head->num_ops);
+ WARN_ON(num_ops != 1); /* For now */
+
+ op = &reply_head->ops[0];
+ opcode = le16_to_cpu(op->op);
+ switch (opcode) {
+ case CEPH_OSD_OP_READ:
+ rbd_osd_read_callback(obj_request, op);
+ break;
+ case CEPH_OSD_OP_WRITE:
+ rbd_osd_write_callback(obj_request, op);
+ break;
+ default:
+ rbd_warn(NULL, "%s: unsupported op %hu\n",
+ obj_request->object_name, (unsigned short) opcode);
+ break;
+ }
+
+ if (atomic_read(&obj_request->done))
+ rbd_obj_request_complete(obj_request);
+}
+
+static struct ceph_osd_request *rbd_osd_req_create(
+ struct rbd_device *rbd_dev,
+ bool write_request,
+ struct rbd_obj_request *obj_request,
+ struct ceph_osd_req_op *op)
+{
+ struct rbd_image_request *image_request = obj_request->image_request;
+ struct ceph_snap_context *snapc = NULL;
+ struct ceph_osd_client *osdc;
+ struct ceph_osd_request *osd_req;
+ struct timespec now;
+ struct timespec *mtime;
+ u64 snap_id = CEPH_NOSNAP;
+ u64 offset = obj_request->offset;
+ u64 length = obj_request->length;
+
+ if (image_request) {
+ rbd_assert(image_request->write_request == write_request);
+ if (image_request->write_request)
+ snapc = image_request->snapc;
+ else
+ snap_id = image_request->snap_id;
+ }
+
+ /* Allocate and initialize the request, for the single op */
+
+ osdc = &rbd_dev->rbd_client->client->osdc;
+ osd_req = ceph_osdc_alloc_request(osdc, snapc, 1, false, GFP_ATOMIC);
+ if (!osd_req)
+ return NULL; /* ENOMEM */
+
+ rbd_assert(obj_req_type_valid(obj_request->type));
+ switch (obj_request->type) {
+ case obj_req_bio:
+ rbd_assert(obj_request->bio_list != NULL);
+ osd_req->r_bio = obj_request->bio_list;
+ bio_get(osd_req->r_bio);
+ /* osd client requires "num pages" even for bio */
+ osd_req->r_num_pages = calc_pages_for(offset, length);
+ break;
+ }
+
+ if (write_request) {
+ osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
+ now = CURRENT_TIME;
+ mtime = &now;
+ } else {
+ osd_req->r_flags = CEPH_OSD_FLAG_READ;
+ mtime = NULL; /* not needed for reads */
+ offset = 0; /* These are not used... */
+ length = 0; /* ...for osd read requests */
+ }
+
+ osd_req->r_callback = rbd_osd_req_callback;
+ osd_req->r_priv = obj_request;
+
+ /* No trailing '\0' required for the object name in the request */
+
+ osd_req->r_oid_len = strlen(obj_request->object_name);
+ rbd_assert(osd_req->r_oid_len <= sizeof (osd_req->r_oid));
+ memcpy(osd_req->r_oid, obj_request->object_name, osd_req->r_oid_len);
+
+ osd_req->r_file_layout = rbd_dev->layout; /* struct */
+
+ /* osd_req will get its own reference to snapc (if non-null) */
+
+ ceph_osdc_build_request(osd_req, offset, length, 1, op,
+ snapc, snap_id, mtime);
+
+ return osd_req;
+}
+
+static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req) +{
+ ceph_osdc_put_request(osd_req);
+}
+
+/* object_name is assumed to be a non-null pointer and NUL-terminated */ +
+static struct rbd_obj_request *rbd_obj_request_create(const char *object_name,
+ u64 offset, u64 length,
+ enum obj_req_type type)
+{
+ struct rbd_obj_request *obj_request;
+ size_t size;
+ char *name;
+
+ rbd_assert(obj_req_type_valid(type));
+
+ size = strlen(object_name) + 1;
+ obj_request = kzalloc(sizeof (*obj_request) + size, GFP_KERNEL);
+ if (!obj_request)
+ return NULL;
+
+ name = (char *)(obj_request + 1);
+ obj_request->object_name = memcpy(name, object_name, size);
+ obj_request->offset = offset;
+ obj_request->length = length;
+ obj_request->type = type;
+ atomic_set(&obj_request->done, 0);
+ kref_init(&obj_request->kref);
+
+ return obj_request;
+}
+
+static void rbd_obj_request_destroy(struct kref *kref)
+{
+ struct rbd_obj_request *obj_request;
+
+ obj_request = container_of(kref, struct rbd_obj_request, kref);
+ if (obj_request->osd_req)
+ rbd_osd_req_destroy(obj_request->osd_req);
+
+ rbd_assert(obj_req_type_valid(obj_request->type));
+ switch (obj_request->type) {
+ case obj_req_bio:
+ if (obj_request->bio_list)
+ bio_chain_put(obj_request->bio_list);
+ break;
+ }
+
+ kfree(obj_request);
+}
+
+/*
+ * Caller is responsible for filling in:
+ * - the Linux request pointer (if there is one)
+ * - the request offset and length (zeroes used otherwise)
+ * - the object request array
+ */
+struct rbd_image_request *rbd_image_request_create(struct rbd_device *rbd_dev,
+ bool write_request, u32 obj_req_count)
+{
+ size_t size;
+ size_t obj_req_size;
+ struct rbd_image_request *image_request;
+ struct ceph_snap_context *snapc = NULL;
+
+ obj_req_size = obj_req_count * sizeof (image_request->obj_requests[0]);
+ size = sizeof (*image_request) + obj_req_size;
+ image_request = kmalloc(size, GFP_ATOMIC);
+ if (!image_request)
+ return NULL;
+
+ if (write_request) {
+ down_read(&rbd_dev->header_rwsem);
+ snapc = ceph_get_snap_context(rbd_dev->header.snapc);
+ up_read(&rbd_dev->header_rwsem);
+ if (WARN_ON(!snapc)) {
+ kfree(image_request);
+ return NULL; /* Shouldn't happen */
+ }
+ }
+
+ image_request->rq = NULL;
+ image_request->rbd_dev = rbd_dev;
+ if (write_request)
+ image_request->snapc = snapc;
+ else
+ image_request->snap_id = rbd_dev->spec->snap_id;
+ image_request->offset = 0;
+ image_request->length = 0;
+ spin_lock_init(&image_request->completion_lock);
+ image_request->next_completion = 0;
+ image_request->write_request = write_request;
+ image_request->obj_req_count = obj_req_count;
+ kref_init(&image_request->kref);
+ memset(&image_request->obj_requests, 0, obj_req_size);
+
+ return image_request;
+}
+
+static void rbd_image_request_destroy(struct kref *kref)
+{
+ struct rbd_image_request *image_request;
+ u32 which;
+
+ image_request = container_of(kref, struct rbd_image_request, kref);
+
+ for (which = 0; which < image_request->obj_req_count; which++)
+ rbd_obj_request_put(image_request->obj_requests[which]);
+
+ if (image_request->write_request)
+ ceph_put_snap_context(image_request->snapc);
+
+ kfree(image_request);
+}
+
+static int rbd_image_request_fill_bio(struct rbd_image_request *image_request,
+ struct bio *bio_list)
+{
+ struct rbd_device *rbd_dev = image_request->rbd_dev;
+ struct rbd_obj_request *obj_request = NULL;
+ unsigned int bio_offset;
+ u64 image_offset;
+ u64 resid;
+ u32 which;
+ u16 opcode;
+
+ opcode = image_request->write_request ? CEPH_OSD_OP_WRITE
+ : CEPH_OSD_OP_READ;
+ bio_offset = 0;
+ image_offset = image_request->offset;
+ rbd_assert(image_offset == bio_list->bi_sector << SECTOR_SHIFT);
+ which = 0;
+ resid = image_request->length;
+ while (resid) {
+ const char *object_name;
+ unsigned int clone_size;
+ struct ceph_osd_req_op *op;
+ u64 offset;
+ u64 length;
+
+ rbd_assert(which < image_request->obj_req_count);
+
+ object_name = rbd_segment_name(rbd_dev, image_offset);
+ if (!object_name)
+ goto out_unwind;
+ offset = rbd_segment_offset(rbd_dev, image_offset);
+ length = rbd_segment_length(rbd_dev, image_offset, resid);
+ obj_request = rbd_obj_request_create(object_name,
+ offset, length, obj_req_bio);
+ kfree(object_name); /* object request has its own copy */
+ if (!obj_request)
+ goto out_unwind;
+ obj_request->image_request = image_request;
+ obj_request->which = which;
+
+ rbd_assert(length <= (u64) UINT_MAX);
+ clone_size = (unsigned int) length;
+ obj_request->bio_list = bio_chain_clone_range(&bio_list,
+ &bio_offset, clone_size,
+ GFP_ATOMIC);
+ if (!obj_request->bio_list)
+ goto out_partial;
+
+ /*
+ * Build up the op to use in building the osd
+ * request. Note that the contents of the op are
+ * copied by rbd_osd_req_create().
+ */
+ op = rbd_osd_req_op_create(opcode, offset, length);
+ if (!op)
+ goto out_partial;
+ obj_request->osd_req = rbd_osd_req_create(rbd_dev,
+ image_request->write_request,
+ obj_request, op);
+ rbd_osd_req_op_destroy(op);
+ if (!obj_request->osd_req)
+ goto out_partial;
+ /* status and version are initially zero-filled */
+
+ image_request->obj_requests[which++] = obj_request;
+
+ image_offset += length;
+ resid -= length;
+ }
+
+ return 0;
+
+out_partial:
+ rbd_obj_request_put(obj_request);
+out_unwind:
+ while (which--) {
+ rbd_obj_request_put(image_request->obj_requests[which]);
+ image_request->obj_requests[which] = NULL;
+ }
+
+ return -ENOMEM;
+}
+
+static void rbd_image_obj_callback(struct rbd_obj_request *obj_request) +{
+ struct rbd_image_request *image_request = obj_request->image_request;
+ u32 which = obj_request->which;
+ bool done;
+
+ rbd_assert(image_request != NULL);
+ rbd_assert(obj_request == image_request->obj_requests[which]);
+
+ done = rbd_block_request_complete(image_request, which);
+ if (done)
+ rbd_image_request_put(image_request);
+}
+
+static int rbd_image_request_submit(struct rbd_image_request *image_request)
+{
+ struct rbd_device *rbd_dev = image_request->rbd_dev;
+ struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
+ u32 which;
+ int ret = 0;
+
+ for (which = 0; which < image_request->obj_req_count; which++) {
+ struct rbd_obj_request *obj_request;
+
+ obj_request = image_request->obj_requests[which];
+ obj_request->callback = rbd_image_obj_callback;
+ rbd_image_request_get(image_request);
+ ret = rbd_obj_request_submit(osdc, obj_request);
+ if (ret)
+ break;
+ }
+
+ return ret;
+}
+
+static void rbd_request_fn(struct request_queue *q)
+{
+ struct rbd_device *rbd_dev = q->queuedata;
+ bool read_only = rbd_dev->mapping.read_only;
+ struct request *rq;
+ int result;
+
+ while ((rq = blk_fetch_request(q))) {
+ bool write_request = rq_data_dir(rq) == WRITE;
+ struct rbd_image_request *image_request;
+ u64 offset;
+ u64 length;
+ int req_count;
+
+ /* Ignore any non-FS requests that filter through. */
+
+ if (rq->cmd_type != REQ_TYPE_FS) {
+ __blk_end_request_all(rq, 0);
+ continue;
+ }
+
+ spin_unlock_irq(q->queue_lock);
+
+ /* Disallow writes to a read-only device */
+
+ if (write_request) {
+ result = -EROFS;
+ if (read_only)
+ goto end_request;
+ rbd_assert(rbd_dev->spec->snap_id == CEPH_NOSNAP);
+ }
+
+ /* Quit early if the snapshot has disappeared */
+
+ if (!atomic_read(&rbd_dev->exists)) {
+ dout("request for non-existent snapshot");
+ rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
+ result = -ENXIO;
+ goto end_request;
+ }
+
+ /* Don't overrun max offset */
+
+ offset = (u64) blk_rq_pos(rq) << SECTOR_SHIFT;
+ length = (u64) blk_rq_bytes(rq);
+
+ result = -EINVAL;
+ if (WARN_ON(offset && length > U64_MAX - offset + 1))
+ goto end_request; /* Shouldn't happen */
+
+ req_count = rbd_get_num_segments(&rbd_dev->header,
+ offset, length);
+ result = -ENOMEM;
+ image_request = rbd_image_request_create(rbd_dev,
+ write_request,
+ (u32) req_count);
+ if (!image_request)
+ goto end_request;
+
+ image_request->rq = rq;
+ image_request->offset = offset;
+ image_request->length = length;
+
+ result = rbd_image_request_fill_bio(image_request, rq->bio);
+ if (!result)
+ result = rbd_image_request_submit(image_request);
+ if (result)
+ rbd_image_request_put(image_request);
+end_request:
+ spin_lock_irq(q->queue_lock);
+ if (result < 0)
+ __blk_end_request_all(rq, result);
+ }
+}
+
/*
* block device queue callback
*/
@@ -1929,8 +2521,8 @@ static int rbd_init_disk(struct rbd_device *rbd_dev)
disk->fops = &rbd_bd_ops;
disk->private_data = rbd_dev;

- /* init rq */
- q = blk_init_queue(rbd_rq_fn, &rbd_dev->lock);
+ (void) rbd_rq_fn; /* avoid a warning */
+ q = blk_init_queue(rbd_request_fn, &rbd_dev->lock);
if (!q)
goto out_disk;

1.7.9.5

Now that the request function has been replaced by one using the new request management data structures the old one can go away. Deleting it makes rbd_dev_do_request() no longer needed, and deleting that makes other functions unneeded, and so on.

---
 drivers/block/rbd.c |  297
---------------------------------------------------
 1 file changed, 297 deletions(-)

index 485fa70..0e1edc0 100644
@@ -609,18 +609,6 @@ static void rbd_put_client(struct rbd_client *rbdc)
 		kref_put(&rbdc->kref, rbd_client_release);
 }
 
-/*
- * Destroy requests collection
- */
-static void rbd_coll_release(struct kref *kref)
-{
-	struct rbd_req_coll *coll =
-		container_of(kref, struct rbd_req_coll, kref);
-
-	dout("rbd_coll_release %p\n", coll);
-	kfree(coll);
-}
-
 static bool rbd_image_format_valid(u32 image_format)
 {
 	return image_format == 1 || image_format == 2;

@@ -1182,52 +1170,6 @@ 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,
-				   s32 ret, u64 len)

-{
- struct request_queue *q;
- int min, max, i;
-
- dout("rbd_coll_end_req_index %p index %d ret %d len %llu\n",
- coll, index, (int)ret, (unsigned long long)len);
-
- if (!rq)
- return;
-
- if (!coll) {
- blk_end_request(rq, ret, len);
- return;
- }
-
- q = rq->q;
-
- spin_lock_irq(q->queue_lock);
- coll->status[index].done = 1;
- coll->status[index].rc = ret;
- coll->status[index].bytes = len;
- max = min = coll->num_done;
- while (max < coll->total && coll->status[max].done)
- max++;
-
- for (i = min; i<max; i++) {
- __blk_end_request(rq, (int)coll->status[i].rc,
- coll->status[i].bytes);
- coll->num_done++;
- kref_put(&coll->kref, rbd_coll_release);
- }
- spin_unlock_irq(q->queue_lock);
-}
-
-static void rbd_coll_end_req(struct rbd_request *rbd_req,
- s32 ret, u64 len)
-{
- rbd_coll_end_req_index(rbd_req->rq,
- rbd_req->coll, rbd_req->coll_index,
- ret, len);
-}
-
/*
* Send ceph osd request
*/
@@ -1327,46 +1269,6 @@ done_osd_req:
return ret;
}

-/*
- * Ceph osd op callback
- */
-static void rbd_req_cb(struct ceph_osd_request *osd_req, struct ceph_msg *msg)
-{
- struct rbd_request *rbd_req = osd_req->r_priv;
- struct ceph_osd_reply_head *replyhead;
- struct ceph_osd_op *op;
- s32 rc;
- u64 bytes;
- int read_op;
-
- /* parse reply */
- replyhead = msg->front.iov_base;
- WARN_ON(le32_to_cpu(replyhead->num_ops) == 0);
- op = (void *)(replyhead + 1);
- rc = (s32)le32_to_cpu(replyhead->result);
- bytes = le64_to_cpu(op->extent.length);
- read_op = (le16_to_cpu(op->op) == CEPH_OSD_OP_READ);
-
- dout("rbd_req_cb bytes=%llu readop=%d rc=%d\n",
- (unsigned long long) bytes, read_op, (int) rc);
-
- if (rc == (s32)-ENOENT && read_op) {
- zero_bio_chain(rbd_req->bio, 0);
- rc = 0;
- } else if (rc == 0 && read_op && bytes < rbd_req->len) {
- zero_bio_chain(rbd_req->bio, bytes);
- bytes = rbd_req->len;
- }
-
- rbd_coll_end_req(rbd_req, rc, bytes);
-
- if (rbd_req->bio)
- bio_chain_put(rbd_req->bio);
-
- ceph_osdc_put_request(osd_req);
- kfree(rbd_req);
-}
-
static void rbd_simple_req_cb(struct ceph_osd_request *osd_req,
struct ceph_msg *msg)
{
@@ -1414,70 +1316,6 @@ done:
return ret;
}

-/*
- * Do an asynchronous ceph osd operation
- */
-static int rbd_do_op(struct request *rq,

-		     struct rbd_device *rbd_dev,
-		     struct ceph_snap_context *snapc,
-		     u64 ofs, u64 len,
-		     struct bio *bio,
-		     struct rbd_req_coll *coll,
-		     int coll_index)

-{
- const char *seg_name;
- u64 seg_ofs;
- u64 seg_len;
- int ret;
- struct ceph_osd_req_op *op;
- int opcode;
- int flags;
- u64 snapid;
-
- seg_name = rbd_segment_name(rbd_dev, ofs);
- if (!seg_name)
- return -ENOMEM;
- seg_len = rbd_segment_length(rbd_dev, ofs, len);
- seg_ofs = rbd_segment_offset(rbd_dev, ofs);
-
- if (rq_data_dir(rq) == WRITE) {
- opcode = CEPH_OSD_OP_WRITE;
- flags = CEPH_OSD_FLAG_WRITE|CEPH_OSD_FLAG_ONDISK;
- snapid = CEPH_NOSNAP;
- } else {
- opcode = CEPH_OSD_OP_READ;
- flags = CEPH_OSD_FLAG_READ;
- rbd_assert(!snapc);
- snapid = rbd_dev->spec->snap_id;
- }
-
- ret = -ENOMEM;
- op = rbd_osd_req_op_create(opcode, seg_ofs, seg_len);
- if (!op)
- goto done;
-
- /* we've taken care of segment sizes earlier when we
- cloned the bios. We should never have a segment
- truncated at this point */
- rbd_assert(seg_len == len);
-
- ret = rbd_do_request(rq, rbd_dev, snapc, snapid,

-			     seg_name, seg_ofs, seg_len,
-			     bio,
-			     NULL, 0,
-			     flags,
-			     op,
-			     coll, coll_index,
-			     rbd_req_cb, NULL);

- if (ret < 0)
- rbd_coll_end_req_index(rq, coll, coll_index,
- (s32)ret, seg_len);
- rbd_osd_req_op_destroy(op);
-done:
- kfree(seg_name);
- return ret;
-}
-
static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
struct rbd_obj_request *obj_request)
{
@@ -1642,78 +1480,6 @@ static int rbd_req_sync_exec(struct rbd_device *rbd_dev,
return ret;
}

-static struct rbd_req_coll *rbd_alloc_coll(int num_reqs) -{
- struct rbd_req_coll *coll =
- kzalloc(sizeof(struct rbd_req_coll) +
- sizeof(struct rbd_req_status) * num_reqs,
- GFP_ATOMIC);
-
- if (!coll)
- return NULL;
- coll->total = num_reqs;
- kref_init(&coll->kref);
- return coll;
-}
-
-static int rbd_dev_do_request(struct request *rq,
- struct rbd_device *rbd_dev,
- struct ceph_snap_context *snapc,
- u64 ofs, unsigned int size,
- struct bio *bio_chain)
-{
- int num_segs;
- struct rbd_req_coll *coll;
- unsigned int bio_offset;
- int cur_seg = 0;
-
- dout("%s 0x%x bytes at 0x%llx\n",
- rq_data_dir(rq) == WRITE ? "write" : "read",
- size, (unsigned long long) blk_rq_pos(rq) * SECTOR_SIZE);
-
- num_segs = rbd_get_num_segments(&rbd_dev->header, ofs, size);
- if (num_segs <= 0)
- return num_segs;
-
- coll = rbd_alloc_coll(num_segs);
- if (!coll)
- return -ENOMEM;
-
- bio_offset = 0;
- do {
- u64 limit = rbd_segment_length(rbd_dev, ofs, size);
- unsigned int clone_size;
- struct bio *bio_clone;
-
- BUG_ON(limit > (u64)UINT_MAX);
- clone_size = (unsigned int)limit;
- dout("bio_chain->bi_vcnt=%hu\n", bio_chain->bi_vcnt);
-
- kref_get(&coll->kref);
-
- /* Pass a cloned bio chain via an osd request */
-
- bio_clone = bio_chain_clone_range(&bio_chain,
- &bio_offset, clone_size,
- GFP_ATOMIC);
- if (bio_clone)
- (void)rbd_do_op(rq, rbd_dev, snapc,
- ofs, clone_size,
- bio_clone, coll, cur_seg);
- else
- rbd_coll_end_req_index(rq, coll, cur_seg,
- (s32)-ENOMEM,
- clone_size);
- size -= clone_size;
- ofs += clone_size;
-
- cur_seg++;
- } while (size > 0);
- kref_put(&coll->kref, rbd_coll_release);
-
- return 0;
-}
-
/* Returns true if this call completed the last object request */

static bool rbd_block_request_complete(struct rbd_image_request
*image_request,
@@ -2211,68 +1977,6 @@ end_request:
}

/*
- * block device queue callback
- */
-static void rbd_rq_fn(struct request_queue *q)
-{
- struct rbd_device *rbd_dev = q->queuedata;
- bool read_only = rbd_dev->mapping.read_only;
- struct request *rq;
-
- while ((rq = blk_fetch_request(q))) {
- struct ceph_snap_context *snapc = NULL;
- unsigned int size = 0;
- int result;
-
- dout("fetched request\n");
-
- /* Filter out block requests we don't understand */
-
- if ((rq->cmd_type != REQ_TYPE_FS)) {
- __blk_end_request_all(rq, 0);
- continue;
- }
- spin_unlock_irq(q->queue_lock);
-
- /* Write requests need a reference to the snapshot context */
-
- if (rq_data_dir(rq) == WRITE) {
- result = -EROFS;
- if (read_only) /* Can't write to a read-only device */
- goto out_end_request;
-
- /*
- * Note that each osd request will take its
- * own reference to the snapshot context
- * supplied. The reference we take here
- * just guarantees the one we provide stays
- * valid.
- */
- down_read(&rbd_dev->header_rwsem);
- snapc = ceph_get_snap_context(rbd_dev->header.snapc);
- up_read(&rbd_dev->header_rwsem);
- rbd_assert(snapc != NULL);
- } else if (!atomic_read(&rbd_dev->exists)) {
- rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
- dout("request for non-existent snapshot");
- result = -ENXIO;
- goto out_end_request;
- }
-
- size = blk_rq_bytes(rq);
- result = rbd_dev_do_request(rq, rbd_dev, snapc,
- blk_rq_pos(rq) * SECTOR_SIZE,
- size, rq->bio);
-out_end_request:
- if (snapc)
- ceph_put_snap_context(snapc);
- spin_lock_irq(q->queue_lock);
- if (!size || result < 0)
- __blk_end_request_all(rq, result);
- }
-}
-
-/*
* a queue callback. Makes sure that we don't create a bio that spans
across
* multiple osd objects. One exception would be with a single page bios,
* which we handle later at bio_chain_clone_range()
@@ -2521,7 +2225,6 @@ static int rbd_init_disk(struct rbd_device *rbd_dev)
disk->fops = &rbd_bd_ops;
disk->private_data = rbd_dev;

- (void) rbd_rq_fn; /* avoid a warning */
q = blk_init_queue(rbd_request_fn, &rbd_dev->lock); if (!q)
goto out_disk;

1.7.9.5

The two remaining callers of rbd_do_request() always pass a null collection pointer, so the "coll" and "coll_index" parameters are not needed. There is no other use of that data structure, so it can be eliminated.

Deleting them means there is no need to allocate a rbd_request structure for the callback function. And since that's the only use of *that* structure, it too can be eliminated.

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

index 0e1edc0..c1b99c7 100644
@@ -162,25 +162,6 @@ struct rbd_client {
 	struct list_head	node;
 };
 
-/*
- * a request completion status
- */
-struct rbd_req_status {
-	int done;
-	s32 rc;
-	u64 bytes;
-};
-
-/*
- * a collection of requests
- */
-struct rbd_req_coll {
-	int			total;
-	int			num_done;
-	struct kref		kref;
-	struct rbd_req_status	status[0];
-};
-
 struct rbd_image_request;
 
 enum obj_req_type { obj_req_bio };	/* More types to come */

@@ -230,18 +211,6 @@ struct rbd_image_request {
 	struct rbd_obj_request	*obj_requests[0];
 };
 
-/*
- * a single io request
- */
-struct rbd_request {
-	struct request		*rq;		/* blk layer request */
-	struct bio		*bio;		/* cloned bio */
-	struct page		**pages;	/* list of used pages */
-	u64			len;
-	int			coll_index;
-	struct rbd_req_coll	*coll;
-};
-
 struct rbd_snap {
 	struct	device		dev;
 	const char		*name;

@@ -1183,21 +1152,18 @@ static int rbd_do_request(struct request *rq,
 			  int num_pages,
 			  int flags,
 			  struct ceph_osd_req_op *op,
-			  struct rbd_req_coll *coll,
-			  int coll_index,
 			  void (*rbd_cb)(struct ceph_osd_request *,
 					 struct ceph_msg *),
 			  u64 *ver)
 {
 	struct ceph_osd_client *osdc;
 	struct ceph_osd_request *osd_req;
-	struct rbd_request *rbd_req = NULL;
 	struct timespec mtime = CURRENT_TIME;
 	int ret;
 
-	dout("rbd_do_request object_name=%s ofs=%llu len=%llu coll=%p[%d]\n",
+	dout("rbd_do_request object_name=%s ofs=%llu len=%llu\n",
 		object_name, (unsigned long long) ofs,
-		(unsigned long long) len, coll, coll_index);
+		(unsigned long long) len);
 
 	osdc = &rbd_dev->rbd_client->client->osdc;
 	osd_req = ceph_osdc_alloc_request(osdc, snapc, 1, false, GFP_NOIO);

@@ -1211,22 +1177,8 @@ static int rbd_do_request(struct request *rq,
 		bio_get(osd_req->r_bio);
 	}
 
-	if (coll) {
-		ret = -ENOMEM;
-		rbd_req = kmalloc(sizeof(*rbd_req), GFP_NOIO);
-		if (!rbd_req)
-			goto done_osd_req;
-
-		rbd_req->rq = rq;
-		rbd_req->bio = bio;
-		rbd_req->pages = pages;
-		rbd_req->len = len;
-		rbd_req->coll = coll;
-		rbd_req->coll_index = coll_index;
-	}
-
 	osd_req->r_callback = rbd_cb;
-	osd_req->r_priv = rbd_req;
+	osd_req->r_priv = NULL;
 
 	strncpy(osd_req->r_oid, object_name, sizeof(osd_req->r_oid));
 	osd_req->r_oid_len = strlen(osd_req->r_oid);

@@ -1262,8 +1214,6 @@ static int rbd_do_request(struct request *rq,
 done_err:
 	if (bio)
 		bio_chain_put(osd_req->r_bio);
-	kfree(rbd_req);
-done_osd_req:
 	ceph_osdc_put_request(osd_req);
 
 	return ret;

@@ -1302,7 +1252,6 @@ static int rbd_req_sync_op(struct rbd_device *rbd_dev,
 			  pages, num_pages,
 			  flags,
 			  op,
-			  NULL, 0,
 			  NULL,
 			  ver);
 	if (ret < 0)

@@ -1371,7 +1320,6 @@ static int rbd_req_sync_notify_ack(struct

rbd_device *rbd_dev,
NULL, 0, CEPH_OSD_FLAG_READ, op,
- NULL, 0,
rbd_simple_req_cb, NULL);

rbd_osd_req_op_destroy(op);

1.7.9.5

Reimplement the synchronous read operation used for reading a version 1 header using the new request tracking code. Name the resulting function rbd_obj_read_sync() to better reflect that it's a full object operation, not an object request. To do this, implement a new obj_req_pages object request type.

This implements a new mechanism to allow the caller to wait for completion for an rbd_obj_request by calling rbd_obj_request_wait().

This partially resolves:

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

---
 drivers/block/rbd.c |   96
++++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 92 insertions(+), 4 deletions(-)

index c1b99c7..5a8fef4 100644
@@ -164,7 +164,7 @@ struct rbd_client {
 
 struct rbd_image_request;
 
-enum obj_req_type { obj_req_bio };	/* More types to come */
+enum obj_req_type { obj_req_bio, obj_req_pages };
 
 struct rbd_obj_request;
 typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);

@@ -178,7 +178,13 @@ struct rbd_obj_request {
 	u32			which;		/* posn in image req array */
 
 	enum obj_req_type	type;
-	struct bio		*bio_list;
+	union {
+		struct bio	*bio_list;
+		struct {
+			struct page	**pages;
+			u32		page_count;
+		};
+	};
 
 	struct ceph_osd_request	*osd_req;
 

@@ -188,6 +194,7 @@ struct rbd_obj_request {
 	atomic_t		done;
 
 	rbd_obj_callback_t	callback;
+	struct completion	completion;
 
 	struct kref		kref;
 };

@@ -1065,6 +1072,7 @@ static bool obj_req_type_valid(enum obj_req_type type)
 {
 	switch (type) {
 	case obj_req_bio:
+	case obj_req_pages:
 		return true;
 	default:
 		return false;

@@ -1278,8 +1286,15 @@ static int rbd_obj_request_submit(struct

ceph_osd_client *osdc,
return ret;
}

+/* Caller is responsible for rbd_obj_request_destroy(obj_request) */ +
+static int rbd_obj_request_wait(struct rbd_obj_request *obj_request) +{
+ return wait_for_completion_interruptible(&obj_request->completion);
+}
+
/*
- * Request sync osd read
+ * Synchronously read a range from an object into a provided buffer */
static int rbd_req_sync_read(struct rbd_device *rbd_dev,
const char *object_name,
@@ -1503,6 +1518,8 @@ static void rbd_obj_request_complete(struct rbd_obj_request *obj_request)
{
if (obj_request->callback)
obj_request->callback(obj_request);
+ else
+ complete_all(&obj_request->completion);
}

static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
@@ -1583,6 +1600,11 @@ static struct ceph_osd_request *rbd_osd_req_create(
/* osd client requires "num pages" even for bio */ osd_req->r_num_pages = calc_pages_for(offset, length); break;
+ case obj_req_pages:
+ osd_req->r_pages = obj_request->pages;
+ osd_req->r_num_pages = obj_request->page_count;
+ osd_req->r_page_alignment = offset & ~PAGE_MASK;
+ break;
}

if (write_request) {
@@ -1643,6 +1665,7 @@ static struct rbd_obj_request *rbd_obj_request_create(const char *object_name,
obj_request->length = length; obj_request->type = type; atomic_set(&obj_request->done, 0);
+ init_completion(&obj_request->completion);
kref_init(&obj_request->kref);

return obj_request;
@@ -1662,6 +1685,11 @@ static void rbd_obj_request_destroy(struct kref *kref)
if (obj_request->bio_list)
bio_chain_put(obj_request->bio_list);
break;
+ case obj_req_pages:
+ if (obj_request->pages)
+ ceph_release_page_vector(obj_request->pages,
+ obj_request->page_count);
+ break;
}

kfree(obj_request);
@@ -1984,6 +2012,65 @@ static void rbd_free_disk(struct rbd_device *rbd_dev)
put_disk(disk);
}

+static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
+ const char *object_name,
+ u64 offset, u64 length,
+ char *buf, u64 *version)
+
+{
+ struct ceph_osd_req_op *op;
+ struct rbd_obj_request *obj_request;
+ struct ceph_osd_client *osdc;
+ struct page **pages = NULL;
+ u32 page_count;
+ int ret;
+
+ page_count = (u32) calc_pages_for(offset, length);
+ pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
+ if (IS_ERR(pages))
+ ret = PTR_ERR(pages);
+
+ ret = -ENOMEM;
+ obj_request = rbd_obj_request_create(object_name, offset, length,
+ obj_req_pages);
+ if (!obj_request)
+ goto out_err;
+
+ obj_request->pages = pages;
+ obj_request->page_count = page_count;
+
+ op = rbd_osd_req_op_create(CEPH_OSD_OP_READ, offset, length);
+ if (!op)
+ goto out_err;
+ obj_request->osd_req = rbd_osd_req_create(rbd_dev, false,
+ obj_request, op);
+ rbd_osd_req_op_destroy(op);
+ if (!obj_request->osd_req)
+ goto out_err;
+
+ osdc = &rbd_dev->rbd_client->client->osdc;
+ ret = rbd_obj_request_submit(osdc, obj_request);
+ if (ret)
+ goto out_err;
+ ret = rbd_obj_request_wait(obj_request);
+ if (ret)
+ goto out_err;
+
+ ret = obj_request->result;
+ if (ret < 0)
+ goto out_err;
+ ret = ceph_copy_from_page_vector(pages, buf, 0, obj_request->xferred);
+ if (version)
+ *version = obj_request->version;
+out_err:
+ if (obj_request)
+ rbd_obj_request_put(obj_request);
+ else
+ ceph_release_page_vector(pages, page_count);
+
+ return ret;
+}
+
/*
* Read the complete header for the given rbd device.
*
@@ -2022,7 +2109,8 @@ rbd_dev_v1_header_read(struct rbd_device *rbd_dev, u64 *version)
if (!ondisk)
return ERR_PTR(-ENOMEM);

- ret = rbd_req_sync_read(rbd_dev, rbd_dev->header_name,
+ (void) rbd_req_sync_read; /* avoid a warning */
+ ret = rbd_obj_read_sync(rbd_dev, rbd_dev->header_name,
0, size,
(char *) ondisk, version);

1.7.9.5

Delete rbd_req_sync_read() is no longer used, so get rid of it.

---
 drivers/block/rbd.c |   24 ------------------------
 1 file changed, 24 deletions(-)

index 5a8fef4..6193c69 100644
@@ -1294,29 +1294,6 @@ static int rbd_obj_request_wait(struct

rbd_obj_request *obj_request)
}

/*
- * Synchronously read a range from an object into a provided buffer
- */
-static int rbd_req_sync_read(struct rbd_device *rbd_dev,

-			  const char *object_name,
-			  u64 ofs, u64 len,
-			  char *buf,
-			  u64 *ver)

-{
- struct ceph_osd_req_op *op;
- int ret;
-
- 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, ver);
- rbd_osd_req_op_destroy(op);
-
- return ret;
-}
-
-/*
* Request sync osd watch
*/
static int rbd_req_sync_notify_ack(struct rbd_device *rbd_dev,
@@ -2109,7 +2086,6 @@ rbd_dev_v1_header_read(struct rbd_device *rbd_dev, u64 *version)
if (!ondisk)
return ERR_PTR(-ENOMEM);

- (void) rbd_req_sync_read; /* avoid a warning */
ret = rbd_obj_read_sync(rbd_dev, rbd_dev->header_name,
0, size,
(char *) ondisk, version);

1.7.9.5

Implement a new function to set up or tear down a watch event for an mapped rbd image header using the new request code.

Create a new object request type "nodata" to handle this. And define rbd_osd_trivial_callback() which simply marks a request done.

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

index 6193c69..3c110b3 100644
@@ -164,7 +164,7 @@ struct rbd_client {
 
 struct rbd_image_request;
 
-enum obj_req_type { obj_req_bio, obj_req_pages };
+enum obj_req_type { obj_req_nodata, obj_req_bio, obj_req_pages };
 
 struct rbd_obj_request;
 typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);

@@ -1071,6 +1071,7 @@ static void rbd_image_request_put(struct

rbd_image_request *image_request)
static bool obj_req_type_valid(enum obj_req_type type) {
switch (type) {
+ case obj_req_nodata:
case obj_req_bio:
case obj_req_pages:
return true;
@@ -1491,6 +1492,12 @@ static void rbd_osd_write_callback(struct rbd_obj_request *obj_request,
atomic_set(&obj_request->done, 1);
}

+static void rbd_osd_trivial_callback(struct rbd_obj_request *obj_request,
+ struct ceph_osd_op *op)
+{
+ atomic_set(&obj_request->done, 1);
+}
+
static void rbd_obj_request_complete(struct rbd_obj_request *obj_request) {
if (obj_request->callback)
@@ -1527,6 +1534,9 @@ static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
case CEPH_OSD_OP_WRITE:
rbd_osd_write_callback(obj_request, op);
break;
+ case CEPH_OSD_OP_WATCH:
+ rbd_osd_trivial_callback(obj_request, op);
+ break;
default:
rbd_warn(NULL, "%s: unsupported op %hu\n",
obj_request->object_name, (unsigned short) opcode);
@@ -1570,6 +1580,8 @@ static struct ceph_osd_request *rbd_osd_req_create(

rbd_assert(obj_req_type_valid(obj_request->type)); switch (obj_request->type) {
+ case obj_req_nodata:
+ break; /* Nothing to do */
case obj_req_bio:
rbd_assert(obj_request->bio_list != NULL); osd_req->r_bio = obj_request->bio_list;
@@ -1658,6 +1670,8 @@ static void rbd_obj_request_destroy(struct kref *kref)

rbd_assert(obj_req_type_valid(obj_request->type)); switch (obj_request->type) {
+ case obj_req_nodata:
+ break; /* Nothing to do */
case obj_req_bio:
if (obj_request->bio_list)
bio_chain_put(obj_request->bio_list);
@@ -1854,6 +1868,72 @@ static int rbd_image_request_submit(struct rbd_image_request *image_request)
return ret;
}

+/*
+ * Request sync osd watch/unwatch. The value of "start" determines
+ * whether a watch request is being initiated or torn down.
+ */
+static int rbd_dev_header_watch_sync(struct rbd_device *rbd_dev, int start) +{
+ struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
+ struct rbd_obj_request *obj_request;
+ struct ceph_osd_req_op *op;
+ int ret;
+
+ rbd_assert(start ^ !!rbd_dev->watch_event);
+ rbd_assert(start ^ !!rbd_dev->watch_request);
+
+ if (start) {
+ ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0, rbd_dev,
+ &rbd_dev->watch_event);
+ if (ret < 0)
+ return ret;
+ }
+
+ ret = -ENOMEM;
+ obj_request = rbd_obj_request_create(rbd_dev->header_name,
+ 0, 0, obj_req_nodata);
+ if (!obj_request)
+ goto out_cancel;
+
+ op = rbd_osd_req_op_create(CEPH_OSD_OP_WATCH,
+ rbd_dev->watch_event->cookie,
+ rbd_dev->header.obj_version, start);
+ if (!op)
+ goto out_cancel;
+ obj_request->osd_req = rbd_osd_req_create(rbd_dev, true,
+ obj_request, op);
+ rbd_osd_req_op_destroy(op);
+ if (!obj_request->osd_req)
+ goto out_cancel;
+
+ if (start) {
+ rbd_dev->watch_request = obj_request->osd_req;
+ ceph_osdc_set_request_linger(osdc, rbd_dev->watch_request);
+ }
+ ret = rbd_obj_request_submit(osdc, obj_request);
+ if (ret)
+ goto out_cancel;
+ ret = rbd_obj_request_wait(obj_request);
+ if (ret)
+ goto out_cancel;
+
+ ret = obj_request->result;
+ if (ret)
+ goto out_cancel;
+
+ if (start)
+ goto done; /* Done if setting up the watch request */
+out_cancel:
+ /* Cancel the event if we're tearing down, or on error */
+ ceph_osdc_cancel_event(rbd_dev->watch_event);
+ rbd_dev->watch_event = NULL;
+done:
+ if (obj_request)
+ rbd_obj_request_put(obj_request);
+
+ return ret;
+}
+
static void rbd_request_fn(struct request_queue *q)
{
struct rbd_device *rbd_dev = q->queuedata;
@@ -3876,7 +3956,8 @@ static int rbd_dev_probe_finish(struct rbd_device *rbd_dev)
if (ret)
goto err_out_bus;

- ret = rbd_req_sync_watch(rbd_dev, 1);
+ (void) rbd_req_sync_watch; /* avoid a warning */
+ ret = rbd_dev_header_watch_sync(rbd_dev, 1);
if (ret)
goto err_out_bus;

@@ -4039,7 +4120,7 @@ static void rbd_dev_release(struct device *dev)
rbd_dev->watch_request);
}
if (rbd_dev->watch_event)
- rbd_req_sync_watch(rbd_dev, 0);
+ rbd_dev_header_watch_sync(rbd_dev, 0);

/* clean up and free blkdev */
rbd_free_disk(rbd_dev);

1.7.9.5

Get rid of rbd_req_sync_watch(), because it is no longer used.

---
 drivers/block/rbd.c |   42 ------------------------------------------
 1 file changed, 42 deletions(-)

index 3c110b3..7dedd18 100644
@@ -1340,47 +1340,6 @@ static void rbd_watch_cb(u64 ver, u64 notify_id,

u8 opcode, void *data)
rbd_req_sync_notify_ack(rbd_dev, hver, notify_id);
}

-/*
- * Request sync osd watch/unwatch. The value of "start" determines
- * whether a watch request is being initiated or torn down.
- */
-static int rbd_req_sync_watch(struct rbd_device *rbd_dev, int start) -{
- struct ceph_osd_req_op *op;
- int ret = 0;
-
- rbd_assert(start ^ !!rbd_dev->watch_event);
- rbd_assert(start ^ !!rbd_dev->watch_request);
-
- if (start) {
- struct ceph_osd_client *osdc;
-
- osdc = &rbd_dev->rbd_client->client->osdc;
- ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0, rbd_dev,
- &rbd_dev->watch_event);
- if (ret < 0)
- return ret;
- }
-
- 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, NULL);

-
- /* 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;
- }
- rbd_osd_req_op_destroy(op);
-
- return ret;
-}

/*
* Synchronous osd object method call
@@ -3956,7 +3915,6 @@ static int rbd_dev_probe_finish(struct rbd_device *rbd_dev)
if (ret)
goto err_out_bus;

- (void) rbd_req_sync_watch; /* avoid a warning */
ret = rbd_dev_header_watch_sync(rbd_dev, 1); if (ret)
goto err_out_bus;

1.7.9.5

Use the new object request tracking mechanism for handling a notify_ack request.

Move the callback function below the definition of this so we don't have to do a pre-declaration.

This resolves:

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

---
 drivers/block/rbd.c |   76
+++++++++++++++++++++++++++++++++++++--------------
 1 file changed, 55 insertions(+), 21 deletions(-)

index 7dedd18..08465ae 100644
@@ -1320,27 +1320,6 @@ static int rbd_req_sync_notify_ack(struct

rbd_device *rbd_dev,
return ret;
}

-static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data) -{
- struct rbd_device *rbd_dev = (struct rbd_device *)data;
- u64 hver;
- int rc;
-
- if (!rbd_dev)
- return;
-
- dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
- rbd_dev->header_name, (unsigned long long) notify_id,
- (unsigned int) opcode);
- rc = rbd_dev_refresh(rbd_dev, &hver);
- if (rc)
- rbd_warn(rbd_dev, "got notification but failed to "
- " update snaps: %d\n", rc);
-
- rbd_req_sync_notify_ack(rbd_dev, hver, notify_id);
-}
-
-
/*
* Synchronous osd object method call
*/
@@ -1493,6 +1472,7 @@ static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
case CEPH_OSD_OP_WRITE:
rbd_osd_write_callback(obj_request, op); break;
+ case CEPH_OSD_OP_NOTIFY_ACK:
case CEPH_OSD_OP_WATCH:
rbd_osd_trivial_callback(obj_request, op); break;
@@ -1827,6 +1807,60 @@ static int rbd_image_request_submit(struct rbd_image_request *image_request)
return ret;
}

+static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev,
+ u64 ver, u64 notify_id)
+{
+ struct rbd_obj_request *obj_request;
+ struct ceph_osd_req_op *op;
+ struct ceph_osd_client *osdc;
+ int ret;
+
+ obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0,
+ obj_req_nodata);
+ if (!obj_request)
+ return -ENOMEM;
+
+ ret = -ENOMEM;
+ op = rbd_osd_req_op_create(CEPH_OSD_OP_NOTIFY_ACK, notify_id, ver);
+ if (!op)
+ goto out_err;
+ obj_request->osd_req = rbd_osd_req_create(rbd_dev, false,
+ obj_request, op);
+ rbd_osd_req_op_destroy(op);
+ if (!obj_request->osd_req)
+ goto out_err;
+
+ osdc = &rbd_dev->rbd_client->client->osdc;
+ ret = rbd_obj_request_submit(osdc, obj_request);
+ if (!ret)
+ ret = rbd_obj_request_wait(obj_request);
+out_err:
+ rbd_obj_request_put(obj_request);
+
+ return ret;
+}
+
+static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data) +{
+ struct rbd_device *rbd_dev = (struct rbd_device *)data;
+ u64 hver;
+ int rc;
+
+ if (!rbd_dev)
+ return;
+
+ dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
+ rbd_dev->header_name, (unsigned long long) notify_id,
+ (unsigned int) opcode);
+ rc = rbd_dev_refresh(rbd_dev, &hver);
+ if (rc)
+ rbd_warn(rbd_dev, "got notification but failed to "
+ " update snaps: %d\n", rc);
+
+ (void) rbd_req_sync_notify_ack; /* avoid a warning */
+ rbd_obj_notify_ack_sync(rbd_dev, hver, notify_id);
+}
+
/*
* Request sync osd watch/unwatch. The value of "start" determines
* whether a watch request is being initiated or torn down.

1.7.9.5

Get rid rbd_req_sync_notify_ack() because it is no longer used. As a result rbd_simple_req_cb() becomes unreferenced, so get rid of that too.

---
 drivers/block/rbd.c |   33 ---------------------------------
 1 file changed, 33 deletions(-)

index 08465ae..a6ca5db 100644
@@ -1228,12 +1228,6 @@ done_err:
 	return ret;
 }
 
-static void rbd_simple_req_cb(struct ceph_osd_request *osd_req,
-				struct ceph_msg *msg)
-{
-	ceph_osdc_put_request(osd_req);
-}
-
 /*
  * Do a synchronous ceph osd operation
  */

@@ -1295,32 +1289,6 @@ static int rbd_obj_request_wait(struct

rbd_obj_request *obj_request)
}

/*
- * Request sync osd watch
- */
-static int rbd_req_sync_notify_ack(struct rbd_device *rbd_dev,
- u64 ver,
- u64 notify_id)
-{
- struct ceph_osd_req_op *op;
- int ret;
-
- op = rbd_osd_req_op_create(CEPH_OSD_OP_NOTIFY_ACK, notify_id, ver);
- if (!op)
- return -ENOMEM;
-
- ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,

-			  rbd_dev->header_name, 0, 0, NULL,
-			  NULL, 0,
-			  CEPH_OSD_FLAG_READ,
-			  op,
-			  rbd_simple_req_cb, NULL);

-
- rbd_osd_req_op_destroy(op);
-
- return ret;
-}
-
-/*
* Synchronous osd object method call
*/
static int rbd_req_sync_exec(struct rbd_device *rbd_dev,
@@ -1857,7 +1825,6 @@ static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
rbd_warn(rbd_dev, "got notification but failed to "
" update snaps: %d\n", rc);

- (void) rbd_req_sync_notify_ack; /* avoid a warning */
rbd_obj_notify_ack_sync(rbd_dev, hver, notify_id);
}

1.7.9.5

When we receive notification of a change to an rbd image's header object we need to refresh our information about the image (its size and snapshot context). Once we have refreshed our rbd image we need to acknowledge the notification.

This acknowledgement was previously done synchronously, but there's really no need to wait for it to complete.

Change it so the caller doesn't wait for the notify acknowledgement request to complete. And change the name to reflect it's no longer synchronous.

This resolves:

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

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

index a6ca5db..e66695a 100644
@@ -1775,7 +1775,7 @@ static int rbd_image_request_submit(struct

rbd_image_request *image_request)
return ret;
}

-static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev, +static int rbd_obj_notify_ack(struct rbd_device *rbd_dev,
u64 ver, u64 notify_id)
{
struct rbd_obj_request *obj_request;
@@ -1799,11 +1799,11 @@ static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev,
goto out_err;

osdc = &rbd_dev->rbd_client->client->osdc;
+ obj_request->callback = rbd_obj_request_put;
ret = rbd_obj_request_submit(osdc, obj_request);
- if (!ret)
- ret = rbd_obj_request_wait(obj_request);
out_err:
- rbd_obj_request_put(obj_request);
+ if (ret)
+ rbd_obj_request_put(obj_request);

return ret;
}
@@ -1825,7 +1825,7 @@ static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
rbd_warn(rbd_dev, "got notification but failed to "
" update snaps: %d\n", rc);

- rbd_obj_notify_ack_sync(rbd_dev, hver, notify_id);
+ rbd_obj_notify_ack(rbd_dev, hver, notify_id);
}

/*

1.7.9.5

When we receive notification of a change to an rbd image's header object we need to refresh our information about the image (its size and snapshot context). Once we have refreshed our rbd image we need to acknowledge the notification.

This acknowledgement was previously done synchronously, but there's really no need to wait for it to complete.

Change it so the caller doesn't wait for the notify acknowledgement request to complete. And change the name to reflect it's no longer synchronous.

This resolves:

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

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

index a6ca5db..e66695a 100644
@@ -1775,7 +1775,7 @@ static int rbd_image_request_submit(struct

rbd_image_request *image_request)
return ret;
}

-static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev, +static int rbd_obj_notify_ack(struct rbd_device *rbd_dev,
u64 ver, u64 notify_id)
{
struct rbd_obj_request *obj_request;
@@ -1799,11 +1799,11 @@ static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev,
goto out_err;

osdc = &rbd_dev->rbd_client->client->osdc;
+ obj_request->callback = rbd_obj_request_put;
ret = rbd_obj_request_submit(osdc, obj_request);
- if (!ret)
- ret = rbd_obj_request_wait(obj_request);
out_err:
- rbd_obj_request_put(obj_request);
+ if (ret)
+ rbd_obj_request_put(obj_request);

return ret;
}
@@ -1825,7 +1825,7 @@ static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
rbd_warn(rbd_dev, "got notification but failed to "
" update snaps: %d\n", rc);

- rbd_obj_notify_ack_sync(rbd_dev, hver, notify_id);
+ rbd_obj_notify_ack(rbd_dev, hver, notify_id);
}

/*

1.7.9.5

Get rid rbd_req_sync_exec() because it is no longer used. That eliminates the last use of rbd_req_sync_op(), so get rid of that too. And finally, that leaves rbd_do_request() unreferenced, so get rid of that.

---
 drivers/block/rbd.c |  160
---------------------------------------------------
 1 file changed, 160 deletions(-)

index 934bb61..b655088 100644
@@ -1148,126 +1148,6 @@ static void rbd_osd_req_op_destroy(struct

ceph_osd_req_op *op)
kfree(op);
}

-/*
- * Send ceph osd request
- */
-static int rbd_do_request(struct request *rq,

-			  struct rbd_device *rbd_dev,
-			  struct ceph_snap_context *snapc,
-			  u64 snapid,
-			  const char *object_name, u64 ofs, u64 len,
-			  struct bio *bio,
-			  struct page **pages,
-			  int num_pages,
-			  int flags,
-			  struct ceph_osd_req_op *op,
-			  void (*rbd_cb)(struct ceph_osd_request *,

- struct ceph_msg *),
- u64 *ver)
-{
- struct ceph_osd_client *osdc;
- struct ceph_osd_request *osd_req;
- struct timespec mtime = CURRENT_TIME;
- int ret;
-
- dout("rbd_do_request object_name=%s ofs=%llu len=%llu\n",
- object_name, (unsigned long long) ofs,
- (unsigned long long) len);
-
- osdc = &rbd_dev->rbd_client->client->osdc;
- osd_req = ceph_osdc_alloc_request(osdc, snapc, 1, false, GFP_NOIO);
- if (!osd_req)
- return -ENOMEM;
-
- osd_req->r_flags = flags;
- osd_req->r_pages = pages;
- if (bio) {
- osd_req->r_bio = bio;
- bio_get(osd_req->r_bio);
- }
-
- osd_req->r_callback = rbd_cb;
- osd_req->r_priv = NULL;
-
- strncpy(osd_req->r_oid, object_name, sizeof(osd_req->r_oid));
- osd_req->r_oid_len = strlen(osd_req->r_oid);
-
- osd_req->r_file_layout = rbd_dev->layout; /* struct */
- osd_req->r_num_pages = calc_pages_for(ofs, len);
- osd_req->r_page_alignment = ofs & ~PAGE_MASK;
-
- ceph_osdc_build_request(osd_req, ofs, len, 1, op,
- snapc, snapid, &mtime);
-
- if (op->op == CEPH_OSD_OP_WATCH && op->watch.flag) {
- ceph_osdc_set_request_linger(osdc, osd_req);
- rbd_dev->watch_request = osd_req;
- }
-
- ret = ceph_osdc_start_request(osdc, osd_req, false);
- if (ret < 0)
- goto done_err;
-
- if (!rbd_cb) {
- u64 version;
-
- ret = ceph_osdc_wait_request(osdc, osd_req);
- version = le64_to_cpu(osd_req->r_reassert_version.version);
- if (ver)
- *ver = version;
- dout("reassert_ver=%llu\n", (unsigned long long) version);
- ceph_osdc_put_request(osd_req);
- }
- return ret;
-
-done_err:
- if (bio)
- bio_chain_put(osd_req->r_bio);
- ceph_osdc_put_request(osd_req);
-
- return ret;
-}
-
-/*
- * Do a synchronous ceph osd operation
- */
-static int rbd_req_sync_op(struct rbd_device *rbd_dev,

-			   int flags,
-			   struct ceph_osd_req_op *op,
-			   const char *object_name,
-			   u64 ofs, u64 inbound_size,
-			   char *inbound,
-			   u64 *ver)

-{
- int ret;
- struct page **pages;
- int num_pages;
-
- rbd_assert(op != NULL);
-
- num_pages = calc_pages_for(ofs, inbound_size);
- pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
- if (IS_ERR(pages))
- return PTR_ERR(pages);
-
- ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,

-			  object_name, ofs, inbound_size, NULL,
-			  pages, num_pages,
-			  flags,
-			  op,
-			  NULL,
-			  ver);

- if (ret < 0)
- goto done;
-
- if ((flags & CEPH_OSD_FLAG_READ) && inbound)
- ret = ceph_copy_from_page_vector(pages, inbound, ofs, ret);
-
-done:
- ceph_release_page_vector(pages, num_pages);
- return ret;
-}
-
static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
struct rbd_obj_request *obj_request)
{
@@ -1288,45 +1168,6 @@ static int rbd_obj_request_wait(struct rbd_obj_request *obj_request)
return wait_for_completion_interruptible(&obj_request->completion);
}

-/*
- * Synchronous osd object method call
- */
-static int rbd_req_sync_exec(struct rbd_device *rbd_dev,

-			     const char *object_name,
-			     const char *class_name,
-			     const char *method_name,
-			     const char *outbound,
-			     size_t outbound_size,
-			     char *inbound,
-			     size_t inbound_size,
-			     u64 *ver)

-{
- struct ceph_osd_req_op *op;
- int ret;
-
- /*
- * Any input parameters required by the method we're calling
- * will be sent along with the class and method names as
- * part of the message payload. That data and its size are
- * supplied via the indata and indata_len fields (named from
- * the perspective of the server side) in the OSD request
- * operation.
- */
- op = rbd_osd_req_op_create(CEPH_OSD_OP_CALL, class_name,
- method_name, outbound, outbound_size);
- if (!op)
- return -ENOMEM;
-
- ret = rbd_req_sync_op(rbd_dev, CEPH_OSD_FLAG_READ, op,
- object_name, 0, inbound_size, inbound,
- ver);
-
- rbd_osd_req_op_destroy(op);
-
- dout("cls_exec returned %d\n", ret);
- return ret;
-}
-
/* Returns true if this call completed the last object request */

static bool rbd_block_request_complete(struct rbd_image_request
*image_request,
@@ -2825,7 +2666,6 @@ static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
__le64 size;
} __attribute__ ((packed)) size_buf = { 0 };

- (void) rbd_req_sync_exec; /* Avoid a warning */
ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
"rbd", "get_size", (char *) &snapid, sizeof (snapid),

1.7.9.5

This is an update of the first patch in my request tracking code series. After posting it the other day I identified some problems related to reference counting of image and object requests. I also am starting to look at the details of implementing layered reads, and ended up making some substantive changes. Since I have not seen any review feedback I thought the best thing would be to just re-post the updated patch.

The remaining patches in the series have changed accordingly, but they have not really changed substantively, so I am not re-posting those (but will if it's requested).

The main functional change is that an image request no longer maintains an array of object request pointers, it maintains a list of object requests. This simplifies some things, and makes the image request structure fixed size.

A few other functional changes:
- Reference counting of object and image requests is now done sensibly.
- Image requests now support a callback when complete, which will be used for layered I/O requests.
- There are a few new helper functions that encapsulate tying an object request to an image request.
- An distinct value is now used for the "which" field for object requests not associated with a image request (mainly used for validation/assertions).

Other changes:
- Everything that was named "image_request" now uses "img_request" instead.
- A few blocks and lines of code have been rearranged.

The updated series is available on the ceph-client git repository in the branch "wip-rbd-review-v2".

-Alex

This patch fully implements the new request tracking code for rbd I/O requests.

Each I/O request to an rbd image will get an rbd_image_request structure allocated to track it. This provides access to all information about the original request, as well as access to the set of one or more object requests that are initiated as a result of the image request.

An rbd_obj_request structure defines a request sent to a single osd object (possibly) as part of an rbd image request. An rbd object request refers to a ceph_osd_request structure built up to represent the request; for now it will contain a single osd operation. It also provides space to hold the result status and the version of the object when the osd request completes.

An rbd_obj_request structure can also stand on its own. This will be used for reading the version 1 header object, for issuing acknowledgements to event notifications, and for making object method calls.

All rbd object requests now complete asynchronously with respect to the osd client--they supply a common callback routine.

This resolves:

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

---
v2: - fixed reference counting
- image request callback support
- image/object connection helper functions
- distinct BAD_WHICH value for non-image object requests

 drivers/block/rbd.c |  622
++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 620 insertions(+), 2 deletions(-)

index 6689363..46a61dd 100644
@@ -181,6 +181,67 @@ struct rbd_req_coll {
 	struct rbd_req_status	status[0];
 };
 
+struct rbd_img_request;
+typedef void (*rbd_img_callback_t)(struct rbd_img_request *);
+
+#define	BAD_WHICH	U32_MAX		/* Good which or bad which, which? */
+
+struct rbd_obj_request;
+typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);
+
+enum obj_req_type { obj_req_bio };	/* More types to come */
+
+struct rbd_obj_request {
+	const char		*object_name;
+	u64			offset;		/* object start byte */
+	u64			length;		/* bytes from offset */
+
+	struct rbd_img_request	*img_request;
+	struct list_head	links;
+	u32			which;		/* posn image request list */
+
+	enum obj_req_type	type;
+	struct bio		*bio_list;
+
+	struct ceph_osd_request	*osd_req;
+
+	u64			xferred;	/* bytes transferred */
+	u64			version;
+	s32			result;
+	atomic_t		done;
+
+	rbd_obj_callback_t	callback;
+
+	struct kref		kref;
+};
+
+struct rbd_img_request {
+	struct request		*rq;
+	struct rbd_device	*rbd_dev;
+	u64			offset;	/* starting image byte offset */
+	u64			length;	/* byte count from offset */
+	bool			write_request;	/* false for read */
+	union {
+		struct ceph_snap_context *snapc;	/* for writes */
+		u64		snap_id;		/* for reads */
+	};
+	spinlock_t		completion_lock;
+	u32			next_completion;
+	rbd_img_callback_t	callback;
+
+	u32			obj_request_count;
+	struct list_head	obj_requests;
+
+	struct kref		kref;
+};
+
+#define for_each_obj_request(ireq, oreq) \
+	list_for_each_entry(oreq, &ireq->obj_requests, links)
+#define for_each_obj_request_from(ireq, oreq) \
+	list_for_each_entry_from(oreq, &ireq->obj_requests, links)
+#define for_each_obj_request_safe(ireq, oreq, n) \
+	list_for_each_entry_safe_reverse(oreq, n, &ireq->obj_requests, links)
+
 /*
  * a single io request
  */

@@ -1031,6 +1092,62 @@ out_err:
 	return NULL;
 }
 
+static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
+{
+	kref_get(&obj_request->kref);
+}
+
+static void rbd_obj_request_destroy(struct kref *kref);
+static void rbd_obj_request_put(struct rbd_obj_request *obj_request)
+{
+	rbd_assert(obj_request != NULL);
+	kref_put(&obj_request->kref, rbd_obj_request_destroy);
+}
+
+static void rbd_img_request_get(struct rbd_img_request *img_request)
+{
+	kref_get(&img_request->kref);
+}
+
+static void rbd_img_request_destroy(struct kref *kref);
+static void rbd_img_request_put(struct rbd_img_request *img_request)
+{
+	rbd_assert(img_request != NULL);
+	kref_put(&img_request->kref, rbd_img_request_destroy);
+}
+
+static inline void rbd_img_obj_request_add(struct rbd_img_request

*img_request,
+ struct rbd_obj_request *obj_request)
+{
+ rbd_obj_request_get(obj_request);
+ obj_request->img_request = img_request;
+ list_add_tail(&obj_request->links, &img_request->obj_requests);
+ obj_request->which = img_request->obj_request_count++;
+ rbd_assert(obj_request->which != BAD_WHICH);
+}
+
+static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
+ struct rbd_obj_request *obj_request)
+{
+ rbd_assert(obj_request->which != BAD_WHICH);
+ obj_request->which = BAD_WHICH;
+ list_del(&obj_request->links);
+ rbd_assert(obj_request->img_request == img_request);
+ obj_request->callback = NULL;
+ obj_request->img_request = NULL;
+ rbd_obj_request_put(obj_request);
+}
+
+static bool obj_req_type_valid(enum obj_req_type type) +{
+ switch (type) {
+ case obj_req_bio:
+ return true;
+ default:
+ return false;
+ }
+}
+
struct ceph_osd_req_op *rbd_osd_req_op_create(u16 opcode, ...) {
struct ceph_osd_req_op *op;
@@ -1395,6 +1512,26 @@ done:
return ret;
}

+static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
+ struct rbd_obj_request *obj_request)
+{
+ return ceph_osdc_start_request(osdc, obj_request->osd_req, false);
+}
+
+static void rbd_img_request_complete(struct rbd_img_request *img_request) +{
+ if (img_request->callback)
+ img_request->callback(img_request);
+ else
+ rbd_img_request_put(img_request);
+}
+
+static void rbd_obj_request_complete(struct rbd_obj_request *obj_request) +{
+ if (obj_request->callback)
+ obj_request->callback(obj_request);
+}
+
/*
* Request sync osd read
*/
@@ -1618,6 +1755,487 @@ static int rbd_dev_do_request(struct request *rq,
return 0;
}

+static void rbd_osd_read_callback(struct rbd_obj_request *obj_request,
+ struct ceph_osd_op *op)
+{
+ u64 xferred;
+
+ /*
+ * We support a 64-bit length, but ultimately it has to be
+ * passed to blk_end_request(), which takes an unsigned int.
+ */
+ xferred = le64_to_cpu(op->extent.length);
+ rbd_assert(xferred < (u64) UINT_MAX);
+ if (obj_request->result == (s32) -ENOENT) {
+ zero_bio_chain(obj_request->bio_list, 0);
+ obj_request->result = 0;
+ } else if (xferred < obj_request->length && !obj_request->result) {
+ zero_bio_chain(obj_request->bio_list, xferred);
+ xferred = obj_request->length;
+ }
+ obj_request->xferred = xferred;
+ atomic_set(&obj_request->done, 1);
+}
+
+static void rbd_osd_write_callback(struct rbd_obj_request *obj_request,
+ struct ceph_osd_op *op)
+{
+ obj_request->xferred = le64_to_cpu(op->extent.length);
+ atomic_set(&obj_request->done, 1);
+}
+
+static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
+ struct ceph_msg *msg)
+{
+ struct rbd_obj_request *obj_request = osd_req->r_priv;
+ struct ceph_osd_reply_head *reply_head;
+ struct ceph_osd_op *op;
+ u32 num_ops;
+ u16 opcode;
+
+ rbd_assert(osd_req == obj_request->osd_req);
+ rbd_assert(!!obj_request->img_request ^
+ (obj_request->which == BAD_WHICH));
+
+ obj_request->xferred = le32_to_cpu(msg->hdr.data_len);
+ reply_head = msg->front.iov_base;
+ obj_request->result = (s32) le32_to_cpu(reply_head->result);
+ obj_request->version = le64_to_cpu(osd_req->r_reassert_version.version);
+
+ num_ops = le32_to_cpu(reply_head->num_ops);
+ WARN_ON(num_ops != 1); /* For now */
+
+ op = &reply_head->ops[0];
+ opcode = le16_to_cpu(op->op);
+ switch (opcode) {
+ case CEPH_OSD_OP_READ:
+ rbd_osd_read_callback(obj_request, op);
+ break;
+ case CEPH_OSD_OP_WRITE:
+ rbd_osd_write_callback(obj_request, op);
+ break;
+ default:
+ rbd_warn(NULL, "%s: unsupported op %hu\n",
+ obj_request->object_name, (unsigned short) opcode);
+ break;
+ }
+
+ if (atomic_read(&obj_request->done))
+ rbd_obj_request_complete(obj_request);
+}
+
+static struct ceph_osd_request *rbd_osd_req_create(
+ struct rbd_device *rbd_dev,
+ bool write_request,
+ struct rbd_obj_request *obj_request,
+ struct ceph_osd_req_op *op)
+{
+ struct rbd_img_request *img_request = obj_request->img_request;
+ struct ceph_snap_context *snapc = NULL;
+ struct ceph_osd_client *osdc;
+ struct ceph_osd_request *osd_req;
+ struct timespec now;
+ struct timespec *mtime;
+ u64 snap_id = CEPH_NOSNAP;
+ u64 offset = obj_request->offset;
+ u64 length = obj_request->length;
+
+ if (img_request) {
+ rbd_assert(img_request->write_request == write_request);
+ if (img_request->write_request)
+ snapc = img_request->snapc;
+ else
+ snap_id = img_request->snap_id;
+ }
+
+ /* Allocate and initialize the request, for the single op */
+
+ osdc = &rbd_dev->rbd_client->client->osdc;
+ osd_req = ceph_osdc_alloc_request(osdc, snapc, 1, false, GFP_ATOMIC);
+ if (!osd_req)
+ return NULL; /* ENOMEM */
+
+ rbd_assert(obj_req_type_valid(obj_request->type));
+ switch (obj_request->type) {
+ case obj_req_bio:
+ rbd_assert(obj_request->bio_list != NULL);
+ osd_req->r_bio = obj_request->bio_list;
+ bio_get(osd_req->r_bio);
+ /* osd client requires "num pages" even for bio */
+ osd_req->r_num_pages = calc_pages_for(offset, length);
+ break;
+ }
+
+ if (write_request) {
+ osd_req->r_flags = CEPH_OSD_FLAG_WRITE | CEPH_OSD_FLAG_ONDISK;
+ now = CURRENT_TIME;
+ mtime = &now;
+ } else {
+ osd_req->r_flags = CEPH_OSD_FLAG_READ;
+ mtime = NULL; /* not needed for reads */
+ offset = 0; /* These are not used... */
+ length = 0; /* ...for osd read requests */
+ }
+
+ osd_req->r_callback = rbd_osd_req_callback;
+ osd_req->r_priv = obj_request;
+
+ /* No trailing '\0' required for the object name in the request */
+
+ osd_req->r_oid_len = strlen(obj_request->object_name);
+ rbd_assert(osd_req->r_oid_len <= sizeof (osd_req->r_oid));
+ memcpy(osd_req->r_oid, obj_request->object_name, osd_req->r_oid_len);
+
+ osd_req->r_file_layout = rbd_dev->layout; /* struct */
+
+ /* osd_req will get its own reference to snapc (if non-null) */
+
+ ceph_osdc_build_request(osd_req, offset, length, 1, op,
+ snapc, snap_id, mtime);
+
+ return osd_req;
+}
+
+static void rbd_osd_req_destroy(struct ceph_osd_request *osd_req) +{
+ ceph_osdc_put_request(osd_req);
+}
+
+/* object_name is assumed to be a non-null pointer and NUL-terminated */ +
+static struct rbd_obj_request *rbd_obj_request_create(const char *object_name,
+ u64 offset, u64 length,
+ enum obj_req_type type)
+{
+ struct rbd_obj_request *obj_request;
+ size_t size;
+ char *name;
+
+ rbd_assert(obj_req_type_valid(type));
+
+ size = strlen(object_name) + 1;
+ obj_request = kzalloc(sizeof (*obj_request) + size, GFP_KERNEL);
+ if (!obj_request)
+ return NULL;
+
+ name = (char *)(obj_request + 1);
+ obj_request->object_name = memcpy(name, object_name, size);
+ obj_request->offset = offset;
+ obj_request->length = length;
+ obj_request->which = BAD_WHICH;
+ obj_request->type = type;
+ INIT_LIST_HEAD(&obj_request->links);
+ atomic_set(&obj_request->done, 0);
+ kref_init(&obj_request->kref);
+
+ return obj_request;
+}
+
+static void rbd_obj_request_destroy(struct kref *kref)
+{
+ struct rbd_obj_request *obj_request;
+
+ obj_request = container_of(kref, struct rbd_obj_request, kref);
+
+ rbd_assert(obj_request->img_request == NULL);
+ rbd_assert(obj_request->which == BAD_WHICH);
+
+ if (obj_request->osd_req)
+ rbd_osd_req_destroy(obj_request->osd_req);
+
+ rbd_assert(obj_req_type_valid(obj_request->type));
+ switch (obj_request->type) {
+ case obj_req_bio:
+ if (obj_request->bio_list)
+ bio_chain_put(obj_request->bio_list);
+ break;
+ }
+
+ kfree(obj_request);
+}
+
+/*
+ * Caller is responsible for filling in the list of object requests
+ * that comprises the image request, and the Linux request pointer
+ * (if there is one).
+ */
+struct rbd_img_request *rbd_img_request_create(struct rbd_device *rbd_dev,
+ u64 offset, u64 length,
+ bool write_request)
+{
+ struct rbd_img_request *img_request;
+ struct ceph_snap_context *snapc = NULL;
+
+ img_request = kmalloc(sizeof (*img_request), GFP_ATOMIC);
+ if (!img_request)
+ return NULL;
+
+ if (write_request) {
+ down_read(&rbd_dev->header_rwsem);
+ snapc = ceph_get_snap_context(rbd_dev->header.snapc);
+ up_read(&rbd_dev->header_rwsem);
+ if (WARN_ON(!snapc)) {
+ kfree(img_request);
+ return NULL; /* Shouldn't happen */
+ }
+ }
+
+ img_request->rq = NULL;
+ img_request->rbd_dev = rbd_dev;
+ img_request->offset = offset;
+ img_request->length = length;
+ img_request->write_request = write_request;
+ if (write_request)
+ img_request->snapc = snapc;
+ else
+ img_request->snap_id = rbd_dev->spec->snap_id;
+ spin_lock_init(&img_request->completion_lock);
+ img_request->next_completion = 0;
+ img_request->callback = NULL;
+ img_request->obj_request_count = 0;
+ INIT_LIST_HEAD(&img_request->obj_requests);
+ kref_init(&img_request->kref);
+
+ rbd_img_request_get(img_request); /* Avoid a warning */
+ rbd_img_request_put(img_request); /* TEMPORARY */
+
+ return img_request;
+}
+
+static void rbd_img_request_destroy(struct kref *kref)
+{
+ struct rbd_img_request *img_request;
+ struct rbd_obj_request *obj_request;
+ struct rbd_obj_request *next_obj_request;
+
+ img_request = container_of(kref, struct rbd_img_request, kref);
+
+ for_each_obj_request_safe(img_request, obj_request, next_obj_request)
+ rbd_img_obj_request_del(img_request, obj_request);
+
+ if (img_request->write_request)
+ ceph_put_snap_context(img_request->snapc);
+
+ kfree(img_request);
+}
+
+static int rbd_img_request_fill_bio(struct rbd_img_request *img_request,
+ struct bio *bio_list)
+{
+ struct rbd_device *rbd_dev = img_request->rbd_dev;
+ struct rbd_obj_request *obj_request = NULL;
+ struct rbd_obj_request *next_obj_request;
+ unsigned int bio_offset;
+ u64 image_offset;
+ u64 resid;
+ u16 opcode;
+
+ opcode = img_request->write_request ? CEPH_OSD_OP_WRITE
+ : CEPH_OSD_OP_READ;
+ bio_offset = 0;
+ image_offset = img_request->offset;
+ rbd_assert(image_offset == bio_list->bi_sector << SECTOR_SHIFT);
+ resid = img_request->length;
+ while (resid) {
+ const char *object_name;
+ unsigned int clone_size;
+ struct ceph_osd_req_op *op;
+ u64 offset;
+ u64 length;
+
+ object_name = rbd_segment_name(rbd_dev, image_offset);
+ if (!object_name)
+ goto out_unwind;
+ offset = rbd_segment_offset(rbd_dev, image_offset);
+ length = rbd_segment_length(rbd_dev, image_offset, resid);
+ obj_request = rbd_obj_request_create(object_name,
+ offset, length, obj_req_bio);
+ kfree(object_name); /* object request has its own copy */
+ if (!obj_request)
+ goto out_unwind;
+
+ rbd_assert(length <= (u64) UINT_MAX);
+ clone_size = (unsigned int) length;
+ obj_request->bio_list = bio_chain_clone_range(&bio_list,
+ &bio_offset, clone_size,
+ GFP_ATOMIC);
+ if (!obj_request->bio_list)
+ goto out_partial;
+
+ /*
+ * Build up the op to use in building the osd
+ * request. Note that the contents of the op are
+ * copied by rbd_osd_req_create().
+ */
+ op = rbd_osd_req_op_create(opcode, offset, length);
+ if (!op)
+ goto out_partial;
+ obj_request->osd_req = rbd_osd_req_create(rbd_dev,
+ img_request->write_request,
+ obj_request, op);
+ rbd_osd_req_op_destroy(op);
+ if (!obj_request->osd_req)
+ goto out_partial;
+ /* status and version are initially zero-filled */
+
+ rbd_img_obj_request_add(img_request, obj_request);
+
+ image_offset += length;
+ resid -= length;
+ }
+
+ return 0;
+
+out_partial:
+ rbd_obj_request_put(obj_request);
+out_unwind:
+ for_each_obj_request_safe(img_request, obj_request, next_obj_request)
+ rbd_obj_request_put(obj_request);
+
+ return -ENOMEM;
+}
+
+static void rbd_img_obj_callback(struct rbd_obj_request *obj_request) +{
+ struct rbd_img_request *img_request;
+ u32 which = obj_request->which;
+ bool more = true;
+
+ img_request = obj_request->img_request;
+ rbd_assert(img_request != NULL);
+ rbd_assert(img_request->rq != NULL);
+ rbd_assert(which != BAD_WHICH);
+ rbd_assert(which < img_request->obj_request_count);
+ rbd_assert(which >= img_request->next_completion);
+
+ spin_lock(&img_request->completion_lock);
+ if (which != img_request->next_completion)
+ goto out;
+
+ for_each_obj_request_from(img_request, obj_request) {
+ unsigned int xferred;
+ int result;
+
+ rbd_assert(more);
+ rbd_assert(which < img_request->obj_request_count);
+
+ if (!atomic_read(&obj_request->done))
+ break;
+
+ rbd_assert(obj_request->xferred <= (u64) UINT_MAX);
+ xferred = (unsigned int) obj_request->xferred;
+ result = (int) obj_request->result;
+ if (result)
+ rbd_warn(NULL, "obj_request %s result %d xferred %u\n",
+ img_request->write_request ? "write" : "read",
+ result, xferred);
+
+ more = blk_end_request(img_request->rq, result, xferred);
+ which++;
+ }
+ rbd_assert(more ^ (which == img_request->obj_request_count));
+ img_request->next_completion = which;
+out:
+ spin_unlock(&img_request->completion_lock);
+
+ if (!more)
+ rbd_img_request_complete(img_request);
+}
+
+static int rbd_img_request_submit(struct rbd_img_request *img_request) +{
+ struct rbd_device *rbd_dev = img_request->rbd_dev;
+ struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
+ struct rbd_obj_request *obj_request;
+
+ for_each_obj_request(img_request, obj_request) {
+ int ret;
+
+ obj_request->callback = rbd_img_obj_callback;
+ ret = rbd_obj_request_submit(osdc, obj_request);
+ if (ret)
+ return ret;
+ /*
+ * The image request has its own reference to each
+ * of its object requests, so we can safely drop the
+ * initial one here.
+ */
+ rbd_obj_request_put(obj_request);
+ }
+
+ return 0;
+}
+
+static void rbd_request_fn(struct request_queue *q)
+{
+ struct rbd_device *rbd_dev = q->queuedata;
+ bool read_only = rbd_dev->mapping.read_only;
+ struct request *rq;
+ int result;
+
+ while ((rq = blk_fetch_request(q))) {
+ bool write_request = rq_data_dir(rq) == WRITE;
+ struct rbd_img_request *img_request;
+ u64 offset;
+ u64 length;
+
+ /* Ignore any non-FS requests that filter through. */
+
+ if (rq->cmd_type != REQ_TYPE_FS) {
+ __blk_end_request_all(rq, 0);
+ continue;
+ }
+
+ spin_unlock_irq(q->queue_lock);
+
+ /* Disallow writes to a read-only device */
+
+ if (write_request) {
+ result = -EROFS;
+ if (read_only)
+ goto end_request;
+ rbd_assert(rbd_dev->spec->snap_id == CEPH_NOSNAP);
+ }
+
+ /* Quit early if the snapshot has disappeared */
+
+ if (!atomic_read(&rbd_dev->exists)) {
+ dout("request for non-existent snapshot");
+ rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
+ result = -ENXIO;
+ goto end_request;
+ }
+
+ offset = (u64) blk_rq_pos(rq) << SECTOR_SHIFT;
+ length = (u64) blk_rq_bytes(rq);
+
+ result = -EINVAL;
+ if (WARN_ON(offset && length > U64_MAX - offset + 1))
+ goto end_request; /* Shouldn't happen */
+
+ result = -ENOMEM;
+ img_request = rbd_img_request_create(rbd_dev, offset, length,
+ write_request);
+ if (!img_request)
+ goto end_request;
+
+ img_request->rq = rq;
+
+ result = rbd_img_request_fill_bio(img_request, rq->bio);
+ if (!result)
+ result = rbd_img_request_submit(img_request);
+ if (result)
+ rbd_img_request_put(img_request);
+end_request:
+ spin_lock_irq(q->queue_lock);
+ if (result < 0) {
+ rbd_warn(rbd_dev, "obj_request %s result %d\n",
+ write_request ? "write" : "read", result);
+ __blk_end_request_all(rq, result);
+ }
+ }
+}
+
/*
* block device queue callback
*/
@@ -1929,8 +2547,8 @@ static int rbd_init_disk(struct rbd_device *rbd_dev)
disk->fops = &rbd_bd_ops;
disk->private_data = rbd_dev;

- /* init rq */
- q = blk_init_queue(rbd_rq_fn, &rbd_dev->lock);
+ (void) rbd_rq_fn; /* avoid a warning */
+ q = blk_init_queue(rbd_request_fn, &rbd_dev->lock);
if (!q)
goto out_disk;

1.7.9.5

This was requested. So I'm going to re-post the remainder of the series. They'll show up as responses to the first one I posted.

-Alex

Now that the request function has been replaced by one using the new request management data structures the old one can go away. Deleting it makes rbd_dev_do_request() no longer needed, and deleting that makes other functions unneeded, and so on.

---
 drivers/block/rbd.c |  319
---------------------------------------------------
 1 file changed, 319 deletions(-)

index 46a61dd..7caddea 100644
@@ -621,18 +621,6 @@ static void rbd_put_client(struct rbd_client *rbdc)
 		kref_put(&rbdc->kref, rbd_client_release);
 }
 
-/*
- * Destroy requests collection
- */
-static void rbd_coll_release(struct kref *kref)
-{
-	struct rbd_req_coll *coll =
-		container_of(kref, struct rbd_req_coll, kref);
-
-	dout("rbd_coll_release %p\n", coll);
-	kfree(coll);
-}
-
 static bool rbd_image_format_valid(u32 image_format)
 {
 	return image_format == 1 || image_format == 2;

@@ -876,28 +864,6 @@ static u64 rbd_segment_length(struct rbd_device

*rbd_dev,
return length;
}

-static int rbd_get_num_segments(struct rbd_image_header *header,
- u64 ofs, u64 len)
-{
- u64 start_seg;
- u64 end_seg;
- u64 result;
-
- if (!len)
- return 0;
- if (len - 1 > U64_MAX - ofs)
- return -ERANGE;
-
- start_seg = ofs >> header->obj_order;
- end_seg = (ofs + len - 1) >> header->obj_order;
-
- result = end_seg - start_seg + 1;
- if (result > (u64) INT_MAX)
- return -ERANGE;
-
- return (int) result;
-}
-
/*
* returns the size of an object in the image
*/
@@ -1216,52 +1182,6 @@ 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,
-				   s32 ret, u64 len)

-{
- struct request_queue *q;
- int min, max, i;
-
- dout("rbd_coll_end_req_index %p index %d ret %d len %llu\n",
- coll, index, (int)ret, (unsigned long long)len);
-
- if (!rq)
- return;
-
- if (!coll) {
- blk_end_request(rq, ret, len);
- return;
- }
-
- q = rq->q;
-
- spin_lock_irq(q->queue_lock);
- coll->status[index].done = 1;
- coll->status[index].rc = ret;
- coll->status[index].bytes = len;
- max = min = coll->num_done;
- while (max < coll->total && coll->status[max].done)
- max++;
-
- for (i = min; i<max; i++) {
- __blk_end_request(rq, (int)coll->status[i].rc,
- coll->status[i].bytes);
- coll->num_done++;
- kref_put(&coll->kref, rbd_coll_release);
- }
- spin_unlock_irq(q->queue_lock);
-}
-
-static void rbd_coll_end_req(struct rbd_request *rbd_req,
- s32 ret, u64 len)
-{
- rbd_coll_end_req_index(rbd_req->rq,
- rbd_req->coll, rbd_req->coll_index,
- ret, len);
-}
-
/*
* Send ceph osd request
*/
@@ -1361,46 +1281,6 @@ done_osd_req:
return ret;
}

-/*
- * Ceph osd op callback
- */
-static void rbd_req_cb(struct ceph_osd_request *osd_req, struct ceph_msg *msg)
-{
- struct rbd_request *rbd_req = osd_req->r_priv;
- struct ceph_osd_reply_head *replyhead;
- struct ceph_osd_op *op;
- s32 rc;
- u64 bytes;
- int read_op;
-
- /* parse reply */
- replyhead = msg->front.iov_base;
- WARN_ON(le32_to_cpu(replyhead->num_ops) == 0);
- op = (void *)(replyhead + 1);
- rc = (s32)le32_to_cpu(replyhead->result);
- bytes = le64_to_cpu(op->extent.length);
- read_op = (le16_to_cpu(op->op) == CEPH_OSD_OP_READ);
-
- dout("rbd_req_cb bytes=%llu readop=%d rc=%d\n",
- (unsigned long long) bytes, read_op, (int) rc);
-
- if (rc == (s32)-ENOENT && read_op) {
- zero_bio_chain(rbd_req->bio, 0);
- rc = 0;
- } else if (rc == 0 && read_op && bytes < rbd_req->len) {
- zero_bio_chain(rbd_req->bio, bytes);
- bytes = rbd_req->len;
- }
-
- rbd_coll_end_req(rbd_req, rc, bytes);
-
- if (rbd_req->bio)
- bio_chain_put(rbd_req->bio);
-
- ceph_osdc_put_request(osd_req);
- kfree(rbd_req);
-}
-
static void rbd_simple_req_cb(struct ceph_osd_request *osd_req,
struct ceph_msg *msg)
{
@@ -1448,70 +1328,6 @@ done:
return ret;
}

-/*
- * Do an asynchronous ceph osd operation
- */
-static int rbd_do_op(struct request *rq,

-		     struct rbd_device *rbd_dev,
-		     struct ceph_snap_context *snapc,
-		     u64 ofs, u64 len,
-		     struct bio *bio,
-		     struct rbd_req_coll *coll,
-		     int coll_index)

-{
- const char *seg_name;
- u64 seg_ofs;
- u64 seg_len;
- int ret;
- struct ceph_osd_req_op *op;
- int opcode;
- int flags;
- u64 snapid;
-
- seg_name = rbd_segment_name(rbd_dev, ofs);
- if (!seg_name)
- return -ENOMEM;
- seg_len = rbd_segment_length(rbd_dev, ofs, len);
- seg_ofs = rbd_segment_offset(rbd_dev, ofs);
-
- if (rq_data_dir(rq) == WRITE) {
- opcode = CEPH_OSD_OP_WRITE;
- flags = CEPH_OSD_FLAG_WRITE|CEPH_OSD_FLAG_ONDISK;
- snapid = CEPH_NOSNAP;
- } else {
- opcode = CEPH_OSD_OP_READ;
- flags = CEPH_OSD_FLAG_READ;
- rbd_assert(!snapc);
- snapid = rbd_dev->spec->snap_id;
- }
-
- ret = -ENOMEM;
- op = rbd_osd_req_op_create(opcode, seg_ofs, seg_len);
- if (!op)
- goto done;
-
- /* we've taken care of segment sizes earlier when we
- cloned the bios. We should never have a segment
- truncated at this point */
- rbd_assert(seg_len == len);
-
- ret = rbd_do_request(rq, rbd_dev, snapc, snapid,

-			     seg_name, seg_ofs, seg_len,
-			     bio,
-			     NULL, 0,
-			     flags,
-			     op,
-			     coll, coll_index,
-			     rbd_req_cb, NULL);

- if (ret < 0)
- rbd_coll_end_req_index(rq, coll, coll_index,
- (s32)ret, seg_len);
- rbd_osd_req_op_destroy(op);
-done:
- kfree(seg_name);
- return ret;
-}
-
static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
struct rbd_obj_request *obj_request)
{
@@ -1683,78 +1499,6 @@ static int rbd_req_sync_exec(struct rbd_device *rbd_dev,
return ret;
}

-static struct rbd_req_coll *rbd_alloc_coll(int num_reqs)
-{
- struct rbd_req_coll *coll =
- kzalloc(sizeof(struct rbd_req_coll) +
- sizeof(struct rbd_req_status) * num_reqs,
- GFP_ATOMIC);
-
- if (!coll)
- return NULL;
- coll->total = num_reqs;
- kref_init(&coll->kref);
- return coll;
-}
-
-static int rbd_dev_do_request(struct request *rq,
- struct rbd_device *rbd_dev,
- struct ceph_snap_context *snapc,
- u64 ofs, unsigned int size,
- struct bio *bio_chain)
-{
- int num_segs;
- struct rbd_req_coll *coll;
- unsigned int bio_offset;
- int cur_seg = 0;
-
- dout("%s 0x%x bytes at 0x%llx\n",
- rq_data_dir(rq) == WRITE ? "write" : "read",
- size, (unsigned long long) blk_rq_pos(rq) * SECTOR_SIZE);
-
- num_segs = rbd_get_num_segments(&rbd_dev->header, ofs, size);
- if (num_segs <= 0)
- return num_segs;
-
- coll = rbd_alloc_coll(num_segs);
- if (!coll)
- return -ENOMEM;
-
- bio_offset = 0;
- do {
- u64 limit = rbd_segment_length(rbd_dev, ofs, size);
- unsigned int clone_size;
- struct bio *bio_clone;
-
- BUG_ON(limit > (u64)UINT_MAX);
- clone_size = (unsigned int)limit;
- dout("bio_chain->bi_vcnt=%hu\n", bio_chain->bi_vcnt);
-
- kref_get(&coll->kref);
-
- /* Pass a cloned bio chain via an osd request */
-
- bio_clone = bio_chain_clone_range(&bio_chain,
- &bio_offset, clone_size,
- GFP_ATOMIC);
- if (bio_clone)
- (void)rbd_do_op(rq, rbd_dev, snapc,
- ofs, clone_size,
- bio_clone, coll, cur_seg);
- else
- rbd_coll_end_req_index(rq, coll, cur_seg,
- (s32)-ENOMEM,
- clone_size);
- size -= clone_size;
- ofs += clone_size;
-
- cur_seg++;
- } while (size > 0);
- kref_put(&coll->kref, rbd_coll_release);
-
- return 0;
-}
-
static void rbd_osd_read_callback(struct rbd_obj_request *obj_request,
struct ceph_osd_op *op)
{
@@ -2237,68 +1981,6 @@ end_request:
}

/*
- * block device queue callback
- */
-static void rbd_rq_fn(struct request_queue *q)
-{
- struct rbd_device *rbd_dev = q->queuedata;
- bool read_only = rbd_dev->mapping.read_only;
- struct request *rq;
-
- while ((rq = blk_fetch_request(q))) {
- struct ceph_snap_context *snapc = NULL;
- unsigned int size = 0;
- int result;
-
- dout("fetched request\n");
-
- /* Filter out block requests we don't understand */
-
- if ((rq->cmd_type != REQ_TYPE_FS)) {
- __blk_end_request_all(rq, 0);
- continue;
- }
- spin_unlock_irq(q->queue_lock);
-
- /* Write requests need a reference to the snapshot context */
-
- if (rq_data_dir(rq) == WRITE) {
- result = -EROFS;
- if (read_only) /* Can't write to a read-only device */
- goto out_end_request;
-
- /*
- * Note that each osd request will take its
- * own reference to the snapshot context
- * supplied. The reference we take here
- * just guarantees the one we provide stays
- * valid.
- */
- down_read(&rbd_dev->header_rwsem);
- snapc = ceph_get_snap_context(rbd_dev->header.snapc);
- up_read(&rbd_dev->header_rwsem);
- rbd_assert(snapc != NULL);
- } else if (!atomic_read(&rbd_dev->exists)) {
- rbd_assert(rbd_dev->spec->snap_id != CEPH_NOSNAP);
- dout("request for non-existent snapshot");
- result = -ENXIO;
- goto out_end_request;
- }
-
- size = blk_rq_bytes(rq);
- result = rbd_dev_do_request(rq, rbd_dev, snapc,
- blk_rq_pos(rq) * SECTOR_SIZE,
- size, rq->bio);
-out_end_request:
- if (snapc)
- ceph_put_snap_context(snapc);
- spin_lock_irq(q->queue_lock);
- if (!size || result < 0)
- __blk_end_request_all(rq, result);
- }
-}
-
-/*
* a queue callback. Makes sure that we don't create a bio that spans
across
* multiple osd objects. One exception would be with a single page bios,
* which we handle later at bio_chain_clone_range()
@@ -2547,7 +2229,6 @@ static int rbd_init_disk(struct rbd_device *rbd_dev)
disk->fops = &rbd_bd_ops;
disk->private_data = rbd_dev;

- (void) rbd_rq_fn; /* avoid a warning */
q = blk_init_queue(rbd_request_fn, &rbd_dev->lock); if (!q)
goto out_disk;

1.7.9.5

The two remaining callers of rbd_do_request() always pass a null collection pointer, so the "coll" and "coll_index" parameters are not needed. There is no other use of that data structure, so it can be eliminated.

Deleting them means there is no need to allocate a rbd_request structure for the callback function. And since that's the only use of *that* structure, it too can be eliminated.

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

index 7caddea..3302cea 100644
@@ -162,25 +162,6 @@ struct rbd_client {
 	struct list_head	node;
 };
 
-/*
- * a request completion status
- */
-struct rbd_req_status {
-	int done;
-	s32 rc;
-	u64 bytes;
-};
-
-/*
- * a collection of requests
- */
-struct rbd_req_coll {
-	int			total;
-	int			num_done;
-	struct kref		kref;
-	struct rbd_req_status	status[0];
-};
-
 struct rbd_img_request;
 typedef void (*rbd_img_callback_t)(struct rbd_img_request *);
 

@@ -242,18 +223,6 @@ struct rbd_img_request {
 #define for_each_obj_request_safe(ireq, oreq, n) \
 	list_for_each_entry_safe_reverse(oreq, n, &ireq->obj_requests, links)
 
-/*
- * a single io request
- */
-struct rbd_request {
-	struct request		*rq;		/* blk layer request */
-	struct bio		*bio;		/* cloned bio */
-	struct page		**pages;	/* list of used pages */
-	u64			len;
-	int			coll_index;
-	struct rbd_req_coll	*coll;
-};
-
 struct rbd_snap {
 	struct	device		dev;
 	const char		*name;

@@ -1195,21 +1164,18 @@ static int rbd_do_request(struct request *rq,
 			  int num_pages,
 			  int flags,
 			  struct ceph_osd_req_op *op,
-			  struct rbd_req_coll *coll,
-			  int coll_index,
 			  void (*rbd_cb)(struct ceph_osd_request *,
 					 struct ceph_msg *),
 			  u64 *ver)
 {
 	struct ceph_osd_client *osdc;
 	struct ceph_osd_request *osd_req;
-	struct rbd_request *rbd_req = NULL;
 	struct timespec mtime = CURRENT_TIME;
 	int ret;
 
-	dout("rbd_do_request object_name=%s ofs=%llu len=%llu coll=%p[%d]\n",
+	dout("rbd_do_request object_name=%s ofs=%llu len=%llu\n",
 		object_name, (unsigned long long) ofs,
-		(unsigned long long) len, coll, coll_index);
+		(unsigned long long) len);
 
 	osdc = &rbd_dev->rbd_client->client->osdc;
 	osd_req = ceph_osdc_alloc_request(osdc, snapc, 1, false, GFP_NOIO);

@@ -1223,22 +1189,8 @@ static int rbd_do_request(struct request *rq,
 		bio_get(osd_req->r_bio);
 	}
 
-	if (coll) {
-		ret = -ENOMEM;
-		rbd_req = kmalloc(sizeof(*rbd_req), GFP_NOIO);
-		if (!rbd_req)
-			goto done_osd_req;
-
-		rbd_req->rq = rq;
-		rbd_req->bio = bio;
-		rbd_req->pages = pages;
-		rbd_req->len = len;
-		rbd_req->coll = coll;
-		rbd_req->coll_index = coll_index;
-	}
-
 	osd_req->r_callback = rbd_cb;
-	osd_req->r_priv = rbd_req;
+	osd_req->r_priv = NULL;
 
 	strncpy(osd_req->r_oid, object_name, sizeof(osd_req->r_oid));
 	osd_req->r_oid_len = strlen(osd_req->r_oid);

@@ -1274,8 +1226,6 @@ static int rbd_do_request(struct request *rq,
 done_err:
 	if (bio)
 		bio_chain_put(osd_req->r_bio);
-	kfree(rbd_req);
-done_osd_req:
 	ceph_osdc_put_request(osd_req);
 
 	return ret;

@@ -1314,7 +1264,6 @@ static int rbd_req_sync_op(struct rbd_device *rbd_dev,
 			  pages, num_pages,
 			  flags,
 			  op,
-			  NULL, 0,
 			  NULL,
 			  ver);
 	if (ret < 0)

@@ -1390,7 +1339,6 @@ static int rbd_req_sync_notify_ack(struct

rbd_device *rbd_dev,
NULL, 0, CEPH_OSD_FLAG_READ, op,
- NULL, 0,
rbd_simple_req_cb, NULL);

rbd_osd_req_op_destroy(op);

1.7.9.5

Reimplement the synchronous read operation used for reading a version 1 header using the new request tracking code. Name the resulting function rbd_obj_read_sync() to better reflect that it's a full object operation, not an object request. To do this, implement a new obj_req_pages object request type.

This implements a new mechanism to allow the caller to wait for completion for an rbd_obj_request by calling rbd_obj_request_wait().

This partially resolves:

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

---
 drivers/block/rbd.c |   96
++++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 92 insertions(+), 4 deletions(-)

index 3302cea..742236b 100644
@@ -170,7 +170,7 @@ typedef void (*rbd_img_callback_t)(struct

rbd_img_request *);
struct rbd_obj_request;
typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);

-enum obj_req_type { obj_req_bio }; /* More types to come */
+enum obj_req_type { obj_req_bio, obj_req_pages };

struct rbd_obj_request {
const char *object_name;
@@ -182,7 +182,13 @@ struct rbd_obj_request {
u32 which; /* posn image request list */

enum obj_req_type type;
- struct bio *bio_list;
+ union {
+ struct bio *bio_list;
+ struct {
+ struct page **pages;
+ u32 page_count;
+ };
+ };

struct ceph_osd_request *osd_req;

@@ -192,6 +198,7 @@ struct rbd_obj_request {
atomic_t done;

rbd_obj_callback_t callback;
+ struct completion completion;

struct kref kref;
};
@@ -1077,6 +1084,7 @@ static bool obj_req_type_valid(enum obj_req_type type)
{
switch (type) {
case obj_req_bio:
+ case obj_req_pages:
return true;
default:
return false;
@@ -1291,14 +1299,23 @@ static void rbd_img_request_complete(struct rbd_img_request *img_request)
rbd_img_request_put(img_request);
}

+/* Caller is responsible for rbd_obj_request_destroy(obj_request) */ +
+static int rbd_obj_request_wait(struct rbd_obj_request *obj_request) +{
+ return wait_for_completion_interruptible(&obj_request->completion);
+}
+
static void rbd_obj_request_complete(struct rbd_obj_request *obj_request) {
if (obj_request->callback)
obj_request->callback(obj_request);
+ else
+ complete_all(&obj_request->completion);
}

/*
- * Request sync osd read
+ * Synchronously read a range from an object into a provided buffer */
static int rbd_req_sync_read(struct rbd_device *rbd_dev,
const char *object_name,
@@ -1556,6 +1573,11 @@ static struct ceph_osd_request *rbd_osd_req_create(
/* osd client requires "num pages" even for bio */ osd_req->r_num_pages = calc_pages_for(offset, length); break;
+ case obj_req_pages:
+ osd_req->r_pages = obj_request->pages;
+ osd_req->r_num_pages = obj_request->page_count;
+ osd_req->r_page_alignment = offset & ~PAGE_MASK;
+ break;
}

if (write_request) {
@@ -1618,6 +1640,7 @@ static struct rbd_obj_request *rbd_obj_request_create(const char *object_name,
obj_request->type = type; INIT_LIST_HEAD(&obj_request->links); atomic_set(&obj_request->done, 0);
+ init_completion(&obj_request->completion);
kref_init(&obj_request->kref);

return obj_request;
@@ -1641,6 +1664,11 @@ static void rbd_obj_request_destroy(struct kref *kref)
if (obj_request->bio_list)
bio_chain_put(obj_request->bio_list);
break;
+ case obj_req_pages:
+ if (obj_request->pages)
+ ceph_release_page_vector(obj_request->pages,
+ obj_request->page_count);
+ break;
}

kfree(obj_request);
@@ -1988,6 +2016,65 @@ static void rbd_free_disk(struct rbd_device *rbd_dev)
put_disk(disk);
}

+static int rbd_obj_read_sync(struct rbd_device *rbd_dev,
+ const char *object_name,
+ u64 offset, u64 length,
+ char *buf, u64 *version)
+
+{
+ struct ceph_osd_req_op *op;
+ struct rbd_obj_request *obj_request;
+ struct ceph_osd_client *osdc;
+ struct page **pages = NULL;
+ u32 page_count;
+ int ret;
+
+ page_count = (u32) calc_pages_for(offset, length);
+ pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
+ if (IS_ERR(pages))
+ ret = PTR_ERR(pages);
+
+ ret = -ENOMEM;
+ obj_request = rbd_obj_request_create(object_name, offset, length,
+ obj_req_pages);
+ if (!obj_request)
+ goto out;
+
+ obj_request->pages = pages;
+ obj_request->page_count = page_count;
+
+ op = rbd_osd_req_op_create(CEPH_OSD_OP_READ, offset, length);
+ if (!op)
+ goto out;
+ obj_request->osd_req = rbd_osd_req_create(rbd_dev, false,
+ obj_request, op);
+ rbd_osd_req_op_destroy(op);
+ if (!obj_request->osd_req)
+ goto out;
+
+ osdc = &rbd_dev->rbd_client->client->osdc;
+ ret = rbd_obj_request_submit(osdc, obj_request);
+ if (ret)
+ goto out;
+ ret = rbd_obj_request_wait(obj_request);
+ if (ret)
+ goto out;
+
+ ret = obj_request->result;
+ if (ret < 0)
+ goto out;
+ ret = ceph_copy_from_page_vector(pages, buf, 0, obj_request->xferred);
+ if (version)
+ *version = obj_request->version;
+out:
+ if (obj_request)
+ rbd_obj_request_put(obj_request);
+ else
+ ceph_release_page_vector(pages, page_count);
+
+ return ret;
+}
+
/*
* Read the complete header for the given rbd device.
*
@@ -2026,7 +2113,8 @@ rbd_dev_v1_header_read(struct rbd_device *rbd_dev, u64 *version)
if (!ondisk)
return ERR_PTR(-ENOMEM);

- ret = rbd_req_sync_read(rbd_dev, rbd_dev->header_name,
+ (void) rbd_req_sync_read; /* avoid a warning */
+ ret = rbd_obj_read_sync(rbd_dev, rbd_dev->header_name,
0, size,
(char *) ondisk, version);

1.7.9.5

Delete rbd_req_sync_read() is no longer used, so get rid of it.

---
 drivers/block/rbd.c |   24 ------------------------
 1 file changed, 24 deletions(-)

index 742236b..750fc73 100644
@@ -1315,29 +1315,6 @@ static void rbd_obj_request_complete(struct

rbd_obj_request *obj_request)
}

/*
- * Synchronously read a range from an object into a provided buffer
- */
-static int rbd_req_sync_read(struct rbd_device *rbd_dev,

-			  const char *object_name,
-			  u64 ofs, u64 len,
-			  char *buf,
-			  u64 *ver)

-{
- struct ceph_osd_req_op *op;
- int ret;
-
- 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, ver);
- rbd_osd_req_op_destroy(op);
-
- return ret;
-}
-
-/*
* Request sync osd watch
*/
static int rbd_req_sync_notify_ack(struct rbd_device *rbd_dev,
@@ -2113,7 +2090,6 @@ rbd_dev_v1_header_read(struct rbd_device *rbd_dev, u64 *version)
if (!ondisk)
return ERR_PTR(-ENOMEM);

- (void) rbd_req_sync_read; /* avoid a warning */
ret = rbd_obj_read_sync(rbd_dev, rbd_dev->header_name,
0, size,
(char *) ondisk, version);

1.7.9.5

Implement a new function to set up or tear down a watch event for an mapped rbd image header using the new request code.

Create a new object request type "nodata" to handle this. And define rbd_osd_trivial_callback() which simply marks a request done.

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

index 750fc73..7b1eddc 100644
@@ -170,7 +170,7 @@ typedef void (*rbd_img_callback_t)(struct

rbd_img_request *);
struct rbd_obj_request;
typedef void (*rbd_obj_callback_t)(struct rbd_obj_request *);

-enum obj_req_type { obj_req_bio, obj_req_pages }; +enum obj_req_type { obj_req_nodata, obj_req_bio, obj_req_pages };

struct rbd_obj_request {
const char *object_name;
@@ -1083,6 +1083,7 @@ static inline void rbd_img_obj_request_del(struct rbd_img_request *img_request,
static bool obj_req_type_valid(enum obj_req_type type)
{
switch (type) {
+ case obj_req_nodata:
case obj_req_bio:
case obj_req_pages:
return true;
@@ -1306,6 +1307,12 @@ static int rbd_obj_request_wait(struct rbd_obj_request *obj_request)
return wait_for_completion_interruptible(&obj_request->completion);
}

+static void rbd_osd_trivial_callback(struct rbd_obj_request *obj_request,
+ struct ceph_osd_op *op)
+{
+ atomic_set(&obj_request->done, 1);
+}
+
static void rbd_obj_request_complete(struct rbd_obj_request *obj_request) {
if (obj_request->callback)
@@ -1500,6 +1507,9 @@ static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
case CEPH_OSD_OP_WRITE:
rbd_osd_write_callback(obj_request, op);
break;
+ case CEPH_OSD_OP_WATCH:
+ rbd_osd_trivial_callback(obj_request, op);
+ break;
default:
rbd_warn(NULL, "%s: unsupported op %hu\n",
obj_request->object_name, (unsigned short) opcode);
@@ -1543,6 +1553,8 @@ static struct ceph_osd_request *rbd_osd_req_create(

rbd_assert(obj_req_type_valid(obj_request->type)); switch (obj_request->type) {
+ case obj_req_nodata:
+ break; /* Nothing to do */
case obj_req_bio:
rbd_assert(obj_request->bio_list != NULL); osd_req->r_bio = obj_request->bio_list;
@@ -1637,6 +1649,8 @@ static void rbd_obj_request_destroy(struct kref *kref)

rbd_assert(obj_req_type_valid(obj_request->type)); switch (obj_request->type) {
+ case obj_req_nodata:
+ break; /* Nothing to do */
case obj_req_bio:
if (obj_request->bio_list)
bio_chain_put(obj_request->bio_list);
@@ -1863,6 +1877,72 @@ static int rbd_img_request_submit(struct rbd_img_request *img_request)
return 0;
}

+/*
+ * Request sync osd watch/unwatch. The value of "start" determines
+ * whether a watch request is being initiated or torn down.
+ */
+static int rbd_dev_header_watch_sync(struct rbd_device *rbd_dev, int start) +{
+ struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
+ struct rbd_obj_request *obj_request;
+ struct ceph_osd_req_op *op;
+ int ret;
+
+ rbd_assert(start ^ !!rbd_dev->watch_event);
+ rbd_assert(start ^ !!rbd_dev->watch_request);
+
+ if (start) {
+ ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0, rbd_dev,
+ &rbd_dev->watch_event);
+ if (ret < 0)
+ return ret;
+ }
+
+ ret = -ENOMEM;
+ obj_request = rbd_obj_request_create(rbd_dev->header_name,
+ 0, 0, obj_req_nodata);
+ if (!obj_request)
+ goto out_cancel;
+
+ op = rbd_osd_req_op_create(CEPH_OSD_OP_WATCH,
+ rbd_dev->watch_event->cookie,
+ rbd_dev->header.obj_version, start);
+ if (!op)
+ goto out_cancel;
+ obj_request->osd_req = rbd_osd_req_create(rbd_dev, true,
+ obj_request, op);
+ rbd_osd_req_op_destroy(op);
+ if (!obj_request->osd_req)
+ goto out_cancel;
+
+ if (start) {
+ rbd_dev->watch_request = obj_request->osd_req;
+ ceph_osdc_set_request_linger(osdc, rbd_dev->watch_request);
+ }
+ ret = rbd_obj_request_submit(osdc, obj_request);
+ if (ret)
+ goto out_cancel;
+ ret = rbd_obj_request_wait(obj_request);
+ if (ret)
+ goto out_cancel;
+
+ ret = obj_request->result;
+ if (ret)
+ goto out_cancel;
+
+ if (start)
+ goto done; /* Done if setting up the watch request */
+out_cancel:
+ /* Cancel the event if we're tearing down, or on error */
+ ceph_osdc_cancel_event(rbd_dev->watch_event);
+ rbd_dev->watch_event = NULL;
+done:
+ if (obj_request)
+ rbd_obj_request_put(obj_request);
+
+ return ret;
+}
+
static void rbd_request_fn(struct request_queue *q)
{
struct rbd_device *rbd_dev = q->queuedata;
@@ -3880,7 +3960,8 @@ static int rbd_dev_probe_finish(struct rbd_device *rbd_dev)
if (ret)
goto err_out_bus;

- ret = rbd_req_sync_watch(rbd_dev, 1);
+ (void) rbd_req_sync_watch; /* avoid a warning */
+ ret = rbd_dev_header_watch_sync(rbd_dev, 1);
if (ret)
goto err_out_bus;

@@ -4043,7 +4124,7 @@ static void rbd_dev_release(struct device *dev)
rbd_dev->watch_request);
}
if (rbd_dev->watch_event)
- rbd_req_sync_watch(rbd_dev, 0);
+ rbd_dev_header_watch_sync(rbd_dev, 0);

/* clean up and free blkdev */
rbd_free_disk(rbd_dev);

1.7.9.5

Get rid of rbd_req_sync_watch(), because it is no longer used.

---
 drivers/block/rbd.c |   42 ------------------------------------------
 1 file changed, 42 deletions(-)

index 7b1eddc..8f659f3 100644
@@ -1367,47 +1367,6 @@ static void rbd_watch_cb(u64 ver, u64 notify_id,

u8 opcode, void *data)
rbd_req_sync_notify_ack(rbd_dev, hver, notify_id);
}

-/*
- * Request sync osd watch/unwatch. The value of "start" determines
- * whether a watch request is being initiated or torn down.
- */
-static int rbd_req_sync_watch(struct rbd_device *rbd_dev, int start) -{
- struct ceph_osd_req_op *op;
- int ret = 0;
-
- rbd_assert(start ^ !!rbd_dev->watch_event);
- rbd_assert(start ^ !!rbd_dev->watch_request);
-
- if (start) {
- struct ceph_osd_client *osdc;
-
- osdc = &rbd_dev->rbd_client->client->osdc;
- ret = ceph_osdc_create_event(osdc, rbd_watch_cb, 0, rbd_dev,
- &rbd_dev->watch_event);
- if (ret < 0)
- return ret;
- }
-
- 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, NULL);

-
- /* 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;
- }
- rbd_osd_req_op_destroy(op);
-
- return ret;
-}

/*
* Synchronous osd object method call
@@ -3960,7 +3919,6 @@ static int rbd_dev_probe_finish(struct rbd_device *rbd_dev)
if (ret)
goto err_out_bus;

- (void) rbd_req_sync_watch; /* avoid a warning */
ret = rbd_dev_header_watch_sync(rbd_dev, 1); if (ret)
goto err_out_bus;

1.7.9.5

Use the new object request tracking mechanism for handling a notify_ack request.

Move the callback function below the definition of this so we don't have to do a pre-declaration.

This resolves:

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

---
 drivers/block/rbd.c |   76
+++++++++++++++++++++++++++++++++++++--------------
 1 file changed, 55 insertions(+), 21 deletions(-)

index 8f659f3..e2b6230 100644
@@ -1347,27 +1347,6 @@ static int rbd_req_sync_notify_ack(struct

rbd_device *rbd_dev,
return ret;
}

-static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data) -{
- struct rbd_device *rbd_dev = (struct rbd_device *)data;
- u64 hver;
- int rc;
-
- if (!rbd_dev)
- return;
-
- dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
- rbd_dev->header_name, (unsigned long long) notify_id,
- (unsigned int) opcode);
- rc = rbd_dev_refresh(rbd_dev, &hver);
- if (rc)
- rbd_warn(rbd_dev, "got notification but failed to "
- " update snaps: %d\n", rc);
-
- rbd_req_sync_notify_ack(rbd_dev, hver, notify_id);
-}
-
-
/*
* Synchronous osd object method call
*/
@@ -1466,6 +1445,7 @@ static void rbd_osd_req_callback(struct ceph_osd_request *osd_req,
case CEPH_OSD_OP_WRITE:
rbd_osd_write_callback(obj_request, op); break;
+ case CEPH_OSD_OP_NOTIFY_ACK:
case CEPH_OSD_OP_WATCH:
rbd_osd_trivial_callback(obj_request, op); break;
@@ -1836,6 +1816,60 @@ static int rbd_img_request_submit(struct rbd_img_request *img_request)
return 0;
}

+static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev,
+ u64 ver, u64 notify_id)
+{
+ struct rbd_obj_request *obj_request;
+ struct ceph_osd_req_op *op;
+ struct ceph_osd_client *osdc;
+ int ret;
+
+ obj_request = rbd_obj_request_create(rbd_dev->header_name, 0, 0,
+ obj_req_nodata);
+ if (!obj_request)
+ return -ENOMEM;
+
+ ret = -ENOMEM;
+ op = rbd_osd_req_op_create(CEPH_OSD_OP_NOTIFY_ACK, notify_id, ver);
+ if (!op)
+ goto out;
+ obj_request->osd_req = rbd_osd_req_create(rbd_dev, false,
+ obj_request, op);
+ rbd_osd_req_op_destroy(op);
+ if (!obj_request->osd_req)
+ goto out;
+
+ osdc = &rbd_dev->rbd_client->client->osdc;
+ ret = rbd_obj_request_submit(osdc, obj_request);
+ if (!ret)
+ ret = rbd_obj_request_wait(obj_request);
+out:
+ rbd_obj_request_put(obj_request);
+
+ return ret;
+}
+
+static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data) +{
+ struct rbd_device *rbd_dev = (struct rbd_device *)data;
+ u64 hver;
+ int rc;
+
+ if (!rbd_dev)
+ return;
+
+ dout("rbd_watch_cb %s notify_id=%llu opcode=%u\n",
+ rbd_dev->header_name, (unsigned long long) notify_id,
+ (unsigned int) opcode);
+ rc = rbd_dev_refresh(rbd_dev, &hver);
+ if (rc)
+ rbd_warn(rbd_dev, "got notification but failed to "
+ " update snaps: %d\n", rc);
+
+ (void) rbd_req_sync_notify_ack; /* avoid a warning */
+ rbd_obj_notify_ack_sync(rbd_dev, hver, notify_id);
+}
+
/*
* Request sync osd watch/unwatch. The value of "start" determines
* whether a watch request is being initiated or torn down.

1.7.9.5

Get rid rbd_req_sync_notify_ack() because it is no longer used. As a result rbd_simple_req_cb() becomes unreferenced, so get rid of that too.

---
 drivers/block/rbd.c |   33 ---------------------------------
 1 file changed, 33 deletions(-)

index e2b6230..b952b2f 100644
@@ -1240,12 +1240,6 @@ done_err:
 	return ret;
 }
 
-static void rbd_simple_req_cb(struct ceph_osd_request *osd_req,
-				struct ceph_msg *msg)
-{
-	ceph_osdc_put_request(osd_req);
-}
-
 /*
  * Do a synchronous ceph osd operation
  */

@@ -1322,32 +1316,6 @@ static void rbd_obj_request_complete(struct

rbd_obj_request *obj_request)
}

/*
- * Request sync osd watch
- */
-static int rbd_req_sync_notify_ack(struct rbd_device *rbd_dev,
- u64 ver,
- u64 notify_id)
-{
- struct ceph_osd_req_op *op;
- int ret;
-
- op = rbd_osd_req_op_create(CEPH_OSD_OP_NOTIFY_ACK, notify_id, ver);
- if (!op)
- return -ENOMEM;
-
- ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,

-			  rbd_dev->header_name, 0, 0, NULL,
-			  NULL, 0,
-			  CEPH_OSD_FLAG_READ,
-			  op,
-			  rbd_simple_req_cb, NULL);

-
- rbd_osd_req_op_destroy(op);
-
- return ret;
-}
-
-/*
* Synchronous osd object method call
*/
static int rbd_req_sync_exec(struct rbd_device *rbd_dev,
@@ -1866,7 +1834,6 @@ static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
rbd_warn(rbd_dev, "got notification but failed to "
" update snaps: %d\n", rc);

- (void) rbd_req_sync_notify_ack; /* avoid a warning */
rbd_obj_notify_ack_sync(rbd_dev, hver, notify_id);
}

1.7.9.5

When we receive notification of a change to an rbd image's header object we need to refresh our information about the image (its size and snapshot context). Once we have refreshed our rbd image we need to acknowledge the notification.

This acknowledgement was previously done synchronously, but there's really no need to wait for it to complete.

Change it so the caller doesn't wait for the notify acknowledgement request to complete. And change the name to reflect it's no longer synchronous.

This resolves:

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

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

index b952b2f..48650d1 100644
@@ -1784,7 +1784,7 @@ static int rbd_img_request_submit(struct

rbd_img_request *img_request)
return 0;
}

-static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev, +static int rbd_obj_notify_ack(struct rbd_device *rbd_dev,
u64 ver, u64 notify_id)
{
struct rbd_obj_request *obj_request;
@@ -1808,11 +1808,11 @@ static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev,
goto out;

osdc = &rbd_dev->rbd_client->client->osdc;
+ obj_request->callback = rbd_obj_request_put;
ret = rbd_obj_request_submit(osdc, obj_request);
- if (!ret)
- ret = rbd_obj_request_wait(obj_request);
out:
- rbd_obj_request_put(obj_request);
+ if (ret)
+ rbd_obj_request_put(obj_request);

return ret;
}
@@ -1834,7 +1834,7 @@ static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
rbd_warn(rbd_dev, "got notification but failed to "
" update snaps: %d\n", rc);

- rbd_obj_notify_ack_sync(rbd_dev, hver, notify_id);
+ rbd_obj_notify_ack(rbd_dev, hver, notify_id);
}

/*

1.7.9.5

Reimplement synchronous object method calls using the new request tracking code. Use the name rbd_obj_method_sync()

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

index 48650d1..5ad2ac2 100644
@@ -1413,6 +1413,7 @@ static void rbd_osd_req_callback(struct

ceph_osd_request *osd_req,
case CEPH_OSD_OP_WRITE:
rbd_osd_write_callback(obj_request, op); break;
+ case CEPH_OSD_OP_CALL:
case CEPH_OSD_OP_NOTIFY_ACK:
case CEPH_OSD_OP_WATCH:
rbd_osd_trivial_callback(obj_request, op);
@@ -1903,6 +1904,81 @@ done:
return ret;
}

+/*
+ * Synchronous osd object method call
+ */
+static int rbd_obj_method_sync(struct rbd_device *rbd_dev,

+			     const char *object_name,
+			     const char *class_name,
+			     const char *method_name,
+			     const char *outbound,
+			     size_t outbound_size,
+			     char *inbound,
+			     size_t inbound_size,
+			     u64 *version)

+{
+ struct rbd_obj_request *obj_request;
+ struct ceph_osd_client *osdc;
+ struct ceph_osd_req_op *op;
+ struct page **pages;
+ u32 page_count;
+ int ret;
+
+ /*
+ * Method calls are ultimately read operations but they
+ * don't involve object data (so no offset or length).
+ * The result should placed into the inbound buffer
+ * provided. They also supply outbound data--parameters for
+ * the object method. Currently if this is present it will
+ * be a snapshot id.
+ */
+ page_count = (u32) calc_pages_for(0, inbound_size);
+ pages = ceph_alloc_page_vector(page_count, GFP_KERNEL);
+ if (IS_ERR(pages))
+ return PTR_ERR(pages);
+
+ ret = -ENOMEM;
+ obj_request = rbd_obj_request_create(object_name, 0, 0, obj_req_pages);
+ if (!obj_request)
+ goto out;
+
+ obj_request->pages = pages;
+ obj_request->page_count = page_count;
+
+ op = rbd_osd_req_op_create(CEPH_OSD_OP_CALL, class_name,
+ method_name, outbound, outbound_size);
+ if (!op)
+ goto out;
+ obj_request->osd_req = rbd_osd_req_create(rbd_dev, false,
+ obj_request, op);
+ rbd_osd_req_op_destroy(op);
+ if (!obj_request->osd_req)
+ goto out;
+
+ osdc = &rbd_dev->rbd_client->client->osdc;
+ ret = rbd_obj_request_submit(osdc, obj_request);
+ if (ret)
+ goto out;
+ ret = rbd_obj_request_wait(obj_request);
+ if (ret)
+ goto out;
+
+ ret = obj_request->result;
+ if (ret < 0)
+ goto out;
+ ret = ceph_copy_from_page_vector(pages, inbound, 0,
+ obj_request->xferred);
+ if (version)
+ *version = obj_request->version;
+out:
+ if (obj_request)
+ rbd_obj_request_put(obj_request);
+ else
+ ceph_release_page_vector(pages, page_count);
+
+ return ret;
+}
+
static void rbd_request_fn(struct request_queue *q) {
struct rbd_device *rbd_dev = q->queuedata;
@@ -2753,11 +2829,12 @@ static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
__le64 size;
} __attribute__ ((packed)) size_buf = { 0 };

- ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
+ (void) rbd_req_sync_exec; /* Avoid a warning */
+ ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
"rbd", "get_size",
(char *) &snapid, sizeof (snapid), (char *) &size_buf, sizeof (size_buf), NULL);
- dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
+ dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
if (ret < 0)
return ret;

@@ -2788,14 +2865,14 @@ static int rbd_dev_v2_object_prefix(struct rbd_device *rbd_dev)
if (!reply_buf)
return -ENOMEM;

- ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
+ ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
"rbd", "get_object_prefix", NULL, 0,
reply_buf, RBD_OBJ_PREFIX_LEN_MAX, NULL);
- dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
+ dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
if (ret < 0)
goto out;
- ret = 0; /* rbd_req_sync_exec() can return positive */
+ ret = 0; /* rbd_obj_method_sync() can return positive */

p = reply_buf;
rbd_dev->header.object_prefix = ceph_extract_encoded_string(&p,
@@ -2826,12 +2903,12 @@ static int _rbd_dev_v2_snap_features(struct rbd_device *rbd_dev, u64 snap_id,
u64 incompat;
int ret;

- ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
+ ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
"rbd", "get_features",
(char *) &snapid, sizeof (snapid), (char *) &features_buf, sizeof (features_buf), NULL);
- dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
+ dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
if (ret < 0)
return ret;

@@ -2882,11 +2959,11 @@ static int rbd_dev_v2_parent_info(struct rbd_device *rbd_dev)
}

snapid = cpu_to_le64(CEPH_NOSNAP);
- ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
+ ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
"rbd", "get_parent", (char *) &snapid, sizeof (snapid), (char *) reply_buf, size, NULL);
- dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
+ dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
if (ret < 0)
goto out_err;

@@ -2953,7 +3030,7 @@ static char *rbd_dev_image_name(struct rbd_device *rbd_dev)
if (!reply_buf)
goto out;

- ret = rbd_req_sync_exec(rbd_dev, RBD_DIRECTORY,
+ ret = rbd_obj_method_sync(rbd_dev, RBD_DIRECTORY,
"rbd", "dir_get_name", image_id, image_id_size, (char *) reply_buf, size, NULL);
@@ -3059,11 +3136,11 @@ static int rbd_dev_v2_snap_context(struct rbd_device *rbd_dev, u64 *ver)
if (!reply_buf)
return -ENOMEM;

- ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
+ ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
"rbd", "get_snapcontext", NULL, 0, reply_buf, size, ver);
- dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
+ dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
if (ret < 0)
goto out;

@@ -3128,11 +3205,11 @@ static char *rbd_dev_v2_snap_name(struct rbd_device *rbd_dev, u32 which)
return ERR_PTR(-ENOMEM);

snap_id = cpu_to_le64(rbd_dev->header.snapc->snaps[which]);
- ret = rbd_req_sync_exec(rbd_dev, rbd_dev->header_name,
+ ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
"rbd", "get_snapshot_name", (char *) &snap_id, sizeof (snap_id), reply_buf, size, NULL);
- dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
+ dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
if (ret < 0)
goto out;

@@ -3720,14 +3797,14 @@ static int rbd_dev_image_id(struct rbd_device *rbd_dev)
goto out;
}

- ret = rbd_req_sync_exec(rbd_dev, object_name,
+ ret = rbd_obj_method_sync(rbd_dev, object_name,
"rbd", "get_id",
NULL, 0,
response, RBD_IMAGE_ID_LEN_MAX, NULL);
- dout("%s: rbd_req_sync_exec returned %d\n", __func__, ret);
+ dout("%s: rbd_obj_method_sync returned %d\n", __func__, ret);
if (ret < 0)
goto out;
- ret = 0; /* rbd_req_sync_exec() can return positive */
+ ret = 0; /* rbd_obj_method_sync() can return positive */

p = response;
rbd_dev->spec->image_id = ceph_extract_encoded_string(&p,

1.7.9.5

Get rid rbd_req_sync_exec() because it is no longer used. That eliminates the last use of rbd_req_sync_op(), so get rid of that too. And finally, that leaves rbd_do_request() unreferenced, so get rid of that.

---
 drivers/block/rbd.c |  160
---------------------------------------------------
 1 file changed, 160 deletions(-)

index 5ad2ac2..c807a4c 100644
@@ -1160,126 +1160,6 @@ static void rbd_osd_req_op_destroy(struct

ceph_osd_req_op *op)
kfree(op);
}

-/*
- * Send ceph osd request
- */
-static int rbd_do_request(struct request *rq,

-			  struct rbd_device *rbd_dev,
-			  struct ceph_snap_context *snapc,
-			  u64 snapid,
-			  const char *object_name, u64 ofs, u64 len,
-			  struct bio *bio,
-			  struct page **pages,
-			  int num_pages,
-			  int flags,
-			  struct ceph_osd_req_op *op,
-			  void (*rbd_cb)(struct ceph_osd_request *,

- struct ceph_msg *),
- u64 *ver)
-{
- struct ceph_osd_client *osdc;
- struct ceph_osd_request *osd_req;
- struct timespec mtime = CURRENT_TIME;
- int ret;
-
- dout("rbd_do_request object_name=%s ofs=%llu len=%llu\n",
- object_name, (unsigned long long) ofs,
- (unsigned long long) len);
-
- osdc = &rbd_dev->rbd_client->client->osdc;
- osd_req = ceph_osdc_alloc_request(osdc, snapc, 1, false, GFP_NOIO);
- if (!osd_req)
- return -ENOMEM;
-
- osd_req->r_flags = flags;
- osd_req->r_pages = pages;
- if (bio) {
- osd_req->r_bio = bio;
- bio_get(osd_req->r_bio);
- }
-
- osd_req->r_callback = rbd_cb;
- osd_req->r_priv = NULL;
-
- strncpy(osd_req->r_oid, object_name, sizeof(osd_req->r_oid));
- osd_req->r_oid_len = strlen(osd_req->r_oid);
-
- osd_req->r_file_layout = rbd_dev->layout; /* struct */
- osd_req->r_num_pages = calc_pages_for(ofs, len);
- osd_req->r_page_alignment = ofs & ~PAGE_MASK;
-
- ceph_osdc_build_request(osd_req, ofs, len, 1, op,
- snapc, snapid, &mtime);
-
- if (op->op == CEPH_OSD_OP_WATCH && op->watch.flag) {
- ceph_osdc_set_request_linger(osdc, osd_req);
- rbd_dev->watch_request = osd_req;
- }
-
- ret = ceph_osdc_start_request(osdc, osd_req, false);
- if (ret < 0)
- goto done_err;
-
- if (!rbd_cb) {
- u64 version;
-
- ret = ceph_osdc_wait_request(osdc, osd_req);
- version = le64_to_cpu(osd_req->r_reassert_version.version);
- if (ver)
- *ver = version;
- dout("reassert_ver=%llu\n", (unsigned long long) version);
- ceph_osdc_put_request(osd_req);
- }
- return ret;
-
-done_err:
- if (bio)
- bio_chain_put(osd_req->r_bio);
- ceph_osdc_put_request(osd_req);
-
- return ret;
-}
-
-/*
- * Do a synchronous ceph osd operation
- */
-static int rbd_req_sync_op(struct rbd_device *rbd_dev,

-			   int flags,
-			   struct ceph_osd_req_op *op,
-			   const char *object_name,
-			   u64 ofs, u64 inbound_size,
-			   char *inbound,
-			   u64 *ver)

-{
- int ret;
- struct page **pages;
- int num_pages;
-
- rbd_assert(op != NULL);
-
- num_pages = calc_pages_for(ofs, inbound_size);
- pages = ceph_alloc_page_vector(num_pages, GFP_KERNEL);
- if (IS_ERR(pages))
- return PTR_ERR(pages);
-
- ret = rbd_do_request(NULL, rbd_dev, NULL, CEPH_NOSNAP,

-			  object_name, ofs, inbound_size, NULL,
-			  pages, num_pages,
-			  flags,
-			  op,
-			  NULL,
-			  ver);

- if (ret < 0)
- goto done;
-
- if ((flags & CEPH_OSD_FLAG_READ) && inbound)
- ret = ceph_copy_from_page_vector(pages, inbound, ofs, ret);
-
-done:
- ceph_release_page_vector(pages, num_pages);
- return ret;
-}
-
static int rbd_obj_request_submit(struct ceph_osd_client *osdc,
struct rbd_obj_request *obj_request)
{
@@ -1315,45 +1195,6 @@ static void rbd_obj_request_complete(struct rbd_obj_request *obj_request)
complete_all(&obj_request->completion);
}

-/*
- * Synchronous osd object method call
- */
-static int rbd_req_sync_exec(struct rbd_device *rbd_dev,

-			     const char *object_name,
-			     const char *class_name,
-			     const char *method_name,
-			     const char *outbound,
-			     size_t outbound_size,
-			     char *inbound,
-			     size_t inbound_size,
-			     u64 *ver)

-{
- struct ceph_osd_req_op *op;
- int ret;
-
- /*
- * Any input parameters required by the method we're calling
- * will be sent along with the class and method names as
- * part of the message payload. That data and its size are
- * supplied via the indata and indata_len fields (named from
- * the perspective of the server side) in the OSD request
- * operation.
- */
- op = rbd_osd_req_op_create(CEPH_OSD_OP_CALL, class_name,
- method_name, outbound, outbound_size);
- if (!op)
- return -ENOMEM;
-
- ret = rbd_req_sync_op(rbd_dev, CEPH_OSD_FLAG_READ, op,
- object_name, 0, inbound_size, inbound,
- ver);
-
- rbd_osd_req_op_destroy(op);
-
- dout("cls_exec returned %d\n", ret);
- return ret;
-}
-
static void rbd_osd_read_callback(struct rbd_obj_request *obj_request,
struct ceph_osd_op *op)
{
@@ -2829,7 +2670,6 @@ static int _rbd_dev_v2_snap_size(struct rbd_device *rbd_dev, u64 snap_id,
__le64 size;
} __attribute__ ((packed)) size_buf = { 0 };

- (void) rbd_req_sync_exec; /* Avoid a warning */
ret = rbd_obj_method_sync(rbd_dev, rbd_dev->header_name,
"rbd", "get_size", (char *) &snapid, sizeof (snapid),

1.7.9.5