[PATCH 0/5] rbd: clean up use of ctl_mutex

[PATCH 0/5] rbd: clean up use of ctl_mutex

[Alex Elder]

This series eliminates most uses of the ctl_mutex for locking.
Its precise purpose was always a little fuzzy to me; with this
series in place it is now only used to protect looking up
ceph clients.  It also eliminates some lockdep problems
reported at unmap time.

					-Alex

[PATCH 1/5] rbd: set removing flag while holding list lock
[PATCH 2/5] rbd: protect against concurrent unmaps
[PATCH 3/5] rbd: don't hold ctl_mutex to get/put device
[PATCH 4/5] rbd: use rwsem to protect header updates
[PATCH 5/5] rbd: take a little credit

[PATCH 1/5] rbd: set removing flag while holding list lock

[Alex Elder]

When unmapping a device, its id is supplied, and that is used to
look up which rbd device should be unmapped.  Looking up the
device involves searching the rbd device list while holding
a spinlock that protects access to that list.

Currently all of this is done under protection of the control lock,
but that protection is going away soon.  To ensure the rbd_dev is
still valid (still on the list) while setting its REMOVING flag, do
so while still holding the list lock.  To do so, get rid of
__rbd_get_dev(), and open code what it did in the one place it
was used.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index aace658..716ef1f 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -5106,23 +5106,6 @@ err_out_module:
 	return (ssize_t)rc;
 }

-static struct rbd_device *__rbd_get_dev(unsigned long dev_id)
-{
-	struct list_head *tmp;
-	struct rbd_device *rbd_dev;
-
-	spin_lock(&rbd_dev_list_lock);
-	list_for_each(tmp, &rbd_dev_list) {
-		rbd_dev = list_entry(tmp, struct rbd_device, node);
-		if (rbd_dev->dev_id == dev_id) {
-			spin_unlock(&rbd_dev_list_lock);
-			return rbd_dev;
-		}
-	}
-	spin_unlock(&rbd_dev_list_lock);
-	return NULL;
-}
-
 static void rbd_dev_device_release(struct device *dev)
 {
 	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
@@ -5167,7 +5150,8 @@ static ssize_t rbd_remove(struct bus_type *bus,
 			  size_t count)
 {
 	struct rbd_device *rbd_dev = NULL;
-	int target_id;
+	struct list_head *tmp;
+	int dev_id;
 	unsigned long ul;
 	int ret;

@@ -5176,26 +5160,33 @@ static ssize_t rbd_remove(struct bus_type *bus,
 		return ret;

 	/* convert to int; abort if we lost anything in the conversion */
-	target_id = (int) ul;
-	if (target_id != ul)
+	dev_id = (int)ul;
+	if (dev_id != ul)
 		return -EINVAL;

 	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);

-	rbd_dev = __rbd_get_dev(target_id);
-	if (!rbd_dev) {
-		ret = -ENOENT;
-		goto done;
+	ret = -ENOENT;
+	spin_lock(&rbd_dev_list_lock);
+	list_for_each(tmp, &rbd_dev_list) {
+		rbd_dev = list_entry(tmp, struct rbd_device, node);
+		if (rbd_dev->dev_id == dev_id) {
+			ret = 0;
+			break;
+		}
 	}
-
-	spin_lock_irq(&rbd_dev->lock);
-	if (rbd_dev->open_count)
-		ret = -EBUSY;
-	else
-		set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
-	spin_unlock_irq(&rbd_dev->lock);
+	if (!ret) {
+		spin_lock_irq(&rbd_dev->lock);
+		if (rbd_dev->open_count)
+			ret = -EBUSY;
+		else
+			set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
+		spin_unlock_irq(&rbd_dev->lock);
+	}
+	spin_unlock(&rbd_dev_list_lock);
 	if (ret < 0)
 		goto done;
+
 	rbd_bus_del_dev(rbd_dev);
 	ret = rbd_dev_header_watch_sync(rbd_dev, false);
 	if (ret)
-- 
1.7.9.5

Re: [PATCH 1/5] rbd: set removing flag while holding list lock

[Josh Durgin]

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

Alex Elder <elder@inktank.com> wrote:
>When unmapping a device, its id is supplied, and that is used to
>look up which rbd device should be unmapped.  Looking up the
>device involves searching the rbd device list while holding
>a spinlock that protects access to that list.
>
>Currently all of this is done under protection of the control lock,
>but that protection is going away soon.  To ensure the rbd_dev is
>still valid (still on the list) while setting its REMOVING flag, do
>so while still holding the list lock.  To do so, get rid of
>__rbd_get_dev(), and open code what it did in the one place it
>was used.
>
>Signed-off-by: Alex Elder <elder@inktank.com>
>---
> drivers/block/rbd.c |   53
>+++++++++++++++++++++------------------------------
> 1 file changed, 22 insertions(+), 31 deletions(-)
>
>diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
>index aace658..716ef1f 100644
>--- a/drivers/block/rbd.c
>+++ b/drivers/block/rbd.c
>@@ -5106,23 +5106,6 @@ err_out_module:
> 	return (ssize_t)rc;
> }
>
>-static struct rbd_device *__rbd_get_dev(unsigned long dev_id)
>-{
>-	struct list_head *tmp;
>-	struct rbd_device *rbd_dev;
>-
>-	spin_lock(&rbd_dev_list_lock);
>-	list_for_each(tmp, &rbd_dev_list) {
>-		rbd_dev = list_entry(tmp, struct rbd_device, node);
>-		if (rbd_dev->dev_id == dev_id) {
>-			spin_unlock(&rbd_dev_list_lock);
>-			return rbd_dev;
>-		}
>-	}
>-	spin_unlock(&rbd_dev_list_lock);
>-	return NULL;
>-}
>-
> static void rbd_dev_device_release(struct device *dev)
> {
> 	struct rbd_device *rbd_dev = dev_to_rbd_dev(dev);
>@@ -5167,7 +5150,8 @@ static ssize_t rbd_remove(struct bus_type *bus,
> 			  size_t count)
> {
> 	struct rbd_device *rbd_dev = NULL;
>-	int target_id;
>+	struct list_head *tmp;
>+	int dev_id;
> 	unsigned long ul;
> 	int ret;
>
>@@ -5176,26 +5160,33 @@ static ssize_t rbd_remove(struct bus_type *bus,
> 		return ret;
>
> 	/* convert to int; abort if we lost anything in the conversion */
>-	target_id = (int) ul;
>-	if (target_id != ul)
>+	dev_id = (int)ul;
>+	if (dev_id != ul)
> 		return -EINVAL;
>
> 	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
>
>-	rbd_dev = __rbd_get_dev(target_id);
>-	if (!rbd_dev) {
>-		ret = -ENOENT;
>-		goto done;
>+	ret = -ENOENT;
>+	spin_lock(&rbd_dev_list_lock);
>+	list_for_each(tmp, &rbd_dev_list) {
>+		rbd_dev = list_entry(tmp, struct rbd_device, node);
>+		if (rbd_dev->dev_id == dev_id) {
>+			ret = 0;
>+			break;
>+		}
> 	}
>-
>-	spin_lock_irq(&rbd_dev->lock);
>-	if (rbd_dev->open_count)
>-		ret = -EBUSY;
>-	else
>-		set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
>-	spin_unlock_irq(&rbd_dev->lock);
>+	if (!ret) {
>+		spin_lock_irq(&rbd_dev->lock);
>+		if (rbd_dev->open_count)
>+			ret = -EBUSY;
>+		else
>+			set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
>+		spin_unlock_irq(&rbd_dev->lock);
>+	}
>+	spin_unlock(&rbd_dev_list_lock);
> 	if (ret < 0)
> 		goto done;
>+
> 	rbd_bus_del_dev(rbd_dev);
> 	ret = rbd_dev_header_watch_sync(rbd_dev, false);
> 	if (ret)

[PATCH 2/5] rbd: protect against concurrent unmaps

[Alex Elder]

Make sure two concurrent unmap operations on the same rbd device
won't collide, by only proceeding with the removal and cleanup of a
device if is not already underway.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 716ef1f..380940d 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -5153,6 +5153,7 @@ static ssize_t rbd_remove(struct bus_type *bus,
 	struct list_head *tmp;
 	int dev_id;
 	unsigned long ul;
+	bool already = false;
 	int ret;

 	ret = strict_strtoul(buf, 10, &ul);
@@ -5180,11 +5181,12 @@ static ssize_t rbd_remove(struct bus_type *bus,
 		if (rbd_dev->open_count)
 			ret = -EBUSY;
 		else
-			set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
+			already = test_and_set_bit(RBD_DEV_FLAG_REMOVING,
+							&rbd_dev->flags);
 		spin_unlock_irq(&rbd_dev->lock);
 	}
 	spin_unlock(&rbd_dev_list_lock);
-	if (ret < 0)
+	if (ret < 0 || already)
 		goto done;

 	rbd_bus_del_dev(rbd_dev);
-- 
1.7.9.5

Re: [PATCH 2/5] rbd: protect against concurrent unmaps

[Josh Durgin]

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

Alex Elder <elder@inktank.com> wrote:
>Make sure two concurrent unmap operations on the same rbd device
>won't collide, by only proceeding with the removal and cleanup of a
>device if is not already underway.
>
>Signed-off-by: Alex Elder <elder@inktank.com>
>---
> drivers/block/rbd.c |    6 ++++--
> 1 file changed, 4 insertions(+), 2 deletions(-)
>
>diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
>index 716ef1f..380940d 100644
>--- a/drivers/block/rbd.c
>+++ b/drivers/block/rbd.c
>@@ -5153,6 +5153,7 @@ static ssize_t rbd_remove(struct bus_type *bus,
> 	struct list_head *tmp;
> 	int dev_id;
> 	unsigned long ul;
>+	bool already = false;
> 	int ret;
>
> 	ret = strict_strtoul(buf, 10, &ul);
>@@ -5180,11 +5181,12 @@ static ssize_t rbd_remove(struct bus_type *bus,
> 		if (rbd_dev->open_count)
> 			ret = -EBUSY;
> 		else
>-			set_bit(RBD_DEV_FLAG_REMOVING, &rbd_dev->flags);
>+			already = test_and_set_bit(RBD_DEV_FLAG_REMOVING,
>+							&rbd_dev->flags);
> 		spin_unlock_irq(&rbd_dev->lock);
> 	}
> 	spin_unlock(&rbd_dev_list_lock);
>-	if (ret < 0)
>+	if (ret < 0 || already)
> 		goto done;
>
> 	rbd_bus_del_dev(rbd_dev);

[PATCH 3/5] rbd: don't hold ctl_mutex to get/put device

[Alex Elder]

When an rbd device is first getting mapped, its device registration
is protected the control mutex.  There is no need to do that though,
because the device has already been assigned an id that's guaranteed
to be unique.

An unmap of an rbd device won't proceed if the device has a non-zero
open count or is already being unmapped.  So there's no need to hold
the control mutex in that case either.

Finally, an rbd device can't be opened if it is being removed, and
it won't go away if there is a non-zero open count.  So here too
there's no need to hold the control mutex while getting or putting a
reference to an rbd device's Linux device structure.

Drop the mutex calls in these cases.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 380940d..9c81a5c 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -489,10 +489,8 @@ static int rbd_open(struct block_device *bdev,
fmode_t mode)
 	if (removing)
 		return -ENOENT;

-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
 	(void) get_device(&rbd_dev->dev);
 	set_device_ro(bdev, rbd_dev->mapping.read_only);
-	mutex_unlock(&ctl_mutex);

 	return 0;
 }
@@ -507,9 +505,7 @@ static int rbd_release(struct gendisk *disk, fmode_t
mode)
 	spin_unlock_irq(&rbd_dev->lock);
 	rbd_assert(open_count_before > 0);

-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
 	put_device(&rbd_dev->dev);
-	mutex_unlock(&ctl_mutex);

 	return 0;
 }
@@ -4348,8 +4344,6 @@ static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
 	struct device *dev;
 	int ret;

-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
-
 	dev = &rbd_dev->dev;
 	dev->bus = &rbd_bus_type;
 	dev->type = &rbd_device_type;
@@ -4358,8 +4352,6 @@ static int rbd_bus_add_dev(struct rbd_device *rbd_dev)
 	dev_set_name(dev, "%d", rbd_dev->dev_id);
 	ret = device_register(dev);

-	mutex_unlock(&ctl_mutex);
-
 	return ret;
 }

@@ -5165,8 +5157,6 @@ static ssize_t rbd_remove(struct bus_type *bus,
 	if (dev_id != ul)
 		return -EINVAL;

-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
-
 	ret = -ENOENT;
 	spin_lock(&rbd_dev_list_lock);
 	list_for_each(tmp, &rbd_dev_list) {
@@ -5187,7 +5177,7 @@ static ssize_t rbd_remove(struct bus_type *bus,
 	}
 	spin_unlock(&rbd_dev_list_lock);
 	if (ret < 0 || already)
-		goto done;
+		return ret;

 	rbd_bus_del_dev(rbd_dev);
 	ret = rbd_dev_header_watch_sync(rbd_dev, false);
@@ -5195,11 +5185,8 @@ static ssize_t rbd_remove(struct bus_type *bus,
 		rbd_warn(rbd_dev, "failed to cancel watch event (%d)\n", ret);
 	rbd_dev_image_release(rbd_dev);
 	module_put(THIS_MODULE);
-	ret = count;
-done:
-	mutex_unlock(&ctl_mutex);

-	return ret;
+	return count;
 }

 /*
-- 
1.7.9.5

Re: [PATCH 3/5] rbd: don't hold ctl_mutex to get/put device

[Josh Durgin]

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

Alex Elder <elder@inktank.com> wrote:
>When an rbd device is first getting mapped, its device registration
>is protected the control mutex.  There is no need to do that though,
>because the device has already been assigned an id that's guaranteed
>to be unique.
>
>An unmap of an rbd device won't proceed if the device has a non-zero
>open count or is already being unmapped.  So there's no need to hold
>the control mutex in that case either.
>
>Finally, an rbd device can't be opened if it is being removed, and
>it won't go away if there is a non-zero open count.  So here too
>there's no need to hold the control mutex while getting or putting a
>reference to an rbd device's Linux device structure.
>
>Drop the mutex calls in these cases.
>
>Signed-off-by: Alex Elder <elder@inktank.com>
>---
> drivers/block/rbd.c |   17 ++---------------
> 1 file changed, 2 insertions(+), 15 deletions(-)
>
>diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
>index 380940d..9c81a5c 100644
>--- a/drivers/block/rbd.c
>+++ b/drivers/block/rbd.c
>@@ -489,10 +489,8 @@ static int rbd_open(struct block_device *bdev,
>fmode_t mode)
> 	if (removing)
> 		return -ENOENT;
>
>-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
> 	(void) get_device(&rbd_dev->dev);
> 	set_device_ro(bdev, rbd_dev->mapping.read_only);
>-	mutex_unlock(&ctl_mutex);
>
> 	return 0;
> }
>@@ -507,9 +505,7 @@ static int rbd_release(struct gendisk *disk,
>fmode_t
>mode)
> 	spin_unlock_irq(&rbd_dev->lock);
> 	rbd_assert(open_count_before > 0);
>
>-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
> 	put_device(&rbd_dev->dev);
>-	mutex_unlock(&ctl_mutex);
>
> 	return 0;
> }
>@@ -4348,8 +4344,6 @@ static int rbd_bus_add_dev(struct rbd_device
>*rbd_dev)
> 	struct device *dev;
> 	int ret;
>
>-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
>-
> 	dev = &rbd_dev->dev;
> 	dev->bus = &rbd_bus_type;
> 	dev->type = &rbd_device_type;
>@@ -4358,8 +4352,6 @@ static int rbd_bus_add_dev(struct rbd_device
>*rbd_dev)
> 	dev_set_name(dev, "%d", rbd_dev->dev_id);
> 	ret = device_register(dev);
>
>-	mutex_unlock(&ctl_mutex);
>-
> 	return ret;
> }
>
>@@ -5165,8 +5157,6 @@ static ssize_t rbd_remove(struct bus_type *bus,
> 	if (dev_id != ul)
> 		return -EINVAL;
>
>-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
>-
> 	ret = -ENOENT;
> 	spin_lock(&rbd_dev_list_lock);
> 	list_for_each(tmp, &rbd_dev_list) {
>@@ -5187,7 +5177,7 @@ static ssize_t rbd_remove(struct bus_type *bus,
> 	}
> 	spin_unlock(&rbd_dev_list_lock);
> 	if (ret < 0 || already)
>-		goto done;
>+		return ret;
>
> 	rbd_bus_del_dev(rbd_dev);
> 	ret = rbd_dev_header_watch_sync(rbd_dev, false);
>@@ -5195,11 +5185,8 @@ static ssize_t rbd_remove(struct bus_type *bus,
> 		rbd_warn(rbd_dev, "failed to cancel watch event (%d)\n", ret);
> 	rbd_dev_image_release(rbd_dev);
> 	module_put(THIS_MODULE);
>-	ret = count;
>-done:
>-	mutex_unlock(&ctl_mutex);
>
>-	return ret;
>+	return count;
> }
>
> /*

[PATCH 4/5] rbd: use rwsem to protect header updates

[Alex Elder]

Updating an image header needs to be protected to ensure it's
done consistently.  However distinct headers can be updated
concurrently without a problem.  Instead of using the global
control lock to serialize headder updates, just rely on the header
semaphore.  (It's already used, this just moves it out to cover
a broader section of the code.)

That leaves the control mutex protecting only the creation of rbd
clients, so rename it.

This resolves:
    http://tracker.ceph.com/issues/5222

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 9c81a5c..107e1e5 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -372,7 +372,7 @@ enum rbd_dev_flags {
 	RBD_DEV_FLAG_REMOVING,	/* this mapping is being removed */
 };

-static DEFINE_MUTEX(ctl_mutex);	  /* Serialize open/close/setup/teardown */
+static DEFINE_MUTEX(client_mutex);	/* Serialize client creation */

 static LIST_HEAD(rbd_dev_list);    /* devices */
 static DEFINE_SPINLOCK(rbd_dev_list_lock);
@@ -518,7 +518,7 @@ static const struct block_device_operations
rbd_bd_ops = {

 /*
  * Initialize an rbd client instance.  Success or not, this function
- * consumes ceph_opts.  Caller holds ctl_mutex.
+ * consumes ceph_opts.  Caller holds client_mutex.
  */
 static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
 {
@@ -675,13 +675,13 @@ static struct rbd_client *rbd_get_client(struct
ceph_options *ceph_opts)
 {
 	struct rbd_client *rbdc;

-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
+	mutex_lock_nested(&client_mutex, SINGLE_DEPTH_NESTING);
 	rbdc = rbd_client_find(ceph_opts);
 	if (rbdc)	/* using an existing client */
 		ceph_destroy_options(ceph_opts);
 	else
 		rbdc = rbd_client_create(ceph_opts);
-	mutex_unlock(&ctl_mutex);
+	mutex_unlock(&client_mutex);

 	return rbdc;
 }
@@ -835,7 +835,6 @@ static int rbd_header_from_disk(struct rbd_device
*rbd_dev,

 	/* We won't fail any more, fill in the header */

-	down_write(&rbd_dev->header_rwsem);
 	if (first_time) {
 		header->object_prefix = object_prefix;
 		header->obj_order = ondisk->options.order;
@@ -864,8 +863,6 @@ static int rbd_header_from_disk(struct rbd_device
*rbd_dev,
 		if (rbd_dev->mapping.size != header->image_size)
 			rbd_dev->mapping.size = header->image_size;

-	up_write(&rbd_dev->header_rwsem);
-
 	return 0;
 out_2big:
 	ret = -EIO;
@@ -3349,17 +3346,17 @@ static int rbd_dev_refresh(struct rbd_device
*rbd_dev)
 	int ret;

 	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
+	down_write(&rbd_dev->header_rwsem);
 	mapping_size = rbd_dev->mapping.size;
 	if (rbd_dev->image_format == 1)
 		ret = rbd_dev_v1_header_info(rbd_dev);
 	else
 		ret = rbd_dev_v2_header_info(rbd_dev);
+	up_write(&rbd_dev->header_rwsem);

 	/* If it's a mapped snapshot, validate its EXISTS flag */

 	rbd_exists_validate(rbd_dev);
-	mutex_unlock(&ctl_mutex);
 	if (mapping_size != rbd_dev->mapping.size) {
 		sector_t size;

@@ -4288,12 +4285,10 @@ static int rbd_dev_v2_header_info(struct
rbd_device *rbd_dev)
 	bool first_time = rbd_dev->header.object_prefix == NULL;
 	int ret;

-	down_write(&rbd_dev->header_rwsem);
-
 	if (first_time) {
 		ret = rbd_dev_v2_header_onetime(rbd_dev);
 		if (ret)
-			goto out;
+			return ret;
 	}

 	/*
@@ -4308,7 +4303,7 @@ static int rbd_dev_v2_header_info(struct
rbd_device *rbd_dev)

 		ret = rbd_dev_v2_parent_info(rbd_dev);
 		if (ret)
-			goto out;
+			return ret;

 		/*
 		 * Print a warning if this is the initial probe and
@@ -4325,7 +4320,7 @@ static int rbd_dev_v2_header_info(struct
rbd_device *rbd_dev)

 	ret = rbd_dev_v2_image_size(rbd_dev);
 	if (ret)
-		goto out;
+		return ret;

 	if (rbd_dev->spec->snap_id == CEPH_NOSNAP)
 		if (rbd_dev->mapping.size != rbd_dev->header.image_size)
@@ -4333,8 +4328,6 @@ static int rbd_dev_v2_header_info(struct
rbd_device *rbd_dev)

 	ret = rbd_dev_v2_snap_context(rbd_dev);
 	dout("rbd_dev_v2_snap_context returned %d\n", ret);
-out:
-	up_write(&rbd_dev->header_rwsem);

 	return ret;
 }
-- 
1.7.9.5

Re: [PATCH 4/5] rbd: use rwsem to protect header updates

[Josh Durgin]

On 05/31/2013 06:20 PM, Alex Elder wrote:
> Updating an image header needs to be protected to ensure it's
> done consistently.  However distinct headers can be updated
> concurrently without a problem.  Instead of using the global
> control lock to serialize headder updates, just rely on the header
> semaphore.  (It's already used, this just moves it out to cover
> a broader section of the code.)
>
> That leaves the control mutex protecting only the creation of rbd
> clients, so rename it.
>
> This resolves:
>      http://tracker.ceph.com/issues/5222
>
> Signed-off-by: Alex Elder <elder@inktank.com>
> ---
>   drivers/block/rbd.c |   25 +++++++++----------------
>   1 file changed, 9 insertions(+), 16 deletions(-)
>
> diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
> index 9c81a5c..107e1e5 100644
> --- a/drivers/block/rbd.c
> +++ b/drivers/block/rbd.c
> @@ -372,7 +372,7 @@ enum rbd_dev_flags {
>   	RBD_DEV_FLAG_REMOVING,	/* this mapping is being removed */
>   };
>
> -static DEFINE_MUTEX(ctl_mutex);	  /* Serialize open/close/setup/teardown */
> +static DEFINE_MUTEX(client_mutex);	/* Serialize client creation */
>
>   static LIST_HEAD(rbd_dev_list);    /* devices */
>   static DEFINE_SPINLOCK(rbd_dev_list_lock);
> @@ -518,7 +518,7 @@ static const struct block_device_operations
> rbd_bd_ops = {
>
>   /*
>    * Initialize an rbd client instance.  Success or not, this function
> - * consumes ceph_opts.  Caller holds ctl_mutex.
> + * consumes ceph_opts.  Caller holds client_mutex.
>    */
>   static struct rbd_client *rbd_client_create(struct ceph_options *ceph_opts)
>   {
> @@ -675,13 +675,13 @@ static struct rbd_client *rbd_get_client(struct
> ceph_options *ceph_opts)
>   {
>   	struct rbd_client *rbdc;
>
> -	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
> +	mutex_lock_nested(&client_mutex, SINGLE_DEPTH_NESTING);
>   	rbdc = rbd_client_find(ceph_opts);
>   	if (rbdc)	/* using an existing client */
>   		ceph_destroy_options(ceph_opts);
>   	else
>   		rbdc = rbd_client_create(ceph_opts);
> -	mutex_unlock(&ctl_mutex);
> +	mutex_unlock(&client_mutex);
>
>   	return rbdc;
>   }
> @@ -835,7 +835,6 @@ static int rbd_header_from_disk(struct rbd_device
> *rbd_dev,
>
>   	/* We won't fail any more, fill in the header */
>
> -	down_write(&rbd_dev->header_rwsem);
>   	if (first_time) {
>   		header->object_prefix = object_prefix;
>   		header->obj_order = ondisk->options.order;
> @@ -864,8 +863,6 @@ static int rbd_header_from_disk(struct rbd_device
> *rbd_dev,
>   		if (rbd_dev->mapping.size != header->image_size)
>   			rbd_dev->mapping.size = header->image_size;
>
> -	up_write(&rbd_dev->header_rwsem);
> -
>   	return 0;
>   out_2big:
>   	ret = -EIO;
> @@ -3349,17 +3346,17 @@ static int rbd_dev_refresh(struct rbd_device
> *rbd_dev)
>   	int ret;
>
>   	rbd_assert(rbd_image_format_valid(rbd_dev->image_format));
> -	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);
> +	down_write(&rbd_dev->header_rwsem);
>   	mapping_size = rbd_dev->mapping.size;
>   	if (rbd_dev->image_format == 1)
>   		ret = rbd_dev_v1_header_info(rbd_dev);
>   	else
>   		ret = rbd_dev_v2_header_info(rbd_dev);
> +	up_write(&rbd_dev->header_rwsem);

I think this needs to happen after rbd_exists_validate() so that
reading the snapshot context via rbd_dev_snap_index() is protected.

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

>
>   	/* If it's a mapped snapshot, validate its EXISTS flag */
>
>   	rbd_exists_validate(rbd_dev);
> -	mutex_unlock(&ctl_mutex);
>   	if (mapping_size != rbd_dev->mapping.size) {
>   		sector_t size;
>
> @@ -4288,12 +4285,10 @@ static int rbd_dev_v2_header_info(struct
> rbd_device *rbd_dev)
>   	bool first_time = rbd_dev->header.object_prefix == NULL;
>   	int ret;
>
> -	down_write(&rbd_dev->header_rwsem);
> -
>   	if (first_time) {
>   		ret = rbd_dev_v2_header_onetime(rbd_dev);
>   		if (ret)
> -			goto out;
> +			return ret;
>   	}
>
>   	/*
> @@ -4308,7 +4303,7 @@ static int rbd_dev_v2_header_info(struct
> rbd_device *rbd_dev)
>
>   		ret = rbd_dev_v2_parent_info(rbd_dev);
>   		if (ret)
> -			goto out;
> +			return ret;
>
>   		/*
>   		 * Print a warning if this is the initial probe and
> @@ -4325,7 +4320,7 @@ static int rbd_dev_v2_header_info(struct
> rbd_device *rbd_dev)
>
>   	ret = rbd_dev_v2_image_size(rbd_dev);
>   	if (ret)
> -		goto out;
> +		return ret;
>
>   	if (rbd_dev->spec->snap_id == CEPH_NOSNAP)
>   		if (rbd_dev->mapping.size != rbd_dev->header.image_size)
> @@ -4333,8 +4328,6 @@ static int rbd_dev_v2_header_info(struct
> rbd_device *rbd_dev)
>
>   	ret = rbd_dev_v2_snap_context(rbd_dev);
>   	dout("rbd_dev_v2_snap_context returned %d\n", ret);
> -out:
> -	up_write(&rbd_dev->header_rwsem);
>
>   	return ret;
>   }
>

[PATCH 5/5] rbd: take a little credit

[Alex Elder]

Add a name to the list of authors.

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

diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
index 107e1e5..cb88a9a 100644
--- a/drivers/block/rbd.c
+++ b/drivers/block/rbd.c
@@ -5287,6 +5287,7 @@ static void __exit rbd_exit(void)
 module_init(rbd_init);
 module_exit(rbd_exit);

+MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
 MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
 MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
 MODULE_DESCRIPTION("rados block device");
-- 
1.7.9.5

Re: [PATCH 5/5] rbd: take a little credit

[Josh Durgin]

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

Alex Elder <elder@inktank.com> wrote:
>Add a name to the list of authors.
>
>Signed-off-by: Alex Elder <elder@inktank.com>
>---
> drivers/block/rbd.c |    1 +
> 1 file changed, 1 insertion(+)
>
>diff --git a/drivers/block/rbd.c b/drivers/block/rbd.c
>index 107e1e5..cb88a9a 100644
>--- a/drivers/block/rbd.c
>+++ b/drivers/block/rbd.c
>@@ -5287,6 +5287,7 @@ static void __exit rbd_exit(void)
> module_init(rbd_init);
> module_exit(rbd_exit);
>
>+MODULE_AUTHOR("Alex Elder <elder@inktank.com>");
> MODULE_AUTHOR("Sage Weil <sage@newdream.net>");
> MODULE_AUTHOR("Yehuda Sadeh <yehuda@hq.newdream.net>");
> MODULE_DESCRIPTION("rados block device");

Re: [PATCH 0/5] rbd: clean up use of ctl_mutex

[Alex Elder]

On 05/31/2013 08:17 PM, Alex Elder wrote:
> This series eliminates most uses of the ctl_mutex for locking.
> Its precise purpose was always a little fuzzy to me; with this
> series in place it is now only used to protect looking up
> ceph clients.  It also eliminates some lockdep problems
> reported at unmap time.

These patches are available in the "review/rbd-mutex"
branch of the ceph-client git repository.


> 
> 					-Alex
> 
> [PATCH 1/5] rbd: set removing flag while holding list lock
> [PATCH 2/5] rbd: protect against concurrent unmaps
> [PATCH 3/5] rbd: don't hold ctl_mutex to get/put device
> [PATCH 4/5] rbd: use rwsem to protect header updates
> [PATCH 5/5] rbd: take a little credit
>