thread: add spdk_thread_exec_msg()

A common pattern is:

	if (foo->thread == spdk_get_thread())
		cb(arg);
	else
		spdk_thread_send_msg(foo->thread, cb, arg);

for cases where it's important the callback runs on a particular thread,
but it doesn't matter if it's synchronous or asynchronous.

Add a new API to support this pattern, and convert over the current
instances.

Signed-off-by: John Levon <john.levon@nutanix.com>
Change-Id: Idfbf77c02c9321c52e07181ffd8b0c437e1ab335
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/11503
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Community-CI: Broadcom CI <spdk-ci.pdl@broadcom.com>
Community-CI: Mellanox Build Bot
Reviewed-by: Changpeng Liu <changpeng.liu@intel.com>
Reviewed-by: Konrad Sztyber <konrad.sztyber@intel.com>

CHANGELOG.md

@@ -16,6 +16,10 @@ remove NVMe error injections.
 Removed deprecated max_qpairs_per_ctrlr parameter from nvmf_create_transport RPC. Use
 max_io_qpairs_per_ctrlr instead.
 
+### thread
+
+Added `spdk_thread_exec_msg()` API.
+
 ## v22.01
 
 ### accel

include/spdk/thread.h

@@ -502,6 +502,32 @@ int spdk_thread_send_msg(const struct spdk_thread *thread, spdk_msg_fn fn, void
  */
 int spdk_thread_send_critical_msg(struct spdk_thread *thread, spdk_msg_fn fn);
 
+/**
+ * Run the msg callback on the given thread. If this happens to be the current
+ * thread, the callback is executed immediately; otherwise a message is sent to
+ * the thread, and it's run asynchronously.
+ *
+ * \param thread The target thread.
+ * \param fn This function will be called on the given thread.
+ * \param ctx This context will be passed to fn when called.
+ *
+ * \return 0 on success
+ * \return -ENOMEM if the message could not be allocated
+ * \return -EIO if the message could not be sent to the destination thread
+ */
+static inline int
+spdk_thread_exec_msg(const struct spdk_thread *thread, spdk_msg_fn fn, void *ctx)
+{
+	assert(thread != NULL);
+
+	if (spdk_get_thread() == thread) {
+		fn(ctx);
+		return 0;
+	}
+
+	return spdk_thread_send_msg(thread, fn, ctx);
+}
+
 /**
  * Send a message to each thread, serially.
  *

lib/nvmf/ctrlr.c

@@ -4180,12 +4180,7 @@ spdk_nvmf_request_complete(struct spdk_nvmf_request *req)
 {
 	struct spdk_nvmf_qpair *qpair = req->qpair;
 
-	if (spdk_likely(qpair->group->thread == spdk_get_thread())) {
+	spdk_thread_exec_msg(qpair->group->thread, _nvmf_request_complete, req);
-		_nvmf_request_complete(req);
-	} else {
-		spdk_thread_send_msg(qpair->group->thread,
-				     _nvmf_request_complete, req);
-	}
 
 	return 0;
 }

lib/nvmf/vfio_user.c

@@ -3128,11 +3128,7 @@ free_ctrlr(struct nvmf_vfio_user_ctrlr *ctrlr)
 		free_qp(ctrlr, i);
 	}
 
-	if (ctrlr->thread == spdk_get_thread()) {
+	spdk_thread_exec_msg(ctrlr->thread, _free_ctrlr, ctrlr);
-		_free_ctrlr(ctrlr);
-	} else {
-		spdk_thread_send_msg(ctrlr->thread, _free_ctrlr, ctrlr);
-	}
 }
 
 static int

lib/scsi/lun.c

@@ -302,19 +302,13 @@ static void
 scsi_lun_remove(struct spdk_scsi_lun *lun)
 {
 	struct spdk_scsi_pr_registrant *reg, *tmp;
-	struct spdk_thread *thread;
 
 	TAILQ_FOREACH_SAFE(reg, &lun->reg_head, link, tmp) {
 		TAILQ_REMOVE(&lun->reg_head, reg, link);
 		free(reg);
 	}
 
-	thread = spdk_get_thread();
+	spdk_thread_exec_msg(lun->thread, _scsi_lun_remove, lun);
-	if (thread != lun->thread) {
-		spdk_thread_send_msg(lun->thread, _scsi_lun_remove, lun);
-	} else {
-		_scsi_lun_remove(lun);
-	}
 }
 
 static int

module/bdev/compress/vbdev_compress.c

@@ -455,11 +455,8 @@ reduce_rw_blocks_cb(void *arg, int reduce_errno)
 
 	/* Send this request to the orig IO thread. */
 	orig_thread = spdk_io_channel_get_thread(ch);
-	if (orig_thread != spdk_get_thread()) {
+
-		spdk_thread_send_msg(orig_thread, _reduce_rw_blocks_cb, io_ctx);
+	spdk_thread_exec_msg(orig_thread, _reduce_rw_blocks_cb, io_ctx);
-	} else {
-		_reduce_rw_blocks_cb(io_ctx);
-	}
 }
 
 static uint64_t