[PATCH v3 0/5] fix shutdown races and snapshot error handling

[PATCH v3 0/5] fix shutdown races and snapshot error handling

[Josh Durgin]

Patches 1-3 fix races between device removal and notify processing.

Patch 2 has an improved summary, fixes reference counting, and
renames the function as suggested by Alex.

Patch 3 is a reworked and simplified version that uses the existing
rbd_dev->flags and lock instead of adding new ones. It also restricts
where it holds the lock to avoid an inversion with bdev->lock.

Patch 4 is the same.

Patch 5 fixes an inconsistency noticed by Alex in his review of patch 4.

Josh Durgin (5):
  rbd: complete notifies before cleaning up osd_client and rbd_dev
  rbd: make rbd_obj_notify_ack() synchronous
  rbd: fix use-after free of rbd_dev->disk
  rbd: ignore unmapped snapshots that no longer exist
  rbd: fix error handling from rbd_snap_name()

 drivers/block/rbd.c |   71 ++++++++++++++++++++++++++++++++++++++-------------
 1 files changed, 53 insertions(+), 18 deletions(-)

-- 
1.7.2.5

[PATCH v3 1/5] rbd: complete notifies before cleaning up osd_client and rbd_dev

[Josh Durgin]

To ensure rbd_dev is not used after it's released, flush all pending
notify callbacks before calling rbd_dev_image_release(). No new
notifies can be added to the queue at this point because the watch has
already be unregistered with the osd_client.

Signed-off-by: Josh Durgin <josh.durgin@inktank.com>
Reviewed-by: Alex Elder <elder@linaro.org>
---
 drivers/block/rbd.c |    7 +++++++
 1 files changed, 7 insertions(+), 0 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index fef3687..bf89e34 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -5163,6 +5163,13 @@ static ssize_t rbd_remove(struct bus_type *bus,
 	ret = rbd_dev_header_watch_sync(rbd_dev, false);
 	if (ret)
 		rbd_warn(rbd_dev, "failed to cancel watch event (%d)\n", ret);
+
+	/*
+	 * flush remaining watch callbacks - these must be complete
+	 * before the osd_client is shutdown
+	 */
+	dout("%s: flushing notifies", __func__);
+	ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
 	rbd_dev_image_release(rbd_dev);
 	module_put(THIS_MODULE);
 
-- 
1.7.2.5

[PATCH v3 2/5] rbd: make rbd_obj_notify_ack() synchronous

[Josh Durgin]

The only user of rbd_obj_notify_ack() is rbd_watch_cb(). It used
asynchronously with no tracking of when the notify ack completes, so
it may still be in progress when the osd_client is shut down.  This
results in a BUG() since the osd client assumes no requests are in
flight when it stops. Since all notifies are flushed before the
osd_client is stopped, waiting for the notify ack to complete before
returning from the watch callback ensures there are no notify acks in
flight during shutdown.

Rename rbd_obj_notify_ack() to rbd_obj_notify_ack_sync() to reflect
its new synchronous nature.

Signed-off-by: Josh Durgin <josh.durgin@inktank.com>
---
 drivers/block/rbd.c |   11 ++++++-----
 1 files changed, 6 insertions(+), 5 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index bf89e34..9e5f07f 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -2808,7 +2808,7 @@ out_err:
 	obj_request_done_set(obj_request);
 }
 
-static int rbd_obj_notify_ack(struct rbd_device *rbd_dev, u64 notify_id)
+static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev, u64 notify_id)
 {
 	struct rbd_obj_request *obj_request;
 	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
@@ -2823,16 +2823,17 @@ static int rbd_obj_notify_ack(struct rbd_device *rbd_dev, u64 notify_id)
 	obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
 	if (!obj_request->osd_req)
 		goto out;
-	obj_request->callback = rbd_obj_request_put;
 
 	osd_req_op_watch_init(obj_request->osd_req, 0, CEPH_OSD_OP_NOTIFY_ACK,
 					notify_id, 0, 0);
 	rbd_osd_req_format_read(obj_request);
 
 	ret = rbd_obj_request_submit(osdc, obj_request);
-out:
 	if (ret)
-		rbd_obj_request_put(obj_request);
+		goto out;
+	ret = rbd_obj_request_wait(obj_request);
+out:
+	rbd_obj_request_put(obj_request);
 
 	return ret;
 }
@@ -2852,7 +2853,7 @@ static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
 	if (ret)
 		rbd_warn(rbd_dev, "header refresh error (%d)\n", ret);
 
-	rbd_obj_notify_ack(rbd_dev, notify_id);
+	rbd_obj_notify_ack_sync(rbd_dev, notify_id);
 }
 
 /*
-- 
1.7.2.5

[PATCH v3 3/5] rbd: fix use-after free of rbd_dev->disk

[Josh Durgin]

Removing a device deallocates the disk, unschedules the watch, and
finally cleans up the rbd_dev structure. rbd_dev_refresh(), called
from the watch callback, updates the disk size and rbd_dev
structure. With no locking between them, rbd_dev_refresh() may use the
device or rbd_dev after they've been freed.

To fix this, check whether RBD_DEV_FLAG_REMOVING is set before
updating the disk size in rbd_dev_refresh(). In order to prevent a
race where rbd_dev_refresh() is already revalidating the disk when
rbd_remove() is called, move the call to rbd_bus_del_dev() after the
watch is unregistered and all notifies are complete. It's safe to
defer deleting this structure because no new requests can be submitted
once the RBD_DEV_FLAG_REMOVING is set, since the device cannot be
opened.

Fixes: http://tracker.ceph.com/issues/5636
Signed-off-by: Josh Durgin <josh.durgin@inktank.com>
---
 drivers/block/rbd.c |   34 +++++++++++++++++++++++++++-------
 1 files changed, 27 insertions(+), 7 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 9e5f07f..fe5767a 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -3325,6 +3325,31 @@ static void rbd_exists_validate(struct rbd_device *rbd_dev)
 		clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
 }
 
+static void rbd_dev_update_size(struct rbd_device *rbd_dev)
+{
+	sector_t size;
+	bool removing;
+
+	/*
+	 * Don't hold the lock while doing disk operations,
+	 * or lock ordering will conflict with the bdev mutex via:
+	 * rbd_add() -> blkdev_get() -> rbd_open()
+	 */
+	spin_lock_irq(&rbd_dev->lock);
+	removing = test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
+	spin_unlock_irq(&rbd_dev->lock);
+	/*
+	 * If the device is being removed, rbd_dev->disk has
+	 * been destroyed, so don't try to update its size
+	 */
+	if (!removing) {
+		size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
+		dout("setting size to %llu sectors", (unsigned long long)size);
+		set_capacity(rbd_dev->disk, size);
+		revalidate_disk(rbd_dev->disk);
+	}
+}
+
 static int rbd_dev_refresh(struct rbd_device *rbd_dev)
 {
 	u64 mapping_size;
@@ -3344,12 +3369,7 @@ static int rbd_dev_refresh(struct rbd_device *rbd_dev)
 	up_write(&rbd_dev->header_rwsem);
 
 	if (mapping_size != rbd_dev->mapping.size) {
-		sector_t size;
-
-		size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
-		dout("setting size to %llu sectors", (unsigned long long)size);
-		set_capacity(rbd_dev->disk, size);
-		revalidate_disk(rbd_dev->disk);
+		rbd_dev_update_size(rbd_dev);
 	}
 
 	return ret;
@@ -5160,7 +5180,6 @@ static ssize_t rbd_remove(struct bus_type *bus,
 	if (ret < 0 || already)
 		return ret;
 
-	rbd_bus_del_dev(rbd_dev);
 	ret = rbd_dev_header_watch_sync(rbd_dev, false);
 	if (ret)
 		rbd_warn(rbd_dev, "failed to cancel watch event (%d)\n", ret);
@@ -5171,6 +5190,7 @@ static ssize_t rbd_remove(struct bus_type *bus,
 	 */
 	dout("%s: flushing notifies", __func__);
 	ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
+	rbd_bus_del_dev(rbd_dev);
 	rbd_dev_image_release(rbd_dev);
 	module_put(THIS_MODULE);
 
-- 
1.7.2.5

[PATCH v3 4/5] rbd: ignore unmapped snapshots that no longer exist

[Josh Durgin]

This prevents erroring out while adding a device when a snapshot
unrelated to the current mapping is deleted between reading the
snapshot context and reading the snapshot names. If the mapped
snapshot name is not found an error still occurs as usual.

Signed-off-by: Josh Durgin <josh.durgin@inktank.com>
---
 drivers/block/rbd.c |    9 +++++++--
 1 files changed, 7 insertions(+), 2 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index fe5767a..226fa04 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -4078,8 +4078,13 @@ static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
 
 		snap_id = snapc->snaps[which];
 		snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
-		if (IS_ERR(snap_name))
-			break;
+		if (IS_ERR(snap_name)) {
+			/* ignore no-longer existing snapshots */
+			if (PTR_ERR(snap_name) == -ENOENT)
+				continue;
+			else
+				break;
+		}
 		found = !strcmp(name, snap_name);
 		kfree(snap_name);
 	}
-- 
1.7.2.5

[PATCH v3 5/5] rbd: fix error handling from rbd_snap_name()

[Josh Durgin]

rbd_snap_name() calls rbd_dev_v{1,2}_snap_name() depending on the
format of the image. The format 1 version returns NULL on error, which
is handled by the caller. The format 2 version returns an ERR_PTR,
which the caller of rbd_snap_name() does not expect.

Fortunately this is unlikely to occur in practice because
rbd_snap_id_by_name() is called before rbd_snap_name(). This would hit
similar errors to rbd_snap_name() (like the snapshot not existing) and
return early, so rbd_snap_name() would not hit an error unless the
snapshot was removed between the two calls or memory was exhausted.

Use an ERR_PTR in rbd_dev_v1_snap_name() so that the specific error
can be propagated, and it is consistent with rbd_dev_v2_snap_name().
Handle the ERR_PTR in the only rbd_snap_name() caller.

Suggested-by: Alex Elder <alex.elder@linaro.org>
Signed-off-by: Josh Durgin <josh.durgin@inktank.com>
---
 drivers/block/rbd.c |   10 ++++++----
 1 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 226fa04..16384b7 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -927,12 +927,14 @@ static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
 					u64 snap_id)
 {
 	u32 which;
+	const char *snap_name;
 
 	which = rbd_dev_snap_index(rbd_dev, snap_id);
 	if (which == BAD_SNAP_INDEX)
-		return NULL;
+		return ERR_PTR(-ENOENT);
 
-	return _rbd_dev_v1_snap_name(rbd_dev, which);
+	snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
+	return snap_name ? snap_name : ERR_PTR(-ENOMEM);
 }
 
 static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
@@ -4163,8 +4165,8 @@ static int rbd_dev_spec_update(struct rbd_device *rbd_dev)
 	/* Look up the snapshot name, and make a copy */
 
 	snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
-	if (!snap_name) {
-		ret = -ENOMEM;
+	if (IS_ERR(snap_name)) {
+		ret = PTR_ERR(snap_name);
 		goto out_err;
 	}
 
-- 
1.7.2.5

Re: [PATCH v3 2/5] rbd: make rbd_obj_notify_ack() synchronous

[Alex Elder]

On 09/09/2013 02:16 AM, Josh Durgin wrote:
> The only user of rbd_obj_notify_ack() is rbd_watch_cb(). It used
> asynchronously with no tracking of when the notify ack completes, so
> it may still be in progress when the osd_client is shut down.  This
> results in a BUG() since the osd client assumes no requests are in
> flight when it stops. Since all notifies are flushed before the
> osd_client is stopped, waiting for the notify ack to complete before
> returning from the watch callback ensures there are no notify acks in
> flight during shutdown.
> 
> Rename rbd_obj_notify_ack() to rbd_obj_notify_ack_sync() to reflect
> its new synchronous nature.
> 
> Signed-off-by: Josh Durgin <josh.durgin@inktank.com>

Looks great.  Nice description.

Reviewed-by: Alex Elder <elder@linaro.org>

> ---
>  drivers/block/rbd.c |   11 ++++++-----
>  1 files changed, 6 insertions(+), 5 deletions(-)
> 
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index bf89e34..9e5f07f 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -2808,7 +2808,7 @@ out_err:
>  	obj_request_done_set(obj_request);
>  }
>  
> -static int rbd_obj_notify_ack(struct rbd_device *rbd_dev, u64 notify_id)
> +static int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev, u64 notify_id)
>  {
>  	struct rbd_obj_request *obj_request;
>  	struct ceph_osd_client *osdc = &rbd_dev->rbd_client->client->osdc;
> @@ -2823,16 +2823,17 @@ static int rbd_obj_notify_ack(struct rbd_device *rbd_dev, u64 notify_id)
>  	obj_request->osd_req = rbd_osd_req_create(rbd_dev, false, obj_request);
>  	if (!obj_request->osd_req)
>  		goto out;
> -	obj_request->callback = rbd_obj_request_put;
>  
>  	osd_req_op_watch_init(obj_request->osd_req, 0, CEPH_OSD_OP_NOTIFY_ACK,
>  					notify_id, 0, 0);
>  	rbd_osd_req_format_read(obj_request);
>  
>  	ret = rbd_obj_request_submit(osdc, obj_request);
> -out:
>  	if (ret)
> -		rbd_obj_request_put(obj_request);
> +		goto out;
> +	ret = rbd_obj_request_wait(obj_request);
> +out:
> +	rbd_obj_request_put(obj_request);
>  
>  	return ret;
>  }
> @@ -2852,7 +2853,7 @@ static void rbd_watch_cb(u64 ver, u64 notify_id, u8 opcode, void *data)
>  	if (ret)
>  		rbd_warn(rbd_dev, "header refresh error (%d)\n", ret);
>  
> -	rbd_obj_notify_ack(rbd_dev, notify_id);
> +	rbd_obj_notify_ack_sync(rbd_dev, notify_id);
>  }
>  
>  /*
> 

Re: [PATCH v3 3/5] rbd: fix use-after free of rbd_dev->disk

[Alex Elder]

On 09/09/2013 02:17 AM, Josh Durgin wrote:
> Removing a device deallocates the disk, unschedules the watch, and
> finally cleans up the rbd_dev structure. rbd_dev_refresh(), called
> from the watch callback, updates the disk size and rbd_dev
> structure. With no locking between them, rbd_dev_refresh() may use the
> device or rbd_dev after they've been freed.
> 
> To fix this, check whether RBD_DEV_FLAG_REMOVING is set before
> updating the disk size in rbd_dev_refresh(). In order to prevent a
> race where rbd_dev_refresh() is already revalidating the disk when
> rbd_remove() is called, move the call to rbd_bus_del_dev() after the
> watch is unregistered and all notifies are complete. It's safe to
> defer deleting this structure because no new requests can be submitted
> once the RBD_DEV_FLAG_REMOVING is set, since the device cannot be
> opened.
> 
> Fixes: http://tracker.ceph.com/issues/5636
> Signed-off-by: Josh Durgin <josh.durgin@inktank.com>

I'm really sorry but I think I still see a problem.  This stuff
is very tricky...

> ---
>  drivers/block/rbd.c |   34 +++++++++++++++++++++++++++-------
>  1 files changed, 27 insertions(+), 7 deletions(-)
> 
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 9e5f07f..fe5767a 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -3325,6 +3325,31 @@ static void rbd_exists_validate(struct rbd_device *rbd_dev)
>  		clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
>  }
>  
> +static void rbd_dev_update_size(struct rbd_device *rbd_dev)
> +{
> +	sector_t size;
> +	bool removing;
> +
> +	/*
> +	 * Don't hold the lock while doing disk operations,
> +	 * or lock ordering will conflict with the bdev mutex via:
> +	 * rbd_add() -> blkdev_get() -> rbd_open()
> +	 */
> +	spin_lock_irq(&rbd_dev->lock);
> +	removing = test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
> +	spin_unlock_irq(&rbd_dev->lock);

(This is not the problem I was referring to, but thinking about
it led me to the real issue.  This is just some explanation
behind the logic, FYI.)

It is safe to do so, of course, but it's not necessary to hold
the lock for just testing the bit.  test_bit() (and clear_bit(),
etc.) are all atomic operations.  The only reason the lock
is held in rbd_open() is so that testing the bit value and
updating the open count are done together, atomically.  (The
same basic reasoning applies in rbd_remove()).

It may be necessary, however, to do a (read or full) memory
barrier before calling test_bit() if you're not using the lock.
(My history with memory barriers in this code is a bit checkered
though.  Looking at it again now I think I still got it wrong.)

> +	/*
> +	 * If the device is being removed, rbd_dev->disk has
> +	 * been destroyed, so don't try to update its size
> +	 */
> +	if (!removing) {

Here's the problem.  It is the same one faced by the open path.

You determined above whether or not the device was getting removed.
But you haven't left any state that indicates that the image is
getting refreshed (or more specifically, getting its size updated).

So, if the device is mapped but isn't actually opened by anybody
there's nothing stopping the code in rbd_remove() from going
ahead anyway.  So the possibility of the structures getting freed
remains (though the window is a little smaller now).

I think you can resolve this problem by using the open count.
That is, use the same interlock between the open count and the
removing flag that's used for rbd_open() and rbd_remove().
The open count in some sense represents someone still needing
the data structures for the image.  So treat the refresh activity
as one of those "somebodies" by claiming one of those opens
while the size update is going on.

If you encapsulated some code now in place for this purpose,
something like the following might be helpful.  (I've renamed
"open_count" to be "inuse_count" here.)

static inline int rbd_request_access(struct rbd_device *rbd_dev)
{
	int ret = 0;

	spin_lock_irq(&rbd_dev->lock);
	if (!test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
		rbd_dev->inuse_count++;
	else
		ret = -ENOENT;
	spin_unlock_irq(&rbd_dev->lock);

	return ret;
}

static inline int rbd_release_access(struct rbd_device *rbd_dev)
{
	int ret = 0;

	spin_lock_irq(&rbd_dev->lock);
	if (rbd_dev->inuse_count)
		rbd_dev->inuse_count--;
	else
		ret = -EINVAL;
	spin_unlock_irq(&rbd_dev->lock);

	return ret;
}

static inline int rbd_prevent_access(struct rbd_device *rbd_dev)
{
	int ret = 0;

	spin_lock_irq(&rbd_dev->lock);
	if (rbd_dev->inuse_count)
		ret = -EBUSY;
	else if (test_and_set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
		ret = -EINPROGRESS;
	spin_unlock_irq(&rbd_dev->lock);

	return ret;
}

I'll assume you know what I'm talking about at this point and
won't go into exactly where and how you'd use these.

> +		size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
> +		dout("setting size to %llu sectors", (unsigned long long)size);
> +		set_capacity(rbd_dev->disk, size);
> +		revalidate_disk(rbd_dev->disk);
> +	}
> +}+-
> +
>  static int rbd_dev_refresh(struct rbd_device *rbd_dev)
>  {
>  	u64 mapping_size;

Everything below is good.

> @@ -3344,12 +3369,7 @@ static int rbd_dev_refresh(struct rbd_device *rbd_dev)
>  	up_write(&rbd_dev->header_rwsem);
>  
>  	if (mapping_size != rbd_dev->mapping.size) {
> -		sector_t size;
> -
> -		size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
> -		dout("setting size to %llu sectors", (unsigned long long)size);
> -		set_capacity(rbd_dev->disk, size);
> -		revalidate_disk(rbd_dev->disk);
> +		rbd_dev_update_size(rbd_dev);
>  	}
>  
>  	return ret;
> @@ -5160,7 +5180,6 @@ static ssize_t rbd_remove(struct bus_type *bus,
>  	if (ret < 0 || already)
>  		return ret;
>  
> -	rbd_bus_del_dev(rbd_dev);
>  	ret = rbd_dev_header_watch_sync(rbd_dev, false);
>  	if (ret)
>  		rbd_warn(rbd_dev, "failed to cancel watch event (%d)\n", ret);
> @@ -5171,6 +5190,7 @@ static ssize_t rbd_remove(struct bus_type *bus,
>  	 */
>  	dout("%s: flushing notifies", __func__);
>  	ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
> +	rbd_bus_del_dev(rbd_dev);
>  	rbd_dev_image_release(rbd_dev);
>  	module_put(THIS_MODULE);
>  
> 

Re: [PATCH v3 4/5] rbd: ignore unmapped snapshots that no longer exist

[Alex Elder]

On 09/09/2013 02:17 AM, Josh Durgin wrote:
> This prevents erroring out while adding a device when a snapshot
> unrelated to the current mapping is deleted between reading the
> snapshot context and reading the snapshot names. If the mapped
> snapshot name is not found an error still occurs as usual.

This looks good.

In fact, you could even go as far as ignoring errors
altogether.  This function is determining the name associated
with a snapshot id, and doubles as an existence check.  We
need to look at the whole list to know that the snapshot
id doesn't exist, so errors along the way are just a distraction.

(But don't bother changing it, what you have is good.)

Reviewed-by: Alex Elder <elder@linaro.org>

> Signed-off-by: Josh Durgin <josh.durgin@inktank.com>
> ---
>  drivers/block/rbd.c |    9 +++++++--
>  1 files changed, 7 insertions(+), 2 deletions(-)
> 
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index fe5767a..226fa04 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -4078,8 +4078,13 @@ static u64 rbd_v2_snap_id_by_name(struct rbd_device *rbd_dev, const char *name)
>  
>  		snap_id = snapc->snaps[which];
>  		snap_name = rbd_dev_v2_snap_name(rbd_dev, snap_id);
> -		if (IS_ERR(snap_name))
> -			break;
> +		if (IS_ERR(snap_name)) {
> +			/* ignore no-longer existing snapshots */
> +			if (PTR_ERR(snap_name) == -ENOENT)
> +				continue;
> +			else
> +				break;
> +		}
>  		found = !strcmp(name, snap_name);
>  		kfree(snap_name);
>  	}
> 

Re: [PATCH v3 5/5] rbd: fix error handling from rbd_snap_name()

[Alex Elder]

On 09/09/2013 02:17 AM, Josh Durgin wrote:
> rbd_snap_name() calls rbd_dev_v{1,2}_snap_name() depending on the
> format of the image. The format 1 version returns NULL on error, which
> is handled by the caller. The format 2 version returns an ERR_PTR,
> which the caller of rbd_snap_name() does not expect.
> 
> Fortunately this is unlikely to occur in practice because
> rbd_snap_id_by_name() is called before rbd_snap_name(). This would hit
> similar errors to rbd_snap_name() (like the snapshot not existing) and
> return early, so rbd_snap_name() would not hit an error unless the
> snapshot was removed between the two calls or memory was exhausted.
> 
> Use an ERR_PTR in rbd_dev_v1_snap_name() so that the specific error
> can be propagated, and it is consistent with rbd_dev_v2_snap_name().
> Handle the ERR_PTR in the only rbd_snap_name() caller.

Looks good.

Reviewed-by: Alex Elder <elder@linaro.org>

> Suggested-by: Alex Elder <alex.elder@linaro.org>
> Signed-off-by: Josh Durgin <josh.durgin@inktank.com>
> ---
>  drivers/block/rbd.c |   10 ++++++----
>  1 files changed, 6 insertions(+), 4 deletions(-)
> 
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 226fa04..16384b7 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -927,12 +927,14 @@ static const char *rbd_dev_v1_snap_name(struct rbd_device *rbd_dev,
>  					u64 snap_id)
>  {
>  	u32 which;
> +	const char *snap_name;
>  
>  	which = rbd_dev_snap_index(rbd_dev, snap_id);
>  	if (which == BAD_SNAP_INDEX)
> -		return NULL;
> +		return ERR_PTR(-ENOENT);
>  
> -	return _rbd_dev_v1_snap_name(rbd_dev, which);
> +	snap_name = _rbd_dev_v1_snap_name(rbd_dev, which);
> +	return snap_name ? snap_name : ERR_PTR(-ENOMEM);
>  }
>  
>  static const char *rbd_snap_name(struct rbd_device *rbd_dev, u64 snap_id)
> @@ -4163,8 +4165,8 @@ static int rbd_dev_spec_update(struct rbd_device *rbd_dev)
>  	/* Look up the snapshot name, and make a copy */
>  
>  	snap_name = rbd_snap_name(rbd_dev, spec->snap_id);
> -	if (!snap_name) {
> -		ret = -ENOMEM;
> +	if (IS_ERR(snap_name)) {
> +		ret = PTR_ERR(snap_name);
>  		goto out_err;
>  	}
>  
> 

Re: [PATCH v3 0/5] fix shutdown races and snapshot error handling

[Alex Elder]

On 09/09/2013 02:16 AM, Josh Durgin wrote:
> Patches 1-3 fix races between device removal and notify processing.
> 
> Patch 2 has an improved summary, fixes reference counting, and
> renames the function as suggested by Alex.
> 
> Patch 3 is a reworked and simplified version that uses the existing
> rbd_dev->flags and lock instead of adding new ones. It also restricts
> where it holds the lock to avoid an inversion with bdev->lock.

I found all but this one were ready to go.

If you can, please just pull this patch out to be resolved
separately.  There's no sense in holding up the other four
until this one is perfect.

Nice work.

					-Alex

> Patch 4 is the same.
> 
> Patch 5 fixes an inconsistency noticed by Alex in his review of patch 4.
> 
> Josh Durgin (5):
>   rbd: complete notifies before cleaning up osd_client and rbd_dev
>   rbd: make rbd_obj_notify_ack() synchronous
>   rbd: fix use-after free of rbd_dev->disk
>   rbd: ignore unmapped snapshots that no longer exist
>   rbd: fix error handling from rbd_snap_name()
> 
>  drivers/block/rbd.c |   71 ++++++++++++++++++++++++++++++++++++++-------------
>  1 files changed, 53 insertions(+), 18 deletions(-)
> 

Re: [PATCH v3 3/5] rbd: fix use-after free of rbd_dev->disk

[Josh Durgin]

On 09/09/2013 06:37 AM, Alex Elder wrote:
> On 09/09/2013 02:17 AM, Josh Durgin wrote:
>> Removing a device deallocates the disk, unschedules the watch, and
>> finally cleans up the rbd_dev structure. rbd_dev_refresh(), called
>> from the watch callback, updates the disk size and rbd_dev
>> structure. With no locking between them, rbd_dev_refresh() may use the
>> device or rbd_dev after they've been freed.
>>
>> To fix this, check whether RBD_DEV_FLAG_REMOVING is set before
>> updating the disk size in rbd_dev_refresh(). In order to prevent a
>> race where rbd_dev_refresh() is already revalidating the disk when
>> rbd_remove() is called, move the call to rbd_bus_del_dev() after the
>> watch is unregistered and all notifies are complete. It's safe to
>> defer deleting this structure because no new requests can be submitted
>> once the RBD_DEV_FLAG_REMOVING is set, since the device cannot be
>> opened.
>>
>> Fixes: http://tracker.ceph.com/issues/5636
>> Signed-off-by: Josh Durgin <josh.durgin@inktank.com>
>
> I'm really sorry but I think I still see a problem.  This stuff
> is very tricky...
>
>> ---
>>   drivers/block/rbd.c |   34 +++++++++++++++++++++++++++-------
>>   1 files changed, 27 insertions(+), 7 deletions(-)
>>
>> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
>> index 9e5f07f..fe5767a 100644
>> --- a/drivers/block/rbd.c
>> +++ b/drivers/block/rbd.c
>> @@ -3325,6 +3325,31 @@ static void rbd_exists_validate(struct rbd_device *rbd_dev)
>>   		clear_bit(RBD_DEV_FLAG_EXISTS, &rbd_dev->flags);
>>   }
>>
>> +static void rbd_dev_update_size(struct rbd_device *rbd_dev)
>> +{
>> +	sector_t size;
>> +	bool removing;
>> +
>> +	/*
>> +	 * Don't hold the lock while doing disk operations,
>> +	 * or lock ordering will conflict with the bdev mutex via:
>> +	 * rbd_add() -> blkdev_get() -> rbd_open()
>> +	 */
>> +	spin_lock_irq(&rbd_dev->lock);
>> +	removing = test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
>> +	spin_unlock_irq(&rbd_dev->lock);
>
> (This is not the problem I was referring to, but thinking about
> it led me to the real issue.  This is just some explanation
> behind the logic, FYI.)
>
> It is safe to do so, of course, but it's not necessary to hold
> the lock for just testing the bit.  test_bit() (and clear_bit(),
> etc.) are all atomic operations.  The only reason the lock
> is held in rbd_open() is so that testing the bit value and
> updating the open count are done together, atomically.  (The
> same basic reasoning applies in rbd_remove()).
>
> It may be necessary, however, to do a (read or full) memory
> barrier before calling test_bit() if you're not using the lock.
> (My history with memory barriers in this code is a bit checkered
> though.  Looking at it again now I think I still got it wrong.)
>
>> +	/*
>> +	 * If the device is being removed, rbd_dev->disk has
>> +	 * been destroyed, so don't try to update its size
>> +	 */
>> +	if (!removing) {
>
> Here's the problem.  It is the same one faced by the open path.
>
> You determined above whether or not the device was getting removed.
> But you haven't left any state that indicates that the image is
> getting refreshed (or more specifically, getting its size updated).
>
> So, if the device is mapped but isn't actually opened by anybody
> there's nothing stopping the code in rbd_remove() from going
> ahead anyway.  So the possibility of the structures getting freed
> remains (though the window is a little smaller now).

I think this is safe with the use of ceph_osdc_flush_notifies()
before rbd_bus_del_dev(). Since the watch is already unregistered,
flushing the notifies waits for all calls to rbd_watch_cb() to
complete. This means that rbd_remove() can't actually free
any of the rbd_dev structures until after the last revalidate_disk()
etc. is done. Does this make sense, or am I missing something?

This should probably go in a comment in rbd_remove(), since
it's not obvious from the code there why the ordering of
the last few calls makes sense.

Josh

> I think you can resolve this problem by using the open count.
> That is, use the same interlock between the open count and the
> removing flag that's used for rbd_open() and rbd_remove().
> The open count in some sense represents someone still needing
> the data structures for the image.  So treat the refresh activity
> as one of those "somebodies" by claiming one of those opens
> while the size update is going on.
>
> If you encapsulated some code now in place for this purpose,
> something like the following might be helpful.  (I've renamed
> "open_count" to be "inuse_count" here.)
>
> static inline int rbd_request_access(struct rbd_device *rbd_dev)
> {
> 	int ret = 0;
>
> 	spin_lock_irq(&rbd_dev->lock);
> 	if (!test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
> 		rbd_dev->inuse_count++;
> 	else
> 		ret = -ENOENT;
> 	spin_unlock_irq(&rbd_dev->lock);
>
> 	return ret;
> }
>
> static inline int rbd_release_access(struct rbd_device *rbd_dev)
> {
> 	int ret = 0;
>
> 	spin_lock_irq(&rbd_dev->lock);
> 	if (rbd_dev->inuse_count)
> 		rbd_dev->inuse_count--;
> 	else
> 		ret = -EINVAL;
> 	spin_unlock_irq(&rbd_dev->lock);
>
> 	return ret;
> }
>
> static inline int rbd_prevent_access(struct rbd_device *rbd_dev)
> {
> 	int ret = 0;
>
> 	spin_lock_irq(&rbd_dev->lock);
> 	if (rbd_dev->inuse_count)
> 		ret = -EBUSY;
> 	else if (test_and_set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
> 		ret = -EINPROGRESS;
> 	spin_unlock_irq(&rbd_dev->lock);
>
> 	return ret;
> }
>
> I'll assume you know what I'm talking about at this point and
> won't go into exactly where and how you'd use these.
>
>> +		size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
>> +		dout("setting size to %llu sectors", (unsigned long long)size);
>> +		set_capacity(rbd_dev->disk, size);
>> +		revalidate_disk(rbd_dev->disk);
>> +	}
>> +}+-
>> +
>>   static int rbd_dev_refresh(struct rbd_device *rbd_dev)
>>   {
>>   	u64 mapping_size;
>
> Everything below is good.
>
>> @@ -3344,12 +3369,7 @@ static int rbd_dev_refresh(struct rbd_device *rbd_dev)
>>   	up_write(&rbd_dev->header_rwsem);
>>
>>   	if (mapping_size != rbd_dev->mapping.size) {
>> -		sector_t size;
>> -
>> -		size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
>> -		dout("setting size to %llu sectors", (unsigned long long)size);
>> -		set_capacity(rbd_dev->disk, size);
>> -		revalidate_disk(rbd_dev->disk);
>> +		rbd_dev_update_size(rbd_dev);
>>   	}
>>
>>   	return ret;
>> @@ -5160,7 +5180,6 @@ static ssize_t rbd_remove(struct bus_type *bus,
>>   	if (ret < 0 || already)
>>   		return ret;
>>
>> -	rbd_bus_del_dev(rbd_dev);
>>   	ret = rbd_dev_header_watch_sync(rbd_dev, false);
>>   	if (ret)
>>   		rbd_warn(rbd_dev, "failed to cancel watch event (%d)\n", ret);
>> @@ -5171,6 +5190,7 @@ static ssize_t rbd_remove(struct bus_type *bus,
>>   	 */
>>   	dout("%s: flushing notifies", __func__);
>>   	ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
>> +	rbd_bus_del_dev(rbd_dev);
>>   	rbd_dev_image_release(rbd_dev);
>>   	module_put(THIS_MODULE);
>>
>>
>

Re: [PATCH v3 3/5] rbd: fix use-after free of rbd_dev->disk

[Alex Elder]

On 09/09/2013 11:15 AM, Josh Durgin wrote:
> On 09/09/2013 06:37 AM, Alex Elder wrote:
>> On 09/09/2013 02:17 AM, Josh Durgin wrote:
>>> Removing a device deallocates the disk, unschedules the watch, and
>>> finally cleans up the rbd_dev structure. rbd_dev_refresh(), called
>>> from the watch callback, updates the disk size and rbd_dev
>>> structure. With no locking between them, rbd_dev_refresh() may use the
>>> device or rbd_dev after they've been freed.

. . .

>>> +    if (!removing) {
>>
>> Here's the problem.  It is the same one faced by the open path.
>>
>> You determined above whether or not the device was getting removed.
>> But you haven't left any state that indicates that the image is
>> getting refreshed (or more specifically, getting its size updated).
>>
>> So, if the device is mapped but isn't actually opened by anybody
>> there's nothing stopping the code in rbd_remove() from going
>> ahead anyway.  So the possibility of the structures getting freed
>> remains (though the window is a little smaller now).
> 
> I think this is safe with the use of ceph_osdc_flush_notifies()
> before rbd_bus_del_dev(). Since the watch is already unregistered,
> flushing the notifies waits for all calls to rbd_watch_cb() to
> complete. This means that rbd_remove() can't actually free
> any of the rbd_dev structures until after the last revalidate_disk()
> etc. is done. Does this make sense, or am I missing something?

Yes, now I remember.  You're right.  That was the point of
doing the flush_notifies call before release anyway...

So my "nothing stopping the code in rbd_remove" was wrong.
In fact, that ceph_osdc_flush_notifies() call is stopping
it from freeing things that may still be in use.

Thanks for explaining it to me (again).

> This should probably go in a comment in rbd_remove(), since
> it's not obvious from the code there why the ordering of
> the last few calls makes sense.

Yes.  I think so.

Reviewed-by: Alex Elder <elder@linaro.org>

> Josh
> 
>> I think you can resolve this problem by using the open count.
>> That is, use the same interlock between the open count and the
>> removing flag that's used for rbd_open() and rbd_remove().
>> The open count in some sense represents someone still needing
>> the data structures for the image.  So treat the refresh activity
>> as one of those "somebodies" by claiming one of those opens
>> while the size update is going on.
>>
>> If you encapsulated some code now in place for this purpose,
>> something like the following might be helpful.  (I've renamed
>> "open_count" to be "inuse_count" here.)
>>
>> static inline int rbd_request_access(struct rbd_device *rbd_dev)
>> {
>>     int ret = 0;
>>
>>     spin_lock_irq(&rbd_dev->lock);
>>     if (!test_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
>>         rbd_dev->inuse_count++;
>>     else
>>         ret = -ENOENT;
>>     spin_unlock_irq(&rbd_dev->lock);
>>
>>     return ret;
>> }
>>
>> static inline int rbd_release_access(struct rbd_device *rbd_dev)
>> {
>>     int ret = 0;
>>
>>     spin_lock_irq(&rbd_dev->lock);
>>     if (rbd_dev->inuse_count)
>>         rbd_dev->inuse_count--;
>>     else
>>         ret = -EINVAL;
>>     spin_unlock_irq(&rbd_dev->lock);
>>
>>     return ret;
>> }
>>
>> static inline int rbd_prevent_access(struct rbd_device *rbd_dev)
>> {
>>     int ret = 0;
>>
>>     spin_lock_irq(&rbd_dev->lock);
>>     if (rbd_dev->inuse_count)
>>         ret = -EBUSY;
>>     else if (test_and_set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags))
>>         ret = -EINPROGRESS;
>>     spin_unlock_irq(&rbd_dev->lock);
>>
>>     return ret;
>> }
>>
>> I'll assume you know what I'm talking about at this point and
>> won't go into exactly where and how you'd use these.
>>
>>> +        size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
>>> +        dout("setting size to %llu sectors", (unsigned long long)size);
>>> +        set_capacity(rbd_dev->disk, size);
>>> +        revalidate_disk(rbd_dev->disk);
>>> +    }
>>> +}+-
>>> +
>>>   static int rbd_dev_refresh(struct rbd_device *rbd_dev)
>>>   {
>>>       u64 mapping_size;
>>
>> Everything below is good.
>>
>>> @@ -3344,12 +3369,7 @@ static int rbd_dev_refresh(struct rbd_device
>>> *rbd_dev)
>>>       up_write(&rbd_dev->header_rwsem);
>>>
>>>       if (mapping_size != rbd_dev->mapping.size) {
>>> -        sector_t size;
>>> -
>>> -        size = (sector_t)rbd_dev->mapping.size / SECTOR_SIZE;
>>> -        dout("setting size to %llu sectors", (unsigned long long)size);
>>> -        set_capacity(rbd_dev->disk, size);
>>> -        revalidate_disk(rbd_dev->disk);
>>> +        rbd_dev_update_size(rbd_dev);
>>>       }
>>>
>>>       return ret;
>>> @@ -5160,7 +5180,6 @@ static ssize_t rbd_remove(struct bus_type *bus,
>>>       if (ret < 0 || already)
>>>           return ret;
>>>
>>> -    rbd_bus_del_dev(rbd_dev);
>>>       ret = rbd_dev_header_watch_sync(rbd_dev, false);
>>>       if (ret)
>>>           rbd_warn(rbd_dev, "failed to cancel watch event (%d)\n", ret);
>>> @@ -5171,6 +5190,7 @@ static ssize_t rbd_remove(struct bus_type *bus,
>>>        */
>>>       dout("%s: flushing notifies", __func__);
>>>       ceph_osdc_flush_notifies(&rbd_dev->rbd_client->client->osdc);
>>> +    rbd_bus_del_dev(rbd_dev);
>>>       rbd_dev_image_release(rbd_dev);
>>>       module_put(THIS_MODULE);
>>>
>>>
>>
>