lib/thread: Add spdk_thread_send_critical_msg function

The patch adds new interface for issuing messages during interrupts,
such as signal handlers. Without this, it'd be possible to deadlock
the application, as two different messages could be trying to enqueue to
the same ring, in the same call stack.

Signed-off-by: Kozlowski Mateusz <mateusz.kozlowski@intel.com>
Change-Id: I917aa41b7f3415af7c7a7d5fa91b964d727609b6
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/478290
Reviewed-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>

include/spdk/thread.h

@@ -381,6 +381,23 @@ int spdk_thread_get_stats(struct spdk_thread_stats *stats);
  */
 int spdk_thread_send_msg(const struct spdk_thread *thread, spdk_msg_fn fn, void *ctx);
 
+/**
+ * Send a message to the given thread. Only one critical message can be outstanding at the same
+ * time. It's intended to use this function in any cases that might interrupt the execution of the
+ * application, such as signal handlers.
+ *
+ * The message will be sent asynchronously - i.e. spdk_thread_send_critical_msg will always return
+ * prior to `fn` being called.
+ *
+ * \param thread The target thread.
+ * \param fn This function will be called on the given thread.
+ *
+ * \return 0 on success
+ * \return -EIO if the message could not be sent to the destination thread, due to an already
+ * outstanding critical message
+ */
+int spdk_thread_send_critical_msg(struct spdk_thread *thread, spdk_msg_fn fn);
+
 /**
  * Send a message to each thread, serially.
  *

lib/thread/thread.c

@@ -34,6 +34,7 @@
 #include "spdk/stdinc.h"
 
 #include "spdk/env.h"
+#include "spdk/likely.h"
 #include "spdk/queue.h"
 #include "spdk/string.h"
 #include "spdk/thread.h"
@@ -149,6 +150,8 @@ struct spdk_thread {
 	SLIST_HEAD(, spdk_msg)		msg_cache;
 	size_t				msg_cache_count;
 
+	spdk_msg_fn			critical_msg;
+
 	/* User context allocated at the end */
 	uint8_t				ctx[0];
 };
@@ -484,6 +487,7 @@ spdk_thread_poll(struct spdk_thread *thread, uint32_t max_msgs, uint64_t now)
 	uint32_t msg_count;
 	struct spdk_thread *orig_thread;
 	struct spdk_poller *poller, *tmp;
+	spdk_msg_fn critical_msg;
 	int rc = 0;
 
 	orig_thread = _get_thread();
@@ -493,6 +497,12 @@ spdk_thread_poll(struct spdk_thread *thread, uint32_t max_msgs, uint64_t now)
 		now = spdk_get_ticks();
 	}
 
+	critical_msg = thread->critical_msg;
+	if (spdk_unlikely(critical_msg != NULL)) {
+		critical_msg(NULL);
+		thread->critical_msg = NULL;
+	}
+
 	msg_count = _spdk_msg_queue_run_batch(thread, max_msgs);
 	if (msg_count) {
 		rc = 1;
@@ -642,7 +652,8 @@ bool
 spdk_thread_is_idle(struct spdk_thread *thread)
 {
 	if (spdk_ring_count(thread->messages) ||
-	    _spdk_thread_has_unpaused_pollers(thread)) {
+	    _spdk_thread_has_unpaused_pollers(thread) ||
+	    thread->critical_msg != NULL) {
 		return false;
 	}
 
@@ -741,6 +752,19 @@ spdk_thread_send_msg(const struct spdk_thread *thread, spdk_msg_fn fn, void *ctx
 	return 0;
 }
 
+int
+spdk_thread_send_critical_msg(struct spdk_thread *thread, spdk_msg_fn fn)
+{
+	spdk_msg_fn expected = NULL;
+
+	if (__atomic_compare_exchange_n(&thread->critical_msg, &expected, fn, false, __ATOMIC_SEQ_CST,
+					__ATOMIC_SEQ_CST)) {
+		return 0;
+	}
+
+	return -EIO;
+}
+
 struct spdk_poller *
 spdk_poller_register(spdk_poller_fn fn,
 		     void *arg,