bdev/rbd: round-robin reactor selection for SpdkContextWQ

Reactor threads are discovered and cached, and each
new RBD bdev gets the next reactor in round-robin order
so SpdkContextWQ work is spread across cores instead of a single reactor.

Signed-off-by: Alexander Indenbaum <aindenba@redhat.com>

Author: Alexander Indenbaum

module/bdev/rbd/bdev_rbd.c

@@ -21,10 +21,6 @@
 #include "spdk/bdev_module.h"
 #include "spdk/log.h"
 
-/* to access reactor framework */
-#include "spdk_internal/event.h"
-#include "spdk_internal/thread.h"
-
 SPDK_LOG_REGISTER_COMPONENT(reservation)
 
 static int bdev_rbd_count = 0;
@@ -478,9 +474,6 @@ bdev_rbd_get_pool_ctx(rados_t *cluster_p, const char *name, const char *namespac
 	return -1;
 }
 
-/* Forward declaration for reactor thread selection function */
-static struct spdk_thread *bdev_rbd_find_reactor_thread(void);
-
 static void *
 bdev_rbd_init_context(void *arg)
 {
@@ -761,46 +754,6 @@ static struct spdk_bdev_module rbd_if = {
 };
 SPDK_BDEV_MODULE_REGISTER(rbd, &rbd_if)
 
-/**
- * Find a reactor thread for SpdkContextWQ.
- * Iterates through reactors to find a reactor thread (not app_thread).
- * Returns NULL if no suitable thread is found.
- */
-static struct spdk_thread *
-bdev_rbd_find_reactor_thread(void)
-{
-	uint32_t lcore;
-
-	SPDK_ENV_FOREACH_CORE(lcore) {
-		struct spdk_reactor *reactor = spdk_reactor_get(lcore);
-		if (reactor == NULL || !reactor->flags.is_valid) {
-			continue;
-		}
-
-		if (reactor->thread_count == 0) {
-			continue;
-		}
-
-		struct spdk_lw_thread *lw_thread = TAILQ_FIRST(&reactor->threads);
-		if (lw_thread == NULL) {
-			continue;
-		}
-
-		struct spdk_thread *thread = spdk_thread_get_from_ctx(lw_thread);
-		if (thread == NULL || spdk_thread_is_app_thread(thread)) {
-			continue;
-		}
-
-		uint64_t thread_id = spdk_thread_get_id(thread);
-		const char *thread_name = spdk_thread_get_name(thread);
-		SPDK_NOTICELOG("bdev_rbd_find_reactor_thread: reactor thread lcore=%u: thread=%p, id=%lu, name=%s\n",
-			       lcore, thread, thread_id, thread_name ? thread_name : "NULL");
-		return thread;
-	}
-
-	return NULL;
-}
-
 static int bdev_rbd_reset_timer(void *arg);
 
 static void

module/bdev/rbd/bdev_rbd_spdk_context_wq.cpp

@@ -7,6 +7,7 @@
 #include <rbd/librbd.h>
 #include <rados/librados.hpp>
 #include <atomic>
+#include <vector>
 
 #include "bdev_rbd_spdk_context_wq.h"
 
@@ -15,8 +16,80 @@ extern "C" {
 #include "spdk/thread.h"
 #include "spdk/log.h"
 #include "spdk/env.h"
+#include "spdk_internal/event.h"
 }
 
+/**
+ * Encapsulates the set of reactor threads for round-robin assignment.
+ */
+class ReactorThreadPool {
+public:
+  /** Ensures reactor list exists (lock-free one-time init via CAS). */
+  static void ensure_discovered() {
+    if (g_reactor_list.load(std::memory_order_acquire) != nullptr) {
+      return;
+    }
+    auto *p = new std::vector<struct spdk_thread *>();
+    discover_into(p);
+    std::vector<struct spdk_thread *> *expected = nullptr;
+    if (!g_reactor_list.compare_exchange_strong(expected, p, std::memory_order_release)) {
+      delete p;
+    }
+  }
+
+  static struct spdk_thread *get_next() {
+    std::vector<struct spdk_thread *> *list =
+        g_reactor_list.load(std::memory_order_acquire);
+    if (list == nullptr || list->empty()) {
+      SPDK_ERRLOG("bdev_rbd: reactor thread pool is empty, no reactor threads available for SpdkContextWQ\n");
+      return NULL;
+    }
+    size_t n = list->size();
+    size_t idx = g_reactor_next.fetch_add(1, std::memory_order_relaxed) % n;
+    struct spdk_thread *t = (*list)[idx];
+    const char *name = spdk_thread_get_name(t);
+    SPDK_NOTICELOG("bdev_rbd: next reactor thread=%p (id=%lu, name=%s, index=%zu/%zu)\n",
+                   t, spdk_thread_get_id(t), name ? name : "NULL", idx, n);
+    return t;
+  }
+
+private:
+  static void discover_into(std::vector<struct spdk_thread *> *vec) {
+    uint32_t lcore;
+    SPDK_ENV_FOREACH_CORE(lcore) {
+      struct spdk_reactor *reactor = spdk_reactor_get(lcore);
+      if (reactor == NULL || !reactor->flags.is_valid) {
+        continue;
+      }
+      if (reactor->thread_count == 0) {
+        continue;
+      }
+      struct spdk_lw_thread *lw_thread = TAILQ_FIRST(&reactor->threads);
+      if (lw_thread == NULL) {
+        continue;
+      }
+      struct spdk_thread *thread = spdk_thread_get_from_ctx(lw_thread);
+      if (thread == NULL || spdk_thread_is_app_thread(thread)) {
+        continue;
+      }
+      const char *name = spdk_thread_get_name(thread);
+      SPDK_NOTICELOG("bdev_rbd: discovered reactor thread=%p (id=%lu, name=%s, index=%zu)\n",
+                     thread, spdk_thread_get_id(thread), name ? name : "NULL", vec->size());
+      vec->push_back(thread);
+    }
+    if (!vec->empty()) {
+      SPDK_NOTICELOG("bdev_rbd: reactor thread pool: discovered %zu reactor(s) for round-robin\n",
+                     vec->size());
+    }
+  }
+
+  static std::atomic<std::vector<struct spdk_thread *> *> g_reactor_list;
+  static std::atomic<uint32_t> g_reactor_next;
+};
+
+std::atomic<std::vector<struct spdk_thread *> *> ReactorThreadPool::g_reactor_list{nullptr};
+std::atomic<uint32_t> ReactorThreadPool::g_reactor_next{0};
+
 namespace librbd {
 namespace asio {
 
@@ -173,4 +246,10 @@ void bdev_rbd_spdk_context_wq_destroy(struct bdev_rbd_spdk_context_wq* context_w
   delete wq;
 }
 
+struct spdk_thread* bdev_rbd_find_reactor_thread(void)
+{
+  ReactorThreadPool::ensure_discovered();
+  return ReactorThreadPool::get_next();
+}
+
 } // extern "C"

module/bdev/rbd/bdev_rbd_spdk_context_wq.h

@@ -47,6 +47,14 @@ struct bdev_rbd_spdk_context_wq* bdev_rbd_spdk_context_wq_create_from_ioctx(rado
  */
 void bdev_rbd_spdk_context_wq_destroy(struct bdev_rbd_spdk_context_wq* context_wq);
 
+/**
+ * Return the next reactor thread for SpdkContextWQ (round-robin).
+ * Lazy-initializes the reactor list on first call (lock-free, atomic CAS).
+ * Each call returns a different reactor so RBD images are balanced across reactors.
+ * Returns NULL if no reactor threads were discovered (error is logged).
+ */
+struct spdk_thread* bdev_rbd_find_reactor_thread(void);
+
 #ifdef __cplusplus
 }