[PATCH 00/11] rbd: layered read functionality

[PATCH 00/11] rbd: layered read functionality

[Alex Elder]

This series puts in place code that is able to handle
read requests on rbd clone images, forwarding them to
a parent snapshot image if necessary.  Missing from this
series is a temporary patch at the end which actually
activates this functionality.  That will not go in
until the rest of rbd layering functionality is in place.

I'm going to restate that.  This code implements reads
for layered rbd images, but the functionality will not
be usable quite yet.

Most of the series is adding flags to image and object
requests, and putting in place some accounting in (parent)
image requests so it can be used to record the completion
of an object request that initiated it.

					-Alex

[PATCH 01/11] rbd: record overall image request result
[PATCH 02/11] rbd: record aggregate image transfer count
[PATCH 03/11] rbd: record image-relative offset in object requests
[PATCH 04/11] rbd: define image request flags
[PATCH 05/11] rbd: define image request originator flag
[PATCH 06/11] rbd: define image request layered flag
[PATCH 07/11] rbd: encapsulate image object end request handling
[PATCH 08/11] rbd: define an rbd object request flags field
[PATCH 09/11] rbd: add an object request flag for image data objects
[PATCH 10/11] rbd: probe the parent of an image if present
[PATCH 11/11] rbd: implement layered reads

[PATCH 01/11] rbd: record overall image request result

[Alex Elder]

If any image object request produces a non-zero result, preserve
that as the result of the overall image request.  If multiple
objects have non-zero results, save only the first one.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 503e64f..69eab66 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -214,6 +214,7 @@ struct rbd_img_request {
 	spinlock_t		completion_lock;/* protects next_completion */
 	u32			next_completion;
 	rbd_img_callback_t	callback;
+	int			result;	/* first nonzero obj_request result */

 	u32			obj_request_count;
 	struct list_head	obj_requests;	/* rbd_obj_request structs */
@@ -1488,6 +1489,7 @@ static struct rbd_img_request *rbd_img_request_create(
 	spin_lock_init(&img_request->completion_lock);
 	img_request->next_completion = 0;
 	img_request->callback = NULL;
+	img_request->result = 0;
 	img_request->obj_request_count = 0;
 	INIT_LIST_HEAD(&img_request->obj_requests);
 	kref_init(&img_request->kref);
@@ -1552,13 +1554,16 @@ static void rbd_img_obj_callback(struct
rbd_obj_request *obj_request)
 		if (!obj_request_done_test(obj_request))
 			break;

-		rbd_assert(obj_request->xferred <= (u64) UINT_MAX);
-		xferred = (unsigned int) obj_request->xferred;
-		result = (int) obj_request->result;
-		if (result)
+		rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
+		xferred = (unsigned int)obj_request->xferred;
+		result = obj_request->result;
+		if (result) {
 			rbd_warn(NULL, "obj_request %s result %d xferred %u\n",
 				img_request->write_request ? "write" : "read",
 				result, xferred);
+			if (!img_request->result)
+				img_request->result = result;
+		}

 		more = blk_end_request(img_request->rq, result, xferred);
 		which++;
-- 
1.7.9.5

[PATCH 02/11] rbd: record aggregate image transfer count

[Alex Elder]

Compute the total number of bytes transferred for an image
request--the sum across each of the request's object requests.
To avoid contention do it only when all object requests are
complete, in rbd_img_request_complete().

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 69eab66..e8374ae 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -214,6 +214,7 @@ struct rbd_img_request {
 	spinlock_t		completion_lock;/* protects next_completion */
 	u32			next_completion;
 	rbd_img_callback_t	callback;
+	u64			xferred;/* aggregate bytes transferred */
 	int			result;	/* first nonzero obj_request result */

 	u32			obj_request_count;
@@ -1148,7 +1149,24 @@ static int rbd_obj_request_submit(struct
ceph_osd_client *osdc,

 static void rbd_img_request_complete(struct rbd_img_request *img_request)
 {
+
 	dout("%s: img %p\n", __func__, img_request);
+
+	/*
+	 * If no error occurred, compute the aggregate transfer
+	 * count for the image request.  We could instead use
+	 * atomic64_cmpxchg() to update it as each object request
+	 * completes; not clear which way is better off hand.
+	 */
+	if (!img_request->result) {
+		struct rbd_obj_request *obj_request;
+		u64 xferred = 0;
+
+		for_each_obj_request(img_request, obj_request)
+			xferred += obj_request->xferred;
+		img_request->xferred = xferred;
+	}
+
 	if (img_request->callback)
 		img_request->callback(img_request);
 	else
-- 
1.7.9.5

[PATCH 03/11] rbd: record image-relative offset in object requests

[Alex Elder]

For an image object request we will need to know what offset within
the rbd image the request covers.  Record that when the object
request gets created.

Update the I/O error warnings so they use this so what's reported
is more informative.

Rename a local variable to fit the convention used everywhere else.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index e8374ae..f0124c5 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -176,6 +176,7 @@ struct rbd_obj_request {
 	u64			length;		/* bytes from offset */

 	struct rbd_img_request	*img_request;
+	u64			img_offset;	/* image relative offset */
 	struct list_head	links;		/* img_request->obj_requests */
 	u32			which;		/* posn image request list */

@@ -1576,8 +1577,13 @@ static void rbd_img_obj_callback(struct
rbd_obj_request *obj_request)
 		xferred = (unsigned int)obj_request->xferred;
 		result = obj_request->result;
 		if (result) {
-			rbd_warn(NULL, "obj_request %s result %d xferred %u\n",
+			struct rbd_device *rbd_dev = img_request->rbd_dev;
+
+			rbd_warn(rbd_dev, "%s %llx at %llx (%llx)\n",
 				img_request->write_request ? "write" : "read",
+				obj_request->length, obj_request->img_offset,
+				obj_request->offset);
+			rbd_warn(rbd_dev, "  result %d xferred %x\n",
 				result, xferred);
 			if (!img_request->result)
 				img_request->result = result;
@@ -1604,7 +1610,7 @@ static int rbd_img_request_fill_bio(struct
rbd_img_request *img_request,
 	struct rbd_obj_request *next_obj_request;
 	bool write_request = img_request->write_request;
 	unsigned int bio_offset;
-	u64 image_offset;
+	u64 img_offset;
 	u64 resid;
 	u16 opcode;

@@ -1612,8 +1618,8 @@ static int rbd_img_request_fill_bio(struct
rbd_img_request *img_request,

 	opcode = 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);
+	img_offset = img_request->offset;
+	rbd_assert(img_offset == bio_list->bi_sector << SECTOR_SHIFT);
 	resid = img_request->length;
 	rbd_assert(resid > 0);
 	while (resid) {
@@ -1623,11 +1629,11 @@ static int rbd_img_request_fill_bio(struct
rbd_img_request *img_request,
 		u64 offset;
 		u64 length;

-		object_name = rbd_segment_name(rbd_dev, image_offset);
+		object_name = rbd_segment_name(rbd_dev, img_offset);
 		if (!object_name)
 			goto out_unwind;
-		offset = rbd_segment_offset(rbd_dev, image_offset);
-		length = rbd_segment_length(rbd_dev, image_offset, resid);
+		offset = rbd_segment_offset(rbd_dev, img_offset);
+		length = rbd_segment_length(rbd_dev, img_offset, resid);
 		obj_request = rbd_obj_request_create(object_name,
 						offset, length,
 						OBJ_REQUEST_BIO);
@@ -1656,9 +1662,10 @@ static int rbd_img_request_fill_bio(struct
rbd_img_request *img_request,
 				obj_request->bio_list, obj_request->length);
 		rbd_osd_req_format(obj_request, write_request);

+		obj_request->img_offset = img_offset;
 		rbd_img_obj_request_add(img_request, obj_request);

-		image_offset += length;
+		img_offset += length;
 		resid -= length;
 	}

@@ -1993,8 +2000,10 @@ static void rbd_request_fn(struct request_queue *q)
 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);
+			rbd_warn(rbd_dev, "%s %llx at %llx result %d\n",
+				write_request ? "write" : "read",
+				length, offset, result);
+
 			__blk_end_request_all(rq, result);
 		}
 	}
-- 
1.7.9.5

[PATCH 04/11] rbd: define image request flags

[Alex Elder]

There are several Boolean values we'll be maintaining for image
requests.  Switch from the single write_request field to a
general-purpose flags field, and use one if its bits to represent
the direction of I/O for the image request.  Define helper functions
for setting and testing that flag.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index f0124c5..5ea2e36 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -202,12 +202,16 @@ struct rbd_obj_request {
 	struct kref		kref;
 };

+enum img_req_flags {
+	IMG_REQ_WRITE,		/* read = 0, write = 1 */
+};
+
 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 */
+	unsigned long		flags;
 	union {
 		struct ceph_snap_context *snapc;	/* for writes */
 		u64		snap_id;		/* for reads */
@@ -1210,6 +1214,23 @@ static bool obj_request_done_test(struct
rbd_obj_request *obj_request)
 	return atomic_read(&obj_request->done) != 0;
 }

+/*
+ * The default/initial value for all image request flags is 0.  Each
+ * is conditionally set to 1 at image request initialization time
+ * and currently never change thereafter.
+ */
+static void img_request_write_set(struct rbd_img_request *img_request)
+{
+	set_bit(IMG_REQ_WRITE, &img_request->flags);
+	smp_mb();
+}
+
+static bool img_request_write_test(struct rbd_img_request *img_request)
+{
+	smp_mb();
+	return test_bit(IMG_REQ_WRITE, &img_request->flags) != 0;
+}
+
 static void
 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
 {
@@ -1369,8 +1390,9 @@ static struct ceph_osd_request *rbd_osd_req_create(
 	struct ceph_osd_request *osd_req;

 	if (img_request) {
-		rbd_assert(img_request->write_request == write_request);
-		if (img_request->write_request)
+		rbd_assert(write_request ==
+				img_request_write_test(img_request));
+		if (write_request)
 			snapc = img_request->snapc;
 	}

@@ -1494,17 +1516,20 @@ static struct rbd_img_request
*rbd_img_request_create(
 			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->flags = 0;
+	if (write_request) {
+		img_request_write_set(img_request);
 		img_request->snapc = snapc;
-	else
+	} 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;
@@ -1537,7 +1562,7 @@ static void rbd_img_request_destroy(struct kref *kref)
 		rbd_img_obj_request_del(img_request, obj_request);
 	rbd_assert(img_request->obj_request_count == 0);

-	if (img_request->write_request)
+	if (img_request_write_test(img_request))
 		ceph_put_snap_context(img_request->snapc);

 	kfree(img_request);
@@ -1580,7 +1605,8 @@ static void rbd_img_obj_callback(struct
rbd_obj_request *obj_request)
 			struct rbd_device *rbd_dev = img_request->rbd_dev;

 			rbd_warn(rbd_dev, "%s %llx at %llx (%llx)\n",
-				img_request->write_request ? "write" : "read",
+				img_request_write_test(img_request) ? "write"
+								    : "read",
 				obj_request->length, obj_request->img_offset,
 				obj_request->offset);
 			rbd_warn(rbd_dev, "  result %d xferred %x\n",
@@ -1608,7 +1634,7 @@ static int rbd_img_request_fill_bio(struct
rbd_img_request *img_request,
 	struct rbd_device *rbd_dev = img_request->rbd_dev;
 	struct rbd_obj_request *obj_request = NULL;
 	struct rbd_obj_request *next_obj_request;
-	bool write_request = img_request->write_request;
+	bool write_request = img_request_write_test(img_request);
 	unsigned int bio_offset;
 	u64 img_offset;
 	u64 resid;
-- 
1.7.9.5

[PATCH 05/11] rbd: define image request originator flag

[Alex Elder]

Define a flag indicating whether an image request originated from
the Linux block layer (from blk_fetch_request()) or whether it was
initiated in order to satisfy an object request for a child image
of a layered rbd device.  For image requests initiated by objects of
child images we'll save a pointer to the object request rather than
the Linux block request.

For now, only block requests are used.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 5ea2e36..7ecd909 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -203,18 +203,22 @@ struct rbd_obj_request {
 };

 enum img_req_flags {
-	IMG_REQ_WRITE,		/* read = 0, write = 1 */
+	IMG_REQ_WRITE,		/* I/O direction: read = 0, write = 1 */
+	IMG_REQ_CHILD,		/* initiator: block = 0, child image = 1 */
 };

 struct rbd_img_request {
-	struct request		*rq;
 	struct rbd_device	*rbd_dev;
 	u64			offset;	/* starting image byte offset */
 	u64			length;	/* byte count from offset */
 	unsigned long		flags;
 	union {
+		u64			snap_id;	/* for reads */
 		struct ceph_snap_context *snapc;	/* for writes */
-		u64		snap_id;		/* for reads */
+	};
+	union {
+		struct request		*rq;		/* block request */
+		struct rbd_obj_request	*obj_request;	/* obj req initiator */
 	};
 	spinlock_t		completion_lock;/* protects next_completion */
 	u32			next_completion;
@@ -1231,6 +1235,18 @@ static bool img_request_write_test(struct
rbd_img_request *img_request)
 	return test_bit(IMG_REQ_WRITE, &img_request->flags) != 0;
 }

+static void img_request_child_set(struct rbd_img_request *img_request)
+{
+	set_bit(IMG_REQ_CHILD, &img_request->flags);
+	smp_mb();
+}
+
+static bool img_request_child_test(struct rbd_img_request *img_request)
+{
+	smp_mb();
+	return test_bit(IMG_REQ_CHILD, &img_request->flags) != 0;
+}
+
 static void
 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
 {
@@ -1499,7 +1515,8 @@ static void rbd_obj_request_destroy(struct kref *kref)
 static struct rbd_img_request *rbd_img_request_create(
 					struct rbd_device *rbd_dev,
 					u64 offset, u64 length,
-					bool write_request)
+					bool write_request,
+					bool child_request)
 {
 	struct rbd_img_request *img_request;
 	struct ceph_snap_context *snapc = NULL;
@@ -1530,6 +1547,8 @@ static struct rbd_img_request *rbd_img_request_create(
 	} else {
 		img_request->snap_id = rbd_dev->spec->snap_id;
 	}
+	if (child_request)
+		img_request_child_set(img_request);
 	spin_lock_init(&img_request->completion_lock);
 	img_request->next_completion = 0;
 	img_request->callback = NULL;
@@ -1578,7 +1597,9 @@ static void rbd_img_obj_callback(struct
rbd_obj_request *obj_request)

 	dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
 	rbd_assert(img_request != NULL);
+	rbd_assert(!img_request_child_test(img_request))
 	rbd_assert(img_request->rq != NULL);
+
 	rbd_assert(img_request->obj_request_count > 0);
 	rbd_assert(which != BAD_WHICH);
 	rbd_assert(which < img_request->obj_request_count);
@@ -2012,7 +2033,7 @@ static void rbd_request_fn(struct request_queue *q)

 		result = -ENOMEM;
 		img_request = rbd_img_request_create(rbd_dev, offset, length,
-							write_request);
+							write_request, false);
 		if (!img_request)
 			goto end_request;

-- 
1.7.9.5

[PATCH 06/11] rbd: define image request layered flag

[Alex Elder]

Define a flag indicating whether an image request is for a layered
image (one with a parent image to which requests will be redirected
if the target object of a request does not exist).  The code that
checks this flag will be added shortly.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 7ecd909..a77157d 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -205,6 +205,7 @@ struct rbd_obj_request {
 enum img_req_flags {
 	IMG_REQ_WRITE,		/* I/O direction: read = 0, write = 1 */
 	IMG_REQ_CHILD,		/* initiator: block = 0, child image = 1 */
+	IMG_REQ_LAYERED,	/* ENOENT handling: normal = 0, layered = 1 */
 };

 struct rbd_img_request {
@@ -1247,6 +1248,18 @@ static bool img_request_child_test(struct
rbd_img_request *img_request)
 	return test_bit(IMG_REQ_CHILD, &img_request->flags) != 0;
 }

+static void img_request_layered_set(struct rbd_img_request *img_request)
+{
+	set_bit(IMG_REQ_LAYERED, &img_request->flags);
+	smp_mb();
+}
+
+static bool img_request_layered_test(struct rbd_img_request *img_request)
+{
+	smp_mb();
+	return test_bit(IMG_REQ_LAYERED, &img_request->flags) != 0;
+}
+
 static void
 rbd_img_obj_request_read_callback(struct rbd_obj_request *obj_request)
 {
@@ -1549,6 +1562,8 @@ static struct rbd_img_request *rbd_img_request_create(
 	}
 	if (child_request)
 		img_request_child_set(img_request);
+	if (rbd_dev->parent_spec)
+		img_request_layered_set(img_request);
 	spin_lock_init(&img_request->completion_lock);
 	img_request->next_completion = 0;
 	img_request->callback = NULL;
@@ -1557,6 +1572,7 @@ static struct rbd_img_request *rbd_img_request_create(
 	INIT_LIST_HEAD(&img_request->obj_requests);
 	kref_init(&img_request->kref);

+	(void) img_request_layered_test(img_request);	/* Avoid a warning */
 	rbd_img_request_get(img_request);	/* Avoid a warning */
 	rbd_img_request_put(img_request);	/* TEMPORARY */

-- 
1.7.9.5

[PATCH 07/11] rbd: encapsulate image object end request handling

[Alex Elder]

Encapsulate the code that completes processing of an object request
that's part of an image request.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index a77157d..2d27115 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -1603,6 +1603,34 @@ static void rbd_img_request_destroy(struct kref
*kref)
 	kfree(img_request);
 }

+static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request)
+{
+	struct rbd_img_request *img_request = obj_request->img_request;
+	unsigned int xferred;
+	int result;
+
+	rbd_assert(!img_request_child_test(img_request));
+	rbd_assert(img_request->rq != NULL);
+
+	rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
+	xferred = (unsigned int)obj_request->xferred;
+	result = obj_request->result;
+	if (result) {
+		struct rbd_device *rbd_dev = img_request->rbd_dev;
+
+		rbd_warn(rbd_dev, "%s %llx at %llx (%llx)\n",
+			img_request_write_test(img_request) ? "write" : "read",
+			obj_request->length, obj_request->img_offset,
+			obj_request->offset);
+		rbd_warn(rbd_dev, "  result %d xferred %x\n",
+			result, xferred);
+		if (!img_request->result)
+			img_request->result = result;
+	}
+
+	return blk_end_request(img_request->rq, result, xferred);
+}
+
 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request)
 {
 	struct rbd_img_request *img_request;
@@ -1613,9 +1641,6 @@ static void rbd_img_obj_callback(struct
rbd_obj_request *obj_request)

 	dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
 	rbd_assert(img_request != NULL);
-	rbd_assert(!img_request_child_test(img_request))
-	rbd_assert(img_request->rq != NULL);
-
 	rbd_assert(img_request->obj_request_count > 0);
 	rbd_assert(which != BAD_WHICH);
 	rbd_assert(which < img_request->obj_request_count);
@@ -1626,33 +1651,12 @@ static void rbd_img_obj_callback(struct
rbd_obj_request *obj_request)
 		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 (!obj_request_done_test(obj_request))
 			break;
-
-		rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
-		xferred = (unsigned int)obj_request->xferred;
-		result = obj_request->result;
-		if (result) {
-			struct rbd_device *rbd_dev = img_request->rbd_dev;
-
-			rbd_warn(rbd_dev, "%s %llx at %llx (%llx)\n",
-				img_request_write_test(img_request) ? "write"
-								    : "read",
-				obj_request->length, obj_request->img_offset,
-				obj_request->offset);
-			rbd_warn(rbd_dev, "  result %d xferred %x\n",
-				result, xferred);
-			if (!img_request->result)
-				img_request->result = result;
-		}
-
-		more = blk_end_request(img_request->rq, result, xferred);
+		more = rbd_img_obj_end_request(obj_request);
 		which++;
 	}

-- 
1.7.9.5

[PATCH 08/11] rbd: define an rbd object request flags field

[Alex Elder]

We're going to need some more Boolean values for object requests,
so create a flags bit field and use it to record whether the request
is done.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 2d27115..f7046e9 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -170,10 +170,15 @@ enum obj_request_type {
 	OBJ_REQUEST_NODATA, OBJ_REQUEST_BIO, OBJ_REQUEST_PAGES
 };

+enum obj_req_flags {
+	OBJ_REQ_DONE,		/* completion flag: not done = 0, done = 1 */
+};
+
 struct rbd_obj_request {
 	const char		*object_name;
 	u64			offset;		/* object start byte */
 	u64			length;		/* bytes from offset */
+	unsigned long		flags;

 	struct rbd_img_request	*img_request;
 	u64			img_offset;	/* image relative offset */
@@ -194,7 +199,6 @@ struct rbd_obj_request {
 	u64			xferred;	/* bytes transferred */
 	u64			version;
 	int			result;
-	atomic_t		done;

 	rbd_obj_callback_t	callback;
 	struct completion	completion;
@@ -1072,6 +1076,29 @@ out_err:
 	return NULL;
 }

+/*
+ * The default/initial value for all object request flags is 0.  For
+ * each flag, once its value is set to 1 it is never reset to 0
+ * again.
+ */
+static void obj_request_done_set(struct rbd_obj_request *obj_request)
+{
+	if (test_and_set_bit(OBJ_REQ_DONE, &obj_request->flags)) {
+		struct rbd_img_request *img_request = obj_request->img_request;
+		struct rbd_device *rbd_dev;
+
+		rbd_dev = img_request ? img_request->rbd_dev : NULL;
+		rbd_warn(rbd_dev, "obj_request %p already marked done\n",
+			obj_request);
+	}
+}
+
+static bool obj_request_done_test(struct rbd_obj_request *obj_request)
+{
+	smp_mb();
+	return test_bit(OBJ_REQ_DONE, &obj_request->flags) != 0;
+}
+
 static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
 {
 	dout("%s: obj %p (was %d)\n", __func__, obj_request,
@@ -1192,33 +1219,6 @@ static int rbd_obj_request_wait(struct
rbd_obj_request *obj_request)
 	return wait_for_completion_interruptible(&obj_request->completion);
 }

-static void obj_request_done_init(struct rbd_obj_request *obj_request)
-{
-	atomic_set(&obj_request->done, 0);
-	smp_wmb();
-}
-
-static void obj_request_done_set(struct rbd_obj_request *obj_request)
-{
-	int done;
-
-	done = atomic_inc_return(&obj_request->done);
-	if (done > 1) {
-		struct rbd_img_request *img_request = obj_request->img_request;
-		struct rbd_device *rbd_dev;
-
-		rbd_dev = img_request ? img_request->rbd_dev : NULL;
-		rbd_warn(rbd_dev, "obj_request %p was already done\n",
-			obj_request);
-	}
-}
-
-static bool obj_request_done_test(struct rbd_obj_request *obj_request)
-{
-	smp_mb();
-	return atomic_read(&obj_request->done) != 0;
-}
-
 /*
  * The default/initial value for all image request flags is 0.  Each
  * is conditionally set to 1 at image request initialization time
@@ -1475,10 +1475,10 @@ static struct rbd_obj_request
*rbd_obj_request_create(const char *object_name,
 	obj_request->object_name = memcpy(name, object_name, size);
 	obj_request->offset = offset;
 	obj_request->length = length;
+	obj_request->flags = 0;
 	obj_request->which = BAD_WHICH;
 	obj_request->type = type;
 	INIT_LIST_HEAD(&obj_request->links);
-	obj_request_done_init(obj_request);
 	init_completion(&obj_request->completion);
 	kref_init(&obj_request->kref);

-- 
1.7.9.5

[PATCH 09/11] rbd: add an object request flag for image data objects

[Alex Elder]

Add a flag to distinguish between object requests being done on
standalone objects and requests being sent for objects representing
rbd image data (i.e., object requests that are the result of image
request).

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index f7046e9..3f162e2 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -172,6 +172,7 @@ enum obj_request_type {

 enum obj_req_flags {
 	OBJ_REQ_DONE,		/* completion flag: not done = 0, done = 1 */
+	OBJ_REQ_IMG_DATA,	/* object usage: standalone = 0, image = 1 */
 };

 struct rbd_obj_request {
@@ -1099,6 +1100,24 @@ static bool obj_request_done_test(struct
rbd_obj_request *obj_request)
 	return test_bit(OBJ_REQ_DONE, &obj_request->flags) != 0;
 }

+static void obj_request_img_data_set(struct rbd_obj_request *obj_request)
+{
+	if (test_and_set_bit(OBJ_REQ_IMG_DATA, &obj_request->flags)) {
+		struct rbd_img_request *img_request = obj_request->img_request;
+		struct rbd_device *rbd_dev;
+
+		rbd_dev = img_request ? img_request->rbd_dev : NULL;
+		rbd_warn(rbd_dev, "obj_request %p already marked img_data\n",
+			obj_request);
+	}
+}
+
+static bool obj_request_img_data_test(struct rbd_obj_request *obj_request)
+{
+	smp_mb();
+	return test_bit(OBJ_REQ_IMG_DATA, &obj_request->flags) != 0;
+}
+
 static void rbd_obj_request_get(struct rbd_obj_request *obj_request)
 {
 	dout("%s: obj %p (was %d)\n", __func__, obj_request,
@@ -1139,6 +1158,8 @@ static inline void rbd_img_obj_request_add(struct
rbd_img_request *img_request,
 	rbd_obj_request_get(obj_request);
 	obj_request->img_request = img_request;
 	obj_request->which = img_request->obj_request_count;
+	rbd_assert(!obj_request_img_data_test(obj_request));
+	obj_request_img_data_set(obj_request);
 	rbd_assert(obj_request->which != BAD_WHICH);
 	img_request->obj_request_count++;
 	list_add_tail(&obj_request->links, &img_request->obj_requests);
@@ -1158,6 +1179,7 @@ static inline void rbd_img_obj_request_del(struct
rbd_img_request *img_request,
 	img_request->obj_request_count--;
 	rbd_assert(obj_request->which == img_request->obj_request_count);
 	obj_request->which = BAD_WHICH;
+	rbd_assert(obj_request_img_data_test(obj_request));
 	rbd_assert(obj_request->img_request == img_request);
 	obj_request->img_request = NULL;
 	obj_request->callback = NULL;
@@ -1343,7 +1365,9 @@ static void rbd_osd_req_callback(struct
ceph_osd_request *osd_req,

 	dout("%s: osd_req %p msg %p\n", __func__, osd_req, msg);
 	rbd_assert(osd_req == obj_request->osd_req);
-	rbd_assert(!!obj_request->img_request ^
+	rbd_assert(obj_request_img_data_test(obj_request) ^
+				!obj_request->img_request);
+	rbd_assert(obj_request_img_data_test(obj_request) ^
 				(obj_request->which == BAD_WHICH));

 	if (osd_req->r_result < 0)
@@ -1413,12 +1437,13 @@ static struct ceph_osd_request *rbd_osd_req_create(
 					bool write_request,
 					struct rbd_obj_request *obj_request)
 {
-	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;

-	if (img_request) {
+	if (obj_request_img_data_test(obj_request)) {
+		struct rbd_img_request *img_request = obj_request->img_request;
+
 		rbd_assert(write_request ==
 				img_request_write_test(img_request));
 		if (write_request)
@@ -1605,10 +1630,13 @@ static void rbd_img_request_destroy(struct kref
*kref)

 static bool rbd_img_obj_end_request(struct rbd_obj_request *obj_request)
 {
-	struct rbd_img_request *img_request = obj_request->img_request;
+	struct rbd_img_request *img_request;
 	unsigned int xferred;
 	int result;

+	rbd_assert(obj_request_img_data_test(obj_request));
+	img_request = obj_request->img_request;
+
 	rbd_assert(!img_request_child_test(img_request));
 	rbd_assert(img_request->rq != NULL);

@@ -1637,6 +1665,7 @@ static void rbd_img_obj_callback(struct
rbd_obj_request *obj_request)
 	u32 which = obj_request->which;
 	bool more = true;

+	rbd_assert(obj_request_img_data_test(obj_request));
 	img_request = obj_request->img_request;

 	dout("%s: img %p obj %p\n", __func__, img_request, obj_request);
-- 
1.7.9.5

[PATCH 10/11] rbd: probe the parent of an image if present

[Alex Elder]

Call the probe function for the parent device if one is present.
Since we don't formally support the layering feature we won't
be using this functionality just yet.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 3f162e2..5c129c5 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -289,6 +289,7 @@ struct rbd_device {

 	struct rbd_spec		*parent_spec;
 	u64			parent_overlap;
+	struct rbd_device	*parent;

 	/* protects updating the header */
 	struct rw_semaphore     header_rwsem;
@@ -335,6 +336,7 @@ static ssize_t rbd_add(struct bus_type *bus, const
char *buf,
 		       size_t count);
 static ssize_t rbd_remove(struct bus_type *bus, const char *buf,
 			  size_t count);
+static int rbd_dev_probe(struct rbd_device *rbd_dev);

 static struct bus_attribute rbd_bus_attrs[] = {
 	__ATTR(add, S_IWUSR, NULL, rbd_add),
@@ -497,6 +499,13 @@ out_opt:
 	return ERR_PTR(ret);
 }

+static struct rbd_client *__rbd_get_client(struct rbd_client *rbdc)
+{
+	kref_get(&rbdc->kref);
+
+	return rbdc;
+}
+
 /*
  * Find a ceph client with specific addr and configuration.  If
  * found, bump its reference count.
@@ -512,7 +521,8 @@ static struct rbd_client *rbd_client_find(struct
ceph_options *ceph_opts)
 	spin_lock(&rbd_client_list_lock);
 	list_for_each_entry(client_node, &rbd_client_list, node) {
 		if (!ceph_compare_options(ceph_opts, client_node->client)) {
-			kref_get(&client_node->kref);
+			__rbd_get_client(client_node);
+
 			found = true;
 			break;
 		}
@@ -2741,8 +2751,6 @@ static struct rbd_spec *rbd_spec_alloc(void)
 		return NULL;
 	kref_init(&spec->kref);

-	rbd_spec_put(rbd_spec_get(spec));	/* TEMPORARY */
-
 	return spec;
 }

@@ -3837,6 +3845,11 @@ static int rbd_dev_image_id(struct rbd_device
*rbd_dev)
 	void *response;
 	void *p;

+	/* If we already have it we don't need to look it up */
+
+	if (rbd_dev->spec->image_id)
+		return 0;
+
 	/*
 	 * When probing a parent image, the image id is already
 	 * known (and the image name likely is not).  There's no
@@ -4014,6 +4027,9 @@ out_err:

 static int rbd_dev_probe_finish(struct rbd_device *rbd_dev)
 {
+	struct rbd_device *parent = NULL;
+	struct rbd_spec *parent_spec = NULL;
+	struct rbd_client *rbdc = NULL;
 	int ret;

 	/* no need to lock here, as rbd_dev is not registered yet */
@@ -4058,6 +4074,31 @@ static int rbd_dev_probe_finish(struct rbd_device
*rbd_dev)
 	 * At this point cleanup in the event of an error is the job
 	 * of the sysfs code (initiated by rbd_bus_del_dev()).
 	 */
+	/* Probe the parent if there is one */
+
+	if (rbd_dev->parent_spec) {
+		/*
+		 * We need to pass a reference to the client and the
+		 * parent spec when creating the parent rbd_dev.
+		 * Images related by parent/child relationships
+		 * always share both.
+		 */
+		parent_spec = rbd_spec_get(rbd_dev->parent_spec);
+		rbdc = __rbd_get_client(rbd_dev->rbd_client);
+
+		parent = rbd_dev_create(rbdc, parent_spec);
+		if (!parent) {
+			ret = -ENOMEM;
+			goto err_out_spec;
+		}
+		rbdc = NULL;		/* parent now owns reference */
+		parent_spec = NULL;	/* parent now owns reference */
+		ret = rbd_dev_probe(parent);
+		if (ret < 0)
+			goto err_out_parent;
+		rbd_dev->parent = parent;
+	}
+
 	down_write(&rbd_dev->header_rwsem);
 	ret = rbd_dev_snaps_register(rbd_dev);
 	up_write(&rbd_dev->header_rwsem);
@@ -4076,6 +4117,12 @@ static int rbd_dev_probe_finish(struct rbd_device
*rbd_dev)
 		(unsigned long long) rbd_dev->mapping.size);

 	return ret;
+
+err_out_parent:
+	rbd_dev_destroy(parent);
+err_out_spec:
+	rbd_spec_put(parent_spec);
+	rbd_put_client(rbdc);
 err_out_bus:
 	/* this will also clean up rest of rbd_dev stuff */

@@ -4239,6 +4286,12 @@ static void rbd_dev_release(struct device *dev)
 	module_put(THIS_MODULE);
 }

+static void __rbd_remove(struct rbd_device *rbd_dev)
+{
+	rbd_remove_all_snaps(rbd_dev);
+	rbd_bus_del_dev(rbd_dev);
+}
+
 static ssize_t rbd_remove(struct bus_type *bus,
 			  const char *buf,
 			  size_t count)
@@ -4274,8 +4327,26 @@ static ssize_t rbd_remove(struct bus_type *bus,
 	if (ret < 0)
 		goto done;

-	rbd_remove_all_snaps(rbd_dev);
-	rbd_bus_del_dev(rbd_dev);
+	while (rbd_dev->parent_spec) {
+		struct rbd_device *first = rbd_dev;
+		struct rbd_device *second = first->parent;
+		struct rbd_device *third;
+
+		/*
+		 * Follow to the parent with no grandparent and
+		 * remove it.
+		 */
+		while (second && (third = second->parent)) {
+			first = second;
+			second = third;
+		}
+		__rbd_remove(second);
+		rbd_spec_put(first->parent_spec);
+		first->parent_spec = NULL;
+		first->parent_overlap = 0;
+		first->parent = NULL;
+	}
+	__rbd_remove(rbd_dev);

 done:
 	mutex_unlock(&ctl_mutex);
-- 
1.7.9.5

[PATCH 11/11] rbd: implement layered reads

[Alex Elder]

Implement layered read requests for format 2 rbd images.

If an rbd image is a clone of a snapshot, the snapshot will be the
clone's "parent" image.  When an object read request on a clone
comes back with ENOENT it indicates that the clone is not yet
populated with that portion of the image's data, and the parent
image should be consulted to satisfy the read.

When this occurs, a new image request is created, directed to the
parent image.  The offset and length of the image are the same as
the image-relative offset and length of the object request that
produced ENOENT.  Data from the parent image therefore satisfies the
object read request for the original image request.

While this code works, it will not be active until we enable the
layering feature (by adding RBD_FEATURE_LAYERING to the value of
RBD_FEATURES_SUPPORTED).

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 5c129c5..13a381b 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -398,6 +398,8 @@ void rbd_warn(struct rbd_device *rbd_dev, const char
*fmt, ...)
 #  define rbd_assert(expr)	((void) 0)
 #endif /* !RBD_DEBUG */

+static void rbd_img_parent_read(struct rbd_obj_request *obj_request);
+
 static int rbd_dev_refresh(struct rbd_device *rbd_dev, u64 *hver);
 static int rbd_dev_v2_refresh(struct rbd_device *rbd_dev, u64 *hver);

@@ -1336,9 +1338,15 @@ static void rbd_osd_trivial_callback(struct
rbd_obj_request *obj_request)

 static void rbd_osd_read_callback(struct rbd_obj_request *obj_request)
 {
-	dout("%s: obj %p result %d %llu/%llu\n", __func__, obj_request,
-		obj_request->result, obj_request->xferred, obj_request->length);
-	if (obj_request->img_request)
+	struct rbd_img_request *img_request = obj_request->img_request;
+	bool layered = img_request && img_request_layered_test(img_request);
+
+	dout("%s: obj %p img %p result %d %llu/%llu\n", __func__,
+		obj_request, img_request, obj_request->result,
+		obj_request->xferred, obj_request->length);
+	if (layered && obj_request->result == -ENOENT)
+		rbd_img_parent_read(obj_request);
+	else if (img_request)
 		rbd_img_obj_request_read_callback(obj_request);
 	else
 		obj_request_done_set(obj_request);
@@ -1349,9 +1357,8 @@ static void rbd_osd_write_callback(struct
rbd_obj_request *obj_request)
 	dout("%s: obj %p result %d %llu\n", __func__, obj_request,
 		obj_request->result, obj_request->length);
 	/*
-	 * There is no such thing as a successful short write.
-	 * Our xferred value is the number of bytes transferred
-	 * back.  Set it to our originally-requested length.
+	 * There is no such thing as a successful short write.  Set
+	 * it to our originally-requested length.
 	 */
 	obj_request->xferred = obj_request->length;
 	obj_request_done_set(obj_request);
@@ -1391,7 +1398,7 @@ static void rbd_osd_req_callback(struct
ceph_osd_request *osd_req,
 	 * passed to blk_end_request(), which takes an unsigned int.
 	 */
 	obj_request->xferred = osd_req->r_reply_op_len[0];
-	rbd_assert(obj_request->xferred < (u64) UINT_MAX);
+	rbd_assert(obj_request->xferred < (u64)UINT_MAX);
 	opcode = osd_req->r_ops[0].op;
 	switch (opcode) {
 	case CEPH_OSD_OP_READ:
@@ -1607,7 +1614,6 @@ static struct rbd_img_request *rbd_img_request_create(
 	INIT_LIST_HEAD(&img_request->obj_requests);
 	kref_init(&img_request->kref);

-	(void) img_request_layered_test(img_request);	/* Avoid a warning */
 	rbd_img_request_get(img_request);	/* Avoid a warning */
 	rbd_img_request_put(img_request);	/* TEMPORARY */

@@ -1635,6 +1641,9 @@ static void rbd_img_request_destroy(struct kref *kref)
 	if (img_request_write_test(img_request))
 		ceph_put_snap_context(img_request->snapc);

+	if (img_request_child_test(img_request))
+		rbd_obj_request_put(img_request->obj_request);
+
 	kfree(img_request);
 }

@@ -1643,13 +1652,11 @@ static bool rbd_img_obj_end_request(struct
rbd_obj_request *obj_request)
 	struct rbd_img_request *img_request;
 	unsigned int xferred;
 	int result;
+	bool more;

 	rbd_assert(obj_request_img_data_test(obj_request));
 	img_request = obj_request->img_request;

-	rbd_assert(!img_request_child_test(img_request));
-	rbd_assert(img_request->rq != NULL);
-
 	rbd_assert(obj_request->xferred <= (u64)UINT_MAX);
 	xferred = (unsigned int)obj_request->xferred;
 	result = obj_request->result;
@@ -1666,7 +1673,15 @@ static bool rbd_img_obj_end_request(struct
rbd_obj_request *obj_request)
 			img_request->result = result;
 	}

-	return blk_end_request(img_request->rq, result, xferred);
+	if (img_request_child_test(img_request)) {
+		rbd_assert(img_request->obj_request != NULL);
+		more = obj_request->which < img_request->obj_request_count - 1;
+	} else {
+		rbd_assert(img_request->rq != NULL);
+		more = blk_end_request(img_request->rq, result, xferred);
+	}
+
+	return more;
 }

 static void rbd_img_obj_callback(struct rbd_obj_request *obj_request)
@@ -1811,6 +1826,64 @@ static int rbd_img_request_submit(struct
rbd_img_request *img_request)
 	return 0;
 }

+static void rbd_img_parent_read_callback(struct rbd_img_request
*img_request)
+{
+	struct rbd_obj_request *obj_request;
+
+	rbd_assert(img_request_child_test(img_request));
+
+	obj_request = img_request->obj_request;
+	rbd_assert(obj_request != NULL);
+	obj_request->result = img_request->result;
+	obj_request->xferred = img_request->xferred;
+
+	rbd_img_obj_request_read_callback(obj_request);
+	rbd_obj_request_complete(obj_request);
+}
+
+static void rbd_img_parent_read(struct rbd_obj_request *obj_request)
+{
+	struct rbd_device *rbd_dev;
+	struct rbd_img_request *img_request;
+	int result;
+
+	rbd_assert(obj_request_img_data_test(obj_request));
+	rbd_assert(obj_request->img_request != NULL);
+	rbd_assert(obj_request->result == (s32) -ENOENT);
+	rbd_assert(obj_request->type == OBJ_REQUEST_BIO);
+
+	rbd_dev = obj_request->img_request->rbd_dev;
+	rbd_assert(rbd_dev->parent != NULL);
+	/* rbd_read_finish(obj_request, obj_request->length); */
+	img_request = rbd_img_request_create(rbd_dev->parent,
+						obj_request->img_offset,
+						obj_request->length,
+						false, true);
+	result = -ENOMEM;
+	if (!img_request)
+		goto out_err;
+
+	rbd_obj_request_get(obj_request);
+	img_request->obj_request = obj_request;
+
+	result = rbd_img_request_fill_bio(img_request, obj_request->bio_list);
+	if (result)
+		goto out_err;
+
+	img_request->callback = rbd_img_parent_read_callback;
+	result = rbd_img_request_submit(img_request);
+	if (result)
+		goto out_err;
+
+	return;
+out_err:
+	if (img_request)
+		rbd_img_request_put(img_request);
+	obj_request->result = result;
+	obj_request->xferred = 0;
+	obj_request_done_set(obj_request);
+}
+
 static int rbd_obj_notify_ack(struct rbd_device *rbd_dev,
 				   u64 ver, u64 notify_id)
 {
-- 
1.7.9.5

Re: [PATCH 00/11] rbd: layered read functionality

[Josh Durgin]

On 04/11/2013 07:14 PM, Alex Elder wrote:
> This series puts in place code that is able to handle
> read requests on rbd clone images, forwarding them to
> a parent snapshot image if necessary.  Missing from this
> series is a temporary patch at the end which actually
> activates this functionality.  That will not go in
> until the rest of rbd layering functionality is in place.
>
> I'm going to restate that.  This code implements reads
> for layered rbd images, but the functionality will not
> be usable quite yet.
>
> Most of the series is adding flags to image and object
> requests, and putting in place some accounting in (parent)
> image requests so it can be used to record the completion
> of an object request that initiated it.
>
> 					-Alex
>
> [PATCH 01/11] rbd: record overall image request result
> [PATCH 02/11] rbd: record aggregate image transfer count
> [PATCH 03/11] rbd: record image-relative offset in object requests
> [PATCH 04/11] rbd: define image request flags
> [PATCH 05/11] rbd: define image request originator flag
> [PATCH 06/11] rbd: define image request layered flag
> [PATCH 07/11] rbd: encapsulate image object end request handling
> [PATCH 08/11] rbd: define an rbd object request flags field
> [PATCH 09/11] rbd: add an object request flag for image data objects
> [PATCH 10/11] rbd: probe the parent of an image if present
> [PATCH 11/11] rbd: implement layered reads

These are all pretty straight-forward since the infrastructure is in
place. They all look good.

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