[RFC PATCH 0/5] rbd_tcm cluster COMPARE AND WRITE

[RFC PATCH 0/5] rbd_tcm cluster COMPARE AND WRITE

[David Disseldorp]

Hi Mike,

This RFC patch-set expands on your recently proposed cluster API changes
to add rbd_tcm COMPARE AND WRITE support.

Rather than offloading the entire COMPARE AND WRITE operation to the
cluster back-end, these changes extend on the existing local semaphore
usage, to additionally take an RBD COMPARE AND WRITE device lock. RBD
device lock acquisition is handled via a newly added rbd_tcm workqueue.

I'd appreciate feedback on the overall direction of these changes.

Cheers, David

--

 drivers/block/Kconfig                |   3 +-
 drivers/block/rbd.h                  |   3 +
 drivers/block/rbd_main.c             |  26 ++--
 drivers/block/rbd_tcm.c              | 208 +++++++++++++++++++++++++++
 drivers/target/target_core_sbc.c     |  28 +++-
 include/target/target_core_cluster.h |   4 +
 6 files changed, 253 insertions(+), 19 deletions(-)

[RFC PATCH 1/5] SQUASH rbd/locking: remove rbd_warn() carriage returns

[David Disseldorp]

rbd_warn() already appends '\n'.

Signed-off-by: David Disseldorp <ddiss@suse.de>
---
 drivers/block/rbd_main.c | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/drivers/block/rbd_main.c b/drivers/block/rbd_main.c
index 774a03a..57e76ed 100644
--- a/drivers/block/rbd_main.c
+++ b/drivers/block/rbd_main.c
@@ -4369,13 +4369,13 @@ static int rbd_dev_print_lock_info(struct rbd_device *rbd_dev, char *name,
 	switch (addr->in_addr.ss_family) {
 	case AF_INET:
 		sin = (struct sockaddr_in *)&addr->in_addr;
-		rbd_warn(rbd_dev, "%s %s %s.%llu %s %pI4\n",
+		rbd_warn(rbd_dev, "%s %s %s.%llu %s %pI4",
 			 name, cookie, ceph_entity_type_name(type), num, desc,
 			 &sin->sin_addr.s_addr);
 		break;
 	case AF_INET6:
 		sin6 = (struct sockaddr_in6 *)&addr->in_addr;
-		rbd_warn(rbd_dev, "%s %s %s.%llu %s %pI6\n",
+		rbd_warn(rbd_dev, "%s %s %s.%llu %s %pI6",
 			 name, cookie, ceph_entity_type_name(type), num, desc,
 			 &sin6->sin6_addr);
 		break;
@@ -4423,7 +4423,7 @@ static size_t rbd_dev_print_locks(struct rbd_device *rbd_dev)
 						   GFP_KERNEL);
 		if (IS_ERR(lock)) {
 			rbd_warn(rbd_dev,
-				 "Could not print info for all locks\n");
+				 "Could not print info for all locks");
 			ret = PTR_ERR(lock);
 			goto free_list_locks_pg;
 		}
@@ -4580,10 +4580,10 @@ static ssize_t rbd_lock_set(struct device *dev, struct device_attribute *attr,
 
 	ret = sscanf(buf, "%15s %15s %15s\n", name, cookie, desc);
 	if (ret != 3) {
-		rbd_warn(rbd_dev, "Invalid number of params. Got %d\n", ret);
+		rbd_warn(rbd_dev, "Invalid number of params. Got %d", ret);
 		return -EINVAL;
 	} else if (!strlen(name) || !strlen(cookie) || !strlen(desc)) {
-		rbd_warn(rbd_dev, "missing param\n");
+		rbd_warn(rbd_dev, "missing param");
 		return -EINVAL;
 	}
 
@@ -4604,10 +4604,10 @@ static ssize_t rbd_unlock_set(struct device *dev, struct device_attribute *attr,
 
 	ret = sscanf(buf, "%15s %15s\n", name, cookie);
 	if (ret != 2) {
-		rbd_warn(rbd_dev, "Invalid number of params. Got %d\n", ret);
+		rbd_warn(rbd_dev, "Invalid number of params. Got %d", ret);
 		return -EINVAL;
 	} else if (!strlen(name) || !strlen(cookie)) {
-		rbd_warn(rbd_dev, "missing param\n");
+		rbd_warn(rbd_dev, "missing param");
 		return -EINVAL;
 	}
 
@@ -4628,10 +4628,10 @@ static ssize_t rbd_break_locks_set(struct device *dev,
 
 	ret = sscanf(buf, "%15s\n", name);
 	if (ret != 1) {
-		rbd_warn(rbd_dev, "Invalid number of params. Got %d\n", ret);
+		rbd_warn(rbd_dev, "Invalid number of params. Got %d", ret);
 		return -EINVAL;
 	} else if (!strlen(name)) {
-		rbd_warn(rbd_dev, "missing param\n");
+		rbd_warn(rbd_dev, "missing param");
 		return -EINVAL;
 	}
 
-- 
2.1.4

[RFC PATCH 2/5] SQUASH rbd: fix recursive BLK_DEV_RBD_TCM dependency

[David Disseldorp]

net/Kconfig:5:error: recursive dependency detected!
net/Kconfig:5:  symbol NET is selected by SCSI_NETLINK
drivers/scsi/Kconfig:50:	symbol SCSI_NETLINK is selected by
SCSI_FC_ATTRS
drivers/scsi/Kconfig:276:	symbol SCSI_FC_ATTRS is selected by
LIBFC
drivers/scsi/Kconfig:659:	symbol LIBFC is selected by TCM_QLA2XXX
drivers/scsi/qla2xxx/Kconfig:31:symbol TCM_QLA2XXX depends on
TARGET_CORE
drivers/target/Kconfig:2:	symbol TARGET_CORE is selected by
BLK_DEV_RBD_TCM
drivers/block/Kconfig:537:	symbol BLK_DEV_RBD_TCM depends on
BLK_DEV_RBD
drivers/block/Kconfig:520:	symbol BLK_DEV_RBD depends on INET
net/Kconfig:54: symbol INET depends on NET

Signed-off-by: David Disseldorp <ddiss@suse.de>
---
 drivers/block/Kconfig | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/drivers/block/Kconfig b/drivers/block/Kconfig
index d517744..fad92be 100644
--- a/drivers/block/Kconfig
+++ b/drivers/block/Kconfig
@@ -550,8 +550,7 @@ config BLK_DEV_RBD
 
 config BLK_DEV_RBD_TCM
        bool "Clustered TCM using Rados block devices (RBD)"
-       depends on BLK_DEV_RBD
-       select TARGET_CORE
+       depends on BLK_DEV_RBD && TARGET_CORE
        select TCM_IBLOCK
        default n
        help
-- 
2.1.4

[RFC PATCH 3/5] rbd: export rbd_dev_[un]lock()

[David Disseldorp]

To be used for COMPARE AND WRITE device locking.

Signed-off-by: David Disseldorp <ddiss@suse.de>
---
 drivers/block/rbd.h      | 3 +++
 drivers/block/rbd_main.c | 8 ++++----
 2 files changed, 7 insertions(+), 4 deletions(-)

diff --git a/drivers/block/rbd.h b/drivers/block/rbd.h
index 60c6b85..1c6f947 100644
--- a/drivers/block/rbd.h
+++ b/drivers/block/rbd.h
@@ -24,6 +24,9 @@ extern int rbd_obj_notify_scsi_event_sync(struct rbd_device *rbd_dev, u32 event,
 extern int rbd_obj_notify_ack_sync(struct rbd_device *rbd_dev, u64 notify_id);
 extern int rbd_attach_tcm_dev(struct rbd_device *rbd_dev, void *data);
 extern int rbd_detach_tcm_dev(struct rbd_device *rbd_dev);
+extern int rbd_dev_lock(struct rbd_device *rbd_dev, char *name, u8 type,
+			char *cookie, char *tag, char *desc, u8 flags);
+extern int rbd_dev_unlock(struct rbd_device *rbd_dev, char *name, char *cookie);
 
 #if defined(CONFIG_TCM_IBLOCK) || defined(CONFIG_TCM_IBLOCK_MODULE)
 
diff --git a/drivers/block/rbd_main.c b/drivers/block/rbd_main.c
index 57e76ed..cb90c08 100644
--- a/drivers/block/rbd_main.c
+++ b/drivers/block/rbd_main.c
@@ -4134,11 +4134,11 @@ int rbd_attach_tcm_dev(struct rbd_device *rbd_dev, void *data)
  * @type: lock type (RADOS_LOCK_EXCLUSIVE or RADOS_LOCK_SHARED)
  * @cookie: user-defined identifier for this instance of the lock
  * @tag: if RADOS_LOCK_SHARED, tag of the lock. NULL if non shared.
- * desc: user-defined lock description
+ * @desc: user-defined lock description
  * @flags: lock flags
  */
-static int rbd_dev_lock(struct rbd_device *rbd_dev, char *name, u8 type,
-			char *cookie, char *tag, char *desc, u8 flags)
+int rbd_dev_lock(struct rbd_device *rbd_dev, char *name, u8 type,
+		 char *cookie, char *tag, char *desc, u8 flags)
 {
 	int lock_op_buf_size;
 	int name_len = strlen(name);
@@ -4194,7 +4194,7 @@ static int rbd_dev_lock(struct rbd_device *rbd_dev, char *name, u8 type,
  * @name: the name of the lock
  * @cookie: user-defined identifier for this instance of the lock
  */
-static int rbd_dev_unlock(struct rbd_device *rbd_dev, char *name, char *cookie)
+int rbd_dev_unlock(struct rbd_device *rbd_dev, char *name, char *cookie)
 {
 	int unlock_op_buf_size;
 	int name_len = strlen(name);
-- 
2.1.4

[RFC PATCH 4/5] drivers/rbd_tcm: Add cluster API CAW locking hooks

[David Disseldorp]

Use an RBD device lock in addition to a local device semaphore to ensure
that COMPARE AND WRITE operations are handled in an atomic fashion.

Lock acquisition is handled via a new workqueue, which currently polls
periodically for the RBD device lock until it can be acquired. This
polling behaviour should be replaced in future with watch/notify
functionality similar to that used for RBD_NOTIFY_OP_SCSI_LUN_RESET.

Signed-off-by: David Disseldorp <ddiss@suse.de>
---
 drivers/block/rbd_tcm.c              | 208 +++++++++++++++++++++++++++++++++++
 include/target/target_core_cluster.h |   4 +
 2 files changed, 212 insertions(+)

diff --git a/drivers/block/rbd_tcm.c b/drivers/block/rbd_tcm.c
index f3ee6ff..47a006f 100644
--- a/drivers/block/rbd_tcm.c
+++ b/drivers/block/rbd_tcm.c
@@ -36,11 +36,26 @@ struct rbd_tcm_reset_event {
 	u64 notify_id;
 };
 
+#define RBD_TCM_CAW_LOCK_POLL_HZ	(1 * HZ)
+#define RBD_TCM_CAW_LOCK_TOUT_HZ	(5 * HZ)
+
+struct rbd_tcm_caw_locker {
+	struct delayed_work caw_work;
+	struct delayed_work caw_tout_work;
+	struct completion lock_finished;
+	int lock_rc;
+	int retries;
+	struct rbd_tcm_device *rbd_tcm_dev;
+};
+
 struct rbd_tcm_device {
 	struct rbd_device *rbd_dev;
 	struct se_device *se_dev;
 
 	struct rbd_tcm_reset_event reset_evt;
+
+	struct rbd_tcm_caw_locker *caw_locker;
+	struct workqueue_struct *caw_wq;
 };
 
 static int rbd_tcm_start_reset(struct se_device *se_dev, u32 timeout)
@@ -92,6 +107,8 @@ static int rbd_tcm_detach_device(struct se_device *se_dev)
 	struct request_queue *q = ibock_se_device_to_q(se_dev);
 	struct rbd_tcm_device *rbd_tcm_dev = se_dev->cluster_dev_data;
 
+	destroy_workqueue(rbd_tcm_dev->caw_wq);
+
 	cancel_work_sync(&rbd_tcm_dev->reset_evt.work);
 	se_dev->cluster_dev_data = NULL;
 	rbd_detach_tcm_dev(q->queuedata);
@@ -112,16 +129,207 @@ static int rbd_tcm_attach_device(struct se_device *se_dev)
 	INIT_WORK(&rbd_tcm_dev->reset_evt.work, rbd_tcm_reset_event_workfn);
 	rbd_tcm_dev->reset_evt.rbd_tcm_dev = rbd_tcm_dev;
 
+	/* work queue to serialise COMPARE AND WRITE handling */
+	rbd_tcm_dev->caw_wq = alloc_workqueue("caw-rbd",
+					      WQ_MEM_RECLAIM | WQ_UNBOUND, 1);
+	if (!rbd_tcm_dev->caw_wq) {
+		rbd_warn(rbd_tcm_dev->rbd_dev,
+			 "Unable to create CAW workqueue for rbd");
+		kfree(rbd_tcm_dev);
+		return -ENOMEM;
+	}
+
 	se_dev->cluster_dev_data = rbd_tcm_dev;
 	return rbd_attach_tcm_dev(q->queuedata, rbd_tcm_dev);
 }
 
+static void
+rbd_tcm_caw_lock_dispatch(struct work_struct *work)
+{
+	struct rbd_tcm_caw_locker *caw_locker
+		= container_of(work, struct rbd_tcm_caw_locker, caw_work.work);
+	struct rbd_device *rbd_dev = caw_locker->rbd_tcm_dev->rbd_dev;
+	int ret;
+
+	pr_debug("CAW lock dispatch running\n");
+
+	ret = rbd_dev_lock(rbd_dev, "rbd_tcm_caw", 1, "cookie", "",
+			   "exclusive COMPARE AND wRITE lock", 0);
+	if (ret == -EEXIST) {
+		rbd_warn(rbd_dev, "CAW lock conflict, deferring operation");
+		/* TODO use unlock notification, instead of polling */
+		caw_locker->retries++;
+		queue_delayed_work(caw_locker->rbd_tcm_dev->caw_wq,
+				   &caw_locker->caw_work,
+				   RBD_TCM_CAW_LOCK_POLL_HZ);
+		return;
+	} else if (ret < 0) {
+		rbd_warn(rbd_dev, "failed to obtain CAW lock: %d", ret);
+		caw_locker->lock_rc = ret;
+		complete(&caw_locker->lock_finished);
+		return;
+	}
+
+	pr_debug("acquired COMPARE AND WRITE lock after %d retries\n",
+		 caw_locker->retries);
+
+	cancel_delayed_work_sync(&caw_locker->caw_tout_work);
+	caw_locker->lock_rc = 0;
+	complete(&caw_locker->lock_finished);
+}
+
+static void
+rbd_tcm_caw_lock_timeout(struct work_struct *work)
+{
+	struct rbd_tcm_caw_locker *caw_locker
+		= container_of(work, struct rbd_tcm_caw_locker,
+			       caw_tout_work.work);
+
+	pr_warn("CAW lock timeout running\n");
+
+	cancel_delayed_work_sync(&caw_locker->caw_work);
+	caw_locker->lock_rc = -ETIMEDOUT;
+	complete(&caw_locker->lock_finished);
+}
+
+/*
+ * Ensure cluster wide exclusive COMPARE AND WRITE access via an RBD device
+ * lock. Local exclusivity is handled via dev->saw_sem.
+ * In future, we may be able to offload the entire atomic read->compare->write
+ * sequence to the OSDs, which would make this interface pretty useless.
+ */
+static int
+rbd_tcm_caw_lock(struct se_device *se_dev)
+{
+	struct rbd_tcm_device *rbd_tcm_dev = se_dev->cluster_dev_data;
+	struct rbd_tcm_caw_locker *caw_locker;
+	int ret;
+
+	ret = down_interruptible(&se_dev->caw_sem);
+	if (ret != 0) {
+		pr_err("failed to obtain local semaphore\n");
+		return ret;
+	}
+
+	BUG_ON(rbd_tcm_dev->caw_locker != NULL);
+
+	pr_debug("got local CAW semaphore\n");
+
+	caw_locker = kzalloc(sizeof(*caw_locker), GFP_KERNEL);
+	if (!caw_locker) {
+		pr_err("Unable to allocate caw_locker\n");
+		ret = -ENOMEM;;
+		goto err_sem_up;
+	}
+
+	init_completion(&caw_locker->lock_finished);
+	/* whichever work finishes first cancels the other */
+	INIT_DELAYED_WORK(&caw_locker->caw_work, rbd_tcm_caw_lock_dispatch);
+	INIT_DELAYED_WORK(&caw_locker->caw_tout_work, rbd_tcm_caw_lock_timeout);
+	caw_locker->lock_rc = -EINTR;
+	caw_locker->rbd_tcm_dev = rbd_tcm_dev;
+
+	rbd_tcm_dev->caw_locker = caw_locker;
+
+	queue_delayed_work(rbd_tcm_dev->caw_wq, &caw_locker->caw_work, 0);
+	queue_delayed_work(rbd_tcm_dev->caw_wq, &caw_locker->caw_tout_work,
+			   RBD_TCM_CAW_LOCK_TOUT_HZ);
+	pr_debug("work queued, awaiting completion\n");
+	wait_for_completion(&caw_locker->lock_finished);
+	if (caw_locker->lock_rc < 0) {
+		ret = caw_locker->lock_rc;
+		goto err_locker_free;
+	}
+
+	/* caw_locker freed following unlock */
+
+	return 0;
+
+err_locker_free:
+	kfree(caw_locker);
+	rbd_tcm_dev->caw_locker = NULL;
+err_sem_up:
+	up(&se_dev->caw_sem);
+	pr_debug("dropped local CAW semaphore on failure\n");
+	return ret;
+}
+
+static void
+rbd_tcm_caw_unlock_dispatch(struct work_struct *work)
+{
+	struct rbd_tcm_caw_locker *caw_locker
+		= container_of(work, struct rbd_tcm_caw_locker, caw_work.work);
+	struct rbd_device *rbd_dev = caw_locker->rbd_tcm_dev->rbd_dev;
+	int ret;
+
+	pr_debug("CAW unlock dispatch running\n");
+
+	ret = rbd_dev_unlock(rbd_dev, "rbd_tcm_caw", "cookie");
+	if (ret < 0)
+		rbd_warn(rbd_dev, "failed to drop CAW lock: %d", ret);
+	else {
+		pr_debug("dropped RBD COMPARE AND WRITE lock\n");
+	}
+
+	cancel_delayed_work_sync(&caw_locker->caw_tout_work);
+	caw_locker->lock_rc = ret;
+	complete(&caw_locker->lock_finished);
+}
+
+static void
+rbd_tcm_caw_unlock_timeout(struct work_struct *work)
+{
+	struct rbd_tcm_caw_locker *caw_locker
+		= container_of(work, struct rbd_tcm_caw_locker,
+			       caw_tout_work.work);
+
+	pr_warn("CAW unlock timeout running\n");
+
+	cancel_delayed_work_sync(&caw_locker->caw_work);
+	caw_locker->lock_rc = -ETIMEDOUT;
+	complete(&caw_locker->lock_finished);
+}
+
+static void
+rbd_tcm_caw_unlock(struct se_device *se_dev)
+{
+	struct rbd_tcm_device *rbd_tcm_dev = se_dev->cluster_dev_data;
+	struct rbd_tcm_caw_locker *caw_locker = rbd_tcm_dev->caw_locker;
+
+	/* set if lock was successfull */
+	BUG_ON(caw_locker == NULL);
+
+	init_completion(&caw_locker->lock_finished);
+	INIT_DELAYED_WORK(&caw_locker->caw_work, rbd_tcm_caw_unlock_dispatch);
+	INIT_DELAYED_WORK(&caw_locker->caw_tout_work, rbd_tcm_caw_unlock_timeout);
+	caw_locker->lock_rc = -EINTR;
+	caw_locker->retries = 0;
+	caw_locker->rbd_tcm_dev = rbd_tcm_dev;
+
+	queue_delayed_work(rbd_tcm_dev->caw_wq, &caw_locker->caw_work, 0);
+	queue_delayed_work(rbd_tcm_dev->caw_wq, &caw_locker->caw_tout_work,
+			   RBD_TCM_CAW_LOCK_TOUT_HZ);
+	pr_debug("work queued, awaiting completion\n");
+	wait_for_completion(&caw_locker->lock_finished);
+	if (caw_locker->lock_rc < 0) {
+		pr_warn("leaving stale RBD CAW lock");
+	}
+
+	kfree(caw_locker);
+	rbd_tcm_dev->caw_locker = NULL;
+
+	up(&se_dev->caw_sem);
+	pr_debug("dropped local CAW semaphore\n");
+}
+
 static struct se_cluster_api rbd_tcm_template = {
 	.name		= "rbd",
 	.owner		= THIS_MODULE,
 	.reset_device	= rbd_tcm_start_reset,
 	.attach_device	= rbd_tcm_attach_device,
 	.detach_device	= rbd_tcm_detach_device,
+	.caw_lock	= rbd_tcm_caw_lock,
+	.caw_unlock	= rbd_tcm_caw_unlock,
 };
 
 int rbd_tcm_register(void)
diff --git a/include/target/target_core_cluster.h b/include/target/target_core_cluster.h
index 4860c2e..cc1e2aa 100644
--- a/include/target/target_core_cluster.h
+++ b/include/target/target_core_cluster.h
@@ -23,6 +23,10 @@ struct se_cluster_api {
 	 * takes longer than timeout seconds then -ETIMEDOUT should be returned.
 	 */
 	int (*reset_device)(struct se_device *dev, u32 timeout);
+
+	/* exclusive device locking for atomic COMPARE AND WRITE */
+	int (*caw_lock)(struct se_device *se_dev);
+	void (*caw_unlock)(struct se_device *se_dev);
 };
 
 extern int core_cluster_api_register(struct se_cluster_api *);
-- 
2.1.4

[RFC PATCH 5/5] target/core_sbc: use cluster API hooks for COMPARE AND WRITE

[David Disseldorp]

If a device cluster API is configured, then defer COMPARE AND WRITE lock
handling to the cluster back-end using the newly added caw_[un]lock()
hooks.

Signed-off-by: David Disseldorp <ddiss@suse.de>
---
 drivers/target/target_core_sbc.c | 28 ++++++++++++++++++++++++----
 1 file changed, 24 insertions(+), 4 deletions(-)

diff --git a/drivers/target/target_core_sbc.c b/drivers/target/target_core_sbc.c
index 751b935..d94aaf4 100644
--- a/drivers/target/target_core_sbc.c
+++ b/drivers/target/target_core_sbc.c
@@ -31,6 +31,7 @@
 #include <target/target_core_base.h>
 #include <target/target_core_backend.h>
 #include <target/target_core_fabric.h>
+#include <target/target_core_cluster.h>
 
 #include "target_core_internal.h"
 #include "target_core_ua.h"
@@ -405,6 +406,24 @@ sbc_execute_rw(struct se_cmd *cmd)
 			       cmd->data_direction);
 }
 
+static int
+sbc_caw_lock(struct se_device *dev)
+{
+	if (dev->cluster_api)
+		return dev->cluster_api->caw_lock(dev);
+
+	return down_interruptible(&dev->caw_sem);
+}
+
+static void
+sbc_caw_unlock(struct se_device *dev)
+{
+	if (dev->cluster_api)
+		dev->cluster_api->caw_unlock(dev);
+	else
+		up(&dev->caw_sem);
+}
+
 static sense_reason_t compare_and_write_post(struct se_cmd *cmd, bool success)
 {
 	struct se_device *dev = cmd->se_dev;
@@ -423,7 +442,7 @@ static sense_reason_t compare_and_write_post(struct se_cmd *cmd, bool success)
 	 * Unlock ->caw_sem originally obtained during sbc_compare_and_write()
 	 * before the original READ I/O submission.
 	 */
-	up(&dev->caw_sem);
+	sbc_caw_unlock(dev);
 
 	return TCM_NO_SENSE;
 }
@@ -571,7 +590,7 @@ out:
 	 * In the MISCOMPARE or failure case, unlock ->caw_sem obtained in
 	 * sbc_compare_and_write() before the original READ I/O submission.
 	 */
-	up(&dev->caw_sem);
+	sbc_caw_unlock(dev);
 	kfree(write_sg);
 	kfree(buf);
 	return ret;
@@ -588,11 +607,12 @@ sbc_compare_and_write(struct se_cmd *cmd)
 	 * Submit the READ first for COMPARE_AND_WRITE to perform the
 	 * comparision using SGLs at cmd->t_bidi_data_sg..
 	 */
-	rc = down_interruptible(&dev->caw_sem);
+	rc = sbc_caw_lock(dev);
 	if (rc != 0) {
 		cmd->transport_complete_callback = NULL;
 		return TCM_LOGICAL_UNIT_COMMUNICATION_FAILURE;
 	}
+
 	/*
 	 * Reset cmd->data_length to individual block_size in order to not
 	 * confuse backend drivers that depend on this value matching the
@@ -604,7 +624,7 @@ sbc_compare_and_write(struct se_cmd *cmd)
 			      DMA_FROM_DEVICE);
 	if (ret) {
 		cmd->transport_complete_callback = NULL;
-		up(&dev->caw_sem);
+		sbc_caw_unlock(dev);
 		return ret;
 	}
 	/*
-- 
2.1.4

Re: [RFC PATCH 0/5] rbd_tcm cluster COMPARE AND WRITE

[Christoph Hellwig]

Hi David,

please introduce a proper compare and write API at the block layer
instead of bypassing it.  Thanks!

Re: [RFC PATCH 0/5] rbd_tcm cluster COMPARE AND WRITE

[Mike Christie]

On 07/28/2015 08:15 PM, David Disseldorp wrote:
> Hi Mike,
> 
> This RFC patch-set expands on your recently proposed cluster API changes
> to add rbd_tcm COMPARE AND WRITE support.
> 
> Rather than offloading the entire COMPARE AND WRITE operation to the
> cluster back-end, these changes extend on the existing local semaphore
> usage, to additionally take an RBD COMPARE AND WRITE device lock. RBD
> device lock acquisition is handled via a newly added rbd_tcm workqueue.
> 
> I'd appreciate feedback on the overall direction of these changes.
> 

I will take a look at them. I just posted my patches too :)

I originally did not want to use cluster locking because I did not want
to have to grab a cluster lock every time we execute the command. I
tried to take advantage that we know which OSD the compare and write is
going to go to, so we can just send the op to it and it can do locking
on the OSD side.

Re: [RFC PATCH 0/5] rbd_tcm cluster COMPARE AND WRITE

[David Disseldorp]

On Wed, 29 Jul 2015 04:31:41 -0500, Mike Christie wrote:

> On 07/28/2015 08:15 PM, David Disseldorp wrote:
> > Hi Mike,
> > 
> > This RFC patch-set expands on your recently proposed cluster API changes
> > to add rbd_tcm COMPARE AND WRITE support.
> > 
> > Rather than offloading the entire COMPARE AND WRITE operation to the
> > cluster back-end, these changes extend on the existing local semaphore
> > usage, to additionally take an RBD COMPARE AND WRITE device lock. RBD
> > device lock acquisition is handled via a newly added rbd_tcm workqueue.
> > 
> > I'd appreciate feedback on the overall direction of these changes.
> > 
> 
> I will take a look at them. I just posted my patches too :)

Thanks for posting your new patch-set.

> I originally did not want to use cluster locking because I did not want
> to have to grab a cluster lock every time we execute the command. I
> tried to take advantage that we know which OSD the compare and write is
> going to go to, so we can just send the op to it and it can do locking
> on the OSD side.

Offloading the entire read/compare/write operation to the OSD is *much*
nicer. Consider my RFC patch-set revoked :)

Cheers, David