rbd: clean up use of ctl_mutex

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

Add a name to the list of authors.

Signed-off-by Alex Elder <[email protected]>

---
drivers/block/rbd.c | 1 +

1 file changed, 1 insertion(+)

drivers/block/rbd.c +1 -0

index 107e1e5..cb88a9a 100644
@@ -5287,6 +5287,7 @@ static void __exit rbd_exit(void)
 module_init(rbd_init);
 module_exit(rbd_exit);
 
+MODULE_AUTHOR("Alex Elder <[email protected]>");
 MODULE_AUTHOR("Sage Weil <[email protected]>");
 MODULE_AUTHOR("Yehuda Sadeh <[email protected]>");
 MODULE_DESCRIPTION("rados block device");

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 <[email protected]>

---
drivers/block/rbd.c | 17 ++---------------

1 file changed, 2 insertions(+), 15 deletions(-)

drivers/block/rbd.c +0 -2

index 380940d..9c81a5c 100644
@@ -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;

}

drivers/block/rbd.c +0 -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);

drivers/block/rbd.c +0 -1

-	mutex_lock_nested(&ctl_mutex, SINGLE_DEPTH_NESTING);

	put_device(&rbd_dev->dev);

drivers/block/rbd.c +0 -1

-	mutex_unlock(&ctl_mutex);

	return 0;
}

drivers/block/rbd.c +0 -2

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

drivers/block/rbd.c +2 -9

@@ -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;
 }
 
 /*

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 <[email protected]>

---
drivers/block/rbd.c | 6 ++++--

1 file changed, 4 insertions(+), 2 deletions(-)

drivers/block/rbd.c +4 -2

index 716ef1f..380940d 100644
@@ -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);

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 <[email protected]>

---
drivers/block/rbd.c | 53
+++++++++++++++++++++------------------------------

1 file changed, 22 insertions(+), 31 deletions(-)

drivers/block/rbd.c +22 -31

index aace658..716ef1f 100644
@@ -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)

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 <[email protected]>

---
  drivers/block/rbd.c |   25 +++++++++----------------
  1 file changed, 9 insertions(+), 16 deletions(-)

drivers/block/rbd.c +2 -2

index 9c81a5c..107e1e5 100644
@@ -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)

  {

drivers/block/rbd.c +2 -2

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

  }

drivers/block/rbd.c +0 -1

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

drivers/block/rbd.c +0 -2

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

drivers/block/rbd.c +0 -0

@@ -3349,17 +3346,17 @@ static int rbd_dev_refresh(struct rbd_device
 *rbd_dev)
  	int ret;

drivers/block/rbd.c +1 -4

  	/* 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;
  	}
 

/*

drivers/block/rbd.c +1 -1

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

drivers/block/rbd.c +1 -1

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

drivers/block/rbd.c +0 -2

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

  }