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