[PATCH 00/12] rbd: new request tracking code

[PATCH 00/12] rbd: new request tracking code

[Alex Elder]

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

[PATCH 01/12] rbd: new request tracking code

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |  596
++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 594 insertions(+), 2 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 6689363..485fa70 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 01/12, v2] rbd: new request tracking code

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
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(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 6689363..46a61dd 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 01/12, v2] rbd: new request tracking code

[Alex Elder]

On 01/24/2013 08:08 AM, Alex Elder wrote:
> 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).

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

[PATCH 02/12, v2] rbd: kill rbd_rq_fn() and all other related code

[Alex Elder]

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.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |  319
---------------------------------------------------
 1 file changed, 319 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 46a61dd..7caddea 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 02/12, v2] rbd: kill rbd_rq_fn() and all other related code

[Josh Durgin]

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:32 AM, Alex Elder wrote:
> 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.
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |  319
> ---------------------------------------------------
>   1 file changed, 319 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 46a61dd..7caddea 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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;
>

[PATCH 03/12, v2] rbd: kill rbd_req_coll and rbd_request

[Alex Elder]

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.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   58
+++------------------------------------------------
 1 file changed, 3 insertions(+), 55 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 7caddea..3302cea 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 03/12, v2] rbd: kill rbd_req_coll and rbd_request

[Josh Durgin]

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:32 AM, Alex Elder wrote:
> 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.
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |   58
> +++------------------------------------------------
>   1 file changed, 3 insertions(+), 55 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 7caddea..3302cea 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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);
>

[PATCH 04/12, v2] rbd: implement sync object read with new code

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   96
++++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 92 insertions(+), 4 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 3302cea..742236b 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 04/12, v2] rbd: implement sync object read with new code

[Josh Durgin]

A couple small comments, but looks good.

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:33 AM, Alex Elder wrote:
> 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
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |   96
> ++++++++++++++++++++++++++++++++++++++++++++++++---
>   1 file changed, 92 insertions(+), 4 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 3302cea..742236b 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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 };

Should be capitalized

>   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;

No need to worry about the version. Nothing should use it.

> +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);
>

[PATCH 05/12, v2] rbd: get rid of rbd_req_sync_read()

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   24 ------------------------
 1 file changed, 24 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 742236b..750fc73 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 05/12, v2] rbd: get rid of rbd_req_sync_read()

[Josh Durgin]

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:33 AM, Alex Elder wrote:
> Delete rbd_req_sync_read() is no longer used, so get rid of it.
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |   24 ------------------------
>   1 file changed, 24 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 742236b..750fc73 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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);
>

[PATCH 06/12, v2] rbd: implement watch/unwatch with new code

[Alex Elder]

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.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   87
+++++++++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 84 insertions(+), 3 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 750fc73..7b1eddc 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 06/12, v2] rbd: implement watch/unwatch with new code

[Josh Durgin]

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:33 AM, Alex Elder wrote:
> 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.
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |   87
> +++++++++++++++++++++++++++++++++++++++++++++++++--
>   1 file changed, 84 insertions(+), 3 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 750fc73..7b1eddc 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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);
>

[PATCH 07/12, v2] rbd: get rid of rbd_req_sync_watch()

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   42 ------------------------------------------
 1 file changed, 42 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 7b1eddc..8f659f3 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 07/12, v2] rbd: get rid of rbd_req_sync_watch()

[Josh Durgin]

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:34 AM, Alex Elder wrote:
> Get rid of rbd_req_sync_watch(), because it is no longer used.
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |   42 ------------------------------------------
>   1 file changed, 42 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 7b1eddc..8f659f3 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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;
>

[PATCH 08/12, v2] rbd: use new code for notify ack

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   76
+++++++++++++++++++++++++++++++++++++--------------
 1 file changed, 55 insertions(+), 21 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 8f659f3..e2b6230 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 08/12, v2] rbd: use new code for notify ack

[Josh Durgin]

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:34 AM, Alex Elder wrote:
> 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
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |   76
> +++++++++++++++++++++++++++++++++++++--------------
>   1 file changed, 55 insertions(+), 21 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 8f659f3..e2b6230 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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.
>

[PATCH 09/12, v2] rbd: get rid of rbd_req_sync_notify_ack()

[Alex Elder]

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.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   33 ---------------------------------
 1 file changed, 33 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index e2b6230..b952b2f 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 09/12, v2] rbd: get rid of rbd_req_sync_notify_ack()

[Josh Durgin]

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:35 AM, Alex Elder wrote:
> 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.
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |   33 ---------------------------------
>   1 file changed, 33 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index e2b6230..b952b2f 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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);
>   }
>

[PATCH 10/12, v2] rbd: send notify ack asynchronously

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index b952b2f..48650d1 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 10/12, v2] rbd: send notify ack asynchronously

[Josh Durgin]

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:35 AM, Alex Elder wrote:
> 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
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |   10 +++++-----
>   1 file changed, 5 insertions(+), 5 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index b952b2f..48650d1 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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);
>   }
>
>   /*
>

[PATCH 11/12, v2] rbd: implement sync method with new code

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |  111
+++++++++++++++++++++++++++++++++++++++++++--------
 1 file changed, 94 insertions(+), 17 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 48650d1..5ad2ac2 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 11/12, v2] rbd: implement sync method with new code

[Josh Durgin]

With the version parameter removed now or in a later patch:

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:36 AM, Alex Elder wrote:
> Reimplement synchronous object method calls using the new request
> tracking code.  Use the name rbd_obj_method_sync()
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |  111
> +++++++++++++++++++++++++++++++++++++++++++--------
>   1 file changed, 94 insertions(+), 17 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 48650d1..5ad2ac2 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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)

You can get rid of the version parameter. It's not useful to any
current or planned rbd code that I can think of.

> +{
> +	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;

Here too

> +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,
>

[PATCH 12/12, v2] rbd: get rid of rbd_req_sync_exec()

[Alex Elder]

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.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |  160
---------------------------------------------------
 1 file changed, 160 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 5ad2ac2..c807a4c 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

Re: [PATCH 12/12, v2] rbd: get rid of rbd_req_sync_exec()

[Josh Durgin]

Reviewed-by: Josh Durgin <josh.durgin@inktank.com>

On 01/24/2013 08:36 AM, Alex Elder wrote:
> 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.
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |  160
> ---------------------------------------------------
>   1 file changed, 160 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 5ad2ac2..c807a4c 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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),
>

Re: [PATCH 01/12, v2] rbd: new request tracking code

[Josh Durgin]

On 01/24/2013 06:08 AM, Alex Elder wrote:
> 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
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
> 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(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 6689363..46a61dd 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -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 */

enum labels should be capitalized.

> +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;

This version is only used (uselessly) for the watch operation. It
should be removed in a future patch (along with the obj_ver in the
header).

> +	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;

It'd be nice to have a comment describing what this lock protects.

> +	u32			next_completion;
> +	rbd_img_callback_t	callback;
> +
> +	u32			obj_request_count;
> +	struct list_head	obj_requests;

Maybe note that these are rbd_obj_requests, and not ceph_osd_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);
> +}

Why rely on the callback to rbd_img_request_put()? Wouldn't it be a
bit simpler to unconditionally do the put here?

> +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 */

It looks like ceph_calc_object_layout() does require the trailing '\0':

osd_client.c:
   __map_request()
     ceph_calc_object_layout(...,->r_oid,...)
       strlen(oid)

> +	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);

In the current equivalent code (rbd_coll_end_req_index), we use
spin_lock_irq(), and don't hold the spinlock while calling
blk_end_request.

Why the change, and is this change safe?

> +	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;
>

Re: [PATCH 01/12, v2] rbd: new request tracking code

[Alex Elder]

On 01/29/2013 04:43 AM, Josh Durgin wrote:
> On 01/24/2013 06:08 AM, Alex Elder wrote:
>> 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.

Responses to your review comments are below.

Thank you very much for careful, thorough, and thoughtful
work on this.  I believe it's very important and you do
a good job of it.

					-Alex

>> 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
>>
>> Signed-off-by: Alex Elder <elder@inktank.com>
>> ---
>> 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(-)
>>
>> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
>> index 6689363..46a61dd 100644
>> --- a/drivers/block/rbd.c
>> +++ b/drivers/block/rbd.c
>> @@ -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 */
> 
> enum labels should be capitalized.

Not where I come from (at least not always), but I'll
convert these to be all caps.

>> +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;
> 
> This version is only used (uselessly) for the watch operation. It
> should be removed in a future patch (along with the obj_ver in the
> header).

I'm not sure whether the description is quite right, but:
   http://tracker.ceph.com/issues/3952

>> +    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;
> 
> It'd be nice to have a comment describing what this lock protects.

OK, I'll add one.  The lock may not be needed, but for now I left
it in.  It protects updates to the next_completion field.

>> +    u32            next_completion;
>> +    rbd_img_callback_t    callback;
>> +
>> +    u32            obj_request_count;
>> +    struct list_head    obj_requests;
> 
> Maybe note that these are rbd_obj_requests, and not ceph_osd_requests.

I meant for the name to suggest that (I'm pretty consistent
about using osd_req and obj_request prefixes), but I'll add
a short comment since the list type doesn't make it obvious.

>> +    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
>>    */

. . .

>> @@ -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);
>> +}
> 
> Why rely on the callback to rbd_img_request_put()? Wouldn't it be a
> bit simpler to unconditionally do the put here?

I think it's because I wanted the callback to have the chance
to defer completion, and hang onto the reference until that
time rather than taking another reference.  But right now it
isn't used so it's sort of moot anyway.

Unless you object, I'm going to leave it as-is for now,
and when I look at upcoming patches that will use this
functionality I may change it to drop the reference
unconditionally as you suggest.

>> +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
>>    */

. . .

>> +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 */
> 
> It looks like ceph_calc_object_layout() does require the trailing '\0':

Bummer.  The request itself doesn't need it.

You're right though.  I'll fix that.  (And I'm inclined to
fix ceph_calc_object_layout() so it doesn't require it...)

All that's required is changing "<=" to "<" in this assertion:
        rbd_assert(osd_req->r_oid_len <= sizeof (osd_req->r_oid));

> osd_client.c:
>   __map_request()
>     ceph_calc_object_layout(...,->r_oid,...)
>       strlen(oid)
> 
>> +    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_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);
> 
> In the current equivalent code (rbd_coll_end_req_index), we use
> spin_lock_irq(), and don't hold the spinlock while calling
> blk_end_request.
> 
> Why the change, and is this change safe?

In the current "collection" code, the queue lock for the rbd
device is used.  And that lock *is* held (as required) when
__blk_end_request() function is called (and interrupts are
disabled)

In the new code, each image request has its own lock.  We
call blk_end_request() when there's something to tell the
block layer about (and that form of the function will
acquire the queue lock itself).  This separates the lock
from the Linux block code.

It's possible it is not be safe though.  Without thinking
a bit harder about this I'm not entirely sure whether I
need to disable interrupts.  So for now I'll just change
it to use spin_lock_irq() to be on the safe (and easier)
side.

As I said earlier, I'm not even sure a spinlock is
required here, atomic operations and barriers around
accesses to the next_completion field may be enough.
But that's an optimization for another day...

>> +    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);
>> +}
>> +

. . .

[PATCH 02/12] rbd: kill rbd_rq_fn() and all other related code

[Alex Elder]

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.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |  297
---------------------------------------------------
 1 file changed, 297 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 485fa70..0e1edc0 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 03/12] rbd: kill rbd_req_coll and rbd_request

[Alex Elder]

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.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   58
+++------------------------------------------------
 1 file changed, 3 insertions(+), 55 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 0e1edc0..c1b99c7 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 04/12] rbd: implement sync object read with new code

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   96
++++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 92 insertions(+), 4 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index c1b99c7..5a8fef4 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 05/12] rbd: get rid of rbd_req_sync_read()

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   24 ------------------------
 1 file changed, 24 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 5a8fef4..6193c69 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 06/12] rbd: implement watch/unwatch with new code

[Alex Elder]

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.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   87
+++++++++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 84 insertions(+), 3 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 6193c69..3c110b3 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 07/12] rbd: get rid of rbd_req_sync_watch()

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   42 ------------------------------------------
 1 file changed, 42 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 3c110b3..7dedd18 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 08/12] rbd: use new code for notify ack

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   76
+++++++++++++++++++++++++++++++++++++--------------
 1 file changed, 55 insertions(+), 21 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 7dedd18..08465ae 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 09/12] rbd: get rid of rbd_req_sync_notify_ack()

[Alex Elder]

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.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   33 ---------------------------------
 1 file changed, 33 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 08465ae..a6ca5db 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 10/12] rbd: send notify ack asynchronously

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index a6ca5db..e66695a 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 11/12] rbd: implement sync method with new code

[Alex Elder]

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

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |   10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index a6ca5db..e66695a 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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

[PATCH 12/12] rbd: get rid of rbd_req_sync_exec()

[Alex Elder]

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.

Signed-off-by: Alex Elder <elder@inktank.com>
---
 drivers/block/rbd.c |  160
---------------------------------------------------
 1 file changed, 160 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 934bb61..b655088 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -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