feat(eloq): key module registry by type and add a configurable visit order

src/bthread/eloq_module.cpp

@@ -26,24 +26,60 @@ extern "C" {
 bthread::TaskControl *bthread_get_task_control();
 }
 
-extern std::array<eloq::EloqModule *, 10> registered_modules;
+extern std::array<eloq::EloqModule *, eloq::kModuleTypeCount> registered_modules;
 extern std::atomic<int> registered_module_cnt;
 extern std::atomic<uint64_t> registered_module_version;
 
 namespace eloq {
+    namespace {
+        struct ModuleTypeNameEntry {
+            ModuleType type_;
+            const char *name_;
+        };
+        constexpr ModuleTypeNameEntry kModuleTypeNames[] = {
+            {ModuleType::kRing, "ring"},
+            {ModuleType::kTxService, "txservice"},
+            {ModuleType::kEloqStore, "eloqstore"},
+        };
+        static_assert(sizeof(kModuleTypeNames) / sizeof(kModuleTypeNames[0]) ==
+                      kModuleTypeCount);
+    }  // namespace
+
+    const char *ModuleTypeName(ModuleType type) {
+        for (const auto &entry : kModuleTypeNames) {
+            if (entry.type_ == type) {
+                return entry.name_;
+            }
+        }
+        return "unknown";
+    }
+
+    bool ParseModuleTypeName(const std::string &name, ModuleType *type) {
+        for (const auto &entry : kModuleTypeNames) {
+            if (name == entry.name_) {
+                *type = entry.type_;
+                return true;
+            }
+        }
+        return false;
+    }
+
     bool EloqModule::NotifyWorker(int thd_id) {
         return bthread_notify_worker(thd_id);
     }
 
     int register_module(EloqModule *module) {
+        // The module's type is its slot, so the registry never shifts and a
+        // slot always denotes the same kind of module.
+        const size_t slot = static_cast<size_t>(module->Type());
+        CHECK_LT(slot, registered_modules.size());
         std::unique_lock lk(module_mutex);
-        size_t i = 0;
-        while (i < registered_modules.size() && registered_modules[i] != nullptr) {
-            // Each module should only be registered once.
-            CHECK(registered_modules[i] != module);
-            i++;
-        }
-        registered_modules[i] = module;
+        // A module type is a singleton; registering a second instance while
+        // the first is live would silently displace it.
+        CHECK(registered_modules[slot] == nullptr)
+                << "module type " << ModuleTypeName(module->Type())
+                << " is already registered";
+        registered_modules[slot] = module;
         registered_module_cnt.fetch_add(1, std::memory_order_release);
         registered_module_version.fetch_add(1, std::memory_order_release);
         const auto non_null_modules =
@@ -76,19 +112,15 @@ namespace eloq {
             bthread_usleep(1000);
         }
         std::unique_lock lk(module_mutex);
-        size_t i = 0;
-        while (i < registered_modules.size() && registered_modules[i] != module) {
-            i++;
-        }
-        if (i == registered_modules.size()) {
+        const size_t slot = static_cast<size_t>(module->Type());
+        if (slot >= registered_modules.size() ||
+            registered_modules[slot] != module) {
             return 0;
         }
-        CHECK(i < registered_module_cnt);
-        while (i < registered_modules.size() - 1) {
-            registered_modules[i] = registered_modules[i + 1];
-            i++;
-        }
-        registered_modules[registered_modules.size() - 1] = nullptr;
+        // Clear the slot in place. Compacting the array would renumber every
+        // higher module, so a slot would stop denoting the same module across
+        // an unregister -- which is what --module_visit_order addresses.
+        registered_modules[slot] = nullptr;
         registered_module_cnt.fetch_sub(1, std::memory_order_release);
         registered_module_version.fetch_add(1, std::memory_order_release);
         const auto non_null_modules =

src/bthread/eloq_module.h

@@ -21,14 +21,51 @@
 #define ELOQ_MODULE_H
 
 #include <atomic>
+#include <cstddef>
 #include <shared_mutex>
+#include <string>
 
 namespace eloq {
     inline std::shared_mutex module_mutex;
+
+    /**
+     * Identifies what a module is, independently of when it registers. The
+     * enumerator is also the module's slot in the registry, so a module always
+     * occupies the same slot: the mapping survives a module being absent (e.g.
+     * RingModule when io_uring is off) and a module restarting (e.g. EloqStore
+     * reopening, which happens during normal startup).
+     *
+     * The order is the order the modules settle into after startup, so visiting
+     * every registered slot once in ascending order is the order workers have
+     * always used.
+     */
+    enum class ModuleType : size_t {
+        kRing = 0,
+        kTxService = 1,
+        kEloqStore = 2,
+    };
+
+    inline constexpr size_t kModuleTypeCount = 3;
+
+    /** @brief Stable name of a module type, e.g. "eloqstore". */
+    const char *ModuleTypeName(ModuleType type);
+
+    /**
+     * @brief Maps a module type name back to its enumerator.
+     * @return true if the name is known, in which case *type is set.
+     */
+    bool ParseModuleTypeName(const std::string &name, ModuleType *type);
+
     class EloqModule {
     public:
         virtual ~EloqModule() = default;
 
+        /**
+         * What this module is. Determines the module's slot in the registry
+         * and the name --module_visit_order uses to refer to it.
+         */
+        virtual ModuleType Type() const = 0;
+
         /**
          * This func is called when worker starts running.
          * @param thd_id

src/bthread/ring_module.h

@@ -28,6 +28,10 @@ class RingListener;
 
 class RingModule : public eloq::EloqModule {
 public:
+    eloq::ModuleType Type() const override {
+        return eloq::ModuleType::kRing;
+    }
+
     void ExtThdStart(int thd_id) override;
 
     void ExtThdEnd(int thd_id) override;

src/bthread/task_group.cpp

@@ -21,6 +21,8 @@
 
 #include <sys/types.h>
 #include <stddef.h>                         // size_t
+#include <stdint.h>                         // uint8_t
+#include <string>
 #include <typeinfo>
 #include <gflags/gflags.h>
 #include "butil/compat.h"                   // OS_MACOSX
@@ -45,7 +47,7 @@
 #include "bthread/ring_listener.h"
 #endif
 
-std::array<eloq::EloqModule *, 10> registered_modules;
+std::array<eloq::EloqModule *, eloq::kModuleTypeCount> registered_modules;
 std::atomic<int> registered_module_cnt;
 std::atomic<uint64_t> registered_module_version;
 
@@ -77,6 +79,14 @@ DEFINE_int32(worker_polling_time_us, 0, "Worker keep busy polling some time befo
                                        "sleeping on parking lot");
 DEFINE_int32(module_process_latency_log_threshold_us, 0,
              "Log module process latency longer than this threshold (0 disables logging)");
+DEFINE_string(module_visit_order, "",
+              "Comma-separated module names giving the visit order of one "
+              "ProcessModulesTask pass. Known names are ring, txservice and "
+              "eloqstore. A name may repeat, which drives that module more "
+              "than once per pass; naming a module that is not registered is "
+              "harmless, it is skipped. Empty selects the default order "
+              "\"ring,eloqstore,txservice,eloqstore\"; pass "
+              "\"ring,txservice,eloqstore\" for one visit each.");
 
 BAIDU_VOLATILE_THREAD_LOCAL(TaskGroup*, tls_task_group, NULL);
 // Sync with TaskMeta::local_storage when a bthread is created or destroyed.
@@ -286,7 +296,59 @@ TaskGroup::~TaskGroup() {
     }
 }
 
+// Resolves FLAGS_module_visit_order into registry slots, returning how many
+// were written. An unparsable entry aborts rather than silently visiting the
+// wrong module: the flag exists to control which module runs when, so a typo
+// must not look like it worked. An empty flag yields every slot once in
+// module-type order, which is the order workers use when the flag is unset.
+static size_t ResolveModuleVisitOrder(
+        std::array<uint8_t, TaskGroup::kMaxModuleVisits> *order) {
+    const std::string &spec = FLAGS_module_visit_order;
+    size_t cnt = 0;
+    size_t pos = 0;
+    while (pos < spec.size()) {
+        size_t comma = spec.find(',', pos);
+        if (comma == std::string::npos) {
+            comma = spec.size();
+        }
+        const std::string name = spec.substr(pos, comma - pos);
+        eloq::ModuleType type;
+        CHECK(eloq::ParseModuleTypeName(name, &type))
+                << "unknown module name \"" << name << "\" in "
+                << "--module_visit_order=" << spec;
+        CHECK_LT(cnt, order->size())
+                << "--module_visit_order has more than " << order->size()
+                << " entries: " << spec;
+        (*order)[cnt++] = static_cast<uint8_t>(type);
+        pos = comma + 1;
+    }
+    if (cnt == 0) {
+        // Default: drive EloqStore twice per pass, once before and once after
+        // the tx service. A shard accumulates completed IO faster than a single
+        // visit per pass can drain, so the second visit shortens the interval
+        // between an IO completing and the shard draining it. Benchmarking
+        // shows this ahead of one-visit-each on read-heavy load and no worse on
+        // a mixed one. A module that is not registered -- EloqStore under a
+        // different storage backend, Ring without io_uring -- is skipped by the
+        // visit loop, so naming it here costs a null check.
+        static constexpr eloq::ModuleType kDefaultVisitOrder[] = {
+            eloq::ModuleType::kRing,
+            eloq::ModuleType::kEloqStore,
+            eloq::ModuleType::kTxService,
+            eloq::ModuleType::kEloqStore,
+        };
+        static_assert(sizeof(kDefaultVisitOrder) /
+                              sizeof(kDefaultVisitOrder[0]) <=
+                      TaskGroup::kMaxModuleVisits);
+        for (eloq::ModuleType type : kDefaultVisitOrder) {
+            (*order)[cnt++] = static_cast<uint8_t>(type);
+        }
+    }
+    return cnt;
+}
+
 int TaskGroup::init(size_t runqueue_capacity) {
+    module_visit_cnt_ = ResolveModuleVisitOrder(&module_visit_order_);
     if (_rq.init(runqueue_capacity) != 0) {
         LOG(FATAL) << "Fail to init _rq";
         return -1;
@@ -1269,7 +1331,10 @@ bool TaskGroup::Wait(){
         }
 
         // Check any module registered or deleted before checking modules' tasks.
-        CheckAndUpdateModules();
+        // This worker is going to sleep, so it must not start anything: the
+        // ExtThdStart() for whatever registers now comes from
+        // NotifyRegisteredModules(Working) when it wakes.
+        CheckAndUpdateModules(false);
         return HasTasks();
     };
 
@@ -1305,18 +1370,17 @@ bool TaskGroup::Wait(){
 }
 
 void TaskGroup::ProcessModulesTask() {
-    int old_modules_cnt = modules_cnt_;
-
-    CheckAndUpdateModules();
-
-    int new_modules_cnt = modules_cnt_;
-    for (int i = old_modules_cnt; i < new_modules_cnt; ++i) {
-        eloq::EloqModule *module = registered_modules_[i];
-        module->ExtThdStart(group_id_);
-    }
+    // Starts modules that are new to this worker. Slots are keyed by module
+    // type and can therefore be sparse, so a module cannot be located by
+    // counting: CheckAndUpdateModules() already diffs the module set and is
+    // the only place that knows which entries are new. This worker is
+    // running, so a module registering now missed the ExtThdStart() that
+    // NotifyRegisteredModules() issues on wake-up and must be started here.
+    CheckAndUpdateModules(true);
 
     const int32_t log_threshold_us = FLAGS_module_process_latency_log_threshold_us;
-    for (auto *module : registered_modules_) {
+    for (size_t i = 0; i < module_visit_cnt_; ++i) {
+        eloq::EloqModule *module = registered_modules_[module_visit_order_[i]];
         if (module == nullptr) {
             continue;
         }
@@ -1347,7 +1411,7 @@ bool TaskGroup::HasTasks() {
     return has_task;
 }
 
-void TaskGroup::CheckAndUpdateModules() {
+void TaskGroup::CheckAndUpdateModules(bool start_new_modules) {
     const uint64_t global_version =
         registered_module_version.load(std::memory_order_acquire);
     if (modules_version_ != global_version) {
@@ -1374,6 +1438,12 @@ void TaskGroup::CheckAndUpdateModules() {
             if (!found) {
                 new_m->registered_workers_.fetch_add(
                     1, std::memory_order_relaxed);
+                // Modules bind per-worker state here (EloqStoreModule binds
+                // the thread's shard), so a module that is never started on a
+                // worker cannot run on it at all.
+                if (start_new_modules) {
+                    new_m->ExtThdStart(group_id_);
+                }
             }
         }
 

src/bthread/task_group.h

@@ -227,10 +227,19 @@ class TaskGroup {
     std::function<bool(bool)> override_shard_heap_{nullptr};
     std::function<bool()> has_tx_processor_work_{nullptr};
 
-    std::array<eloq::EloqModule *, 10> registered_modules_{};
+    std::array<eloq::EloqModule *, eloq::kModuleTypeCount> registered_modules_{};
     int modules_cnt_{0};
     uint64_t modules_version_{0};
 
+    // Registry slots to visit in one ProcessModulesTask() pass, resolved from
+    // --module_visit_order once in init(). Held inline and pre-resolved
+    // because ProcessModulesTask() runs millions of times a second: reading it
+    // must cost no more than the plain array walk it replaced. A slot may
+    // repeat, so the order can be longer than the number of modules.
+    static constexpr size_t kMaxModuleVisits = 8;
+    std::array<uint8_t, kMaxModuleVisits> module_visit_order_{};
+    size_t module_visit_cnt_{0};
+
 #ifdef IO_URING_ENABLED
     int RegisterSocket(SocketRegisterData *data);
     int UnregisterSocket(SocketUnRegisterData *data);
@@ -305,7 +314,9 @@ class TaskGroup {
 
     bool HasTasks();
 
-    void CheckAndUpdateModules();
+    // start_new_modules: call ExtThdStart() on modules new to this worker.
+    // Only safe when the worker is running, not when it is going to sleep.
+    void CheckAndUpdateModules(bool start_new_modules);
 
     enum struct WorkerStatus
     {