From d2158a4ac27d64c03170e8156f09f31caf95b3f0 Mon Sep 17 00:00:00 2001 From: Liang Jeff Chen Date: Mon, 17 Aug 2026 00:23:43 -0700 Subject: [PATCH 1/2] feat(eloq): key module registry by type and add a configurable visit order register_module() assigned the first free slot and unregister_module() compacted the array, so a module's index depended on registration order and shifted whenever another module unregistered. EloqStore restarts once during normal startup, which renumbered TxService underneath it: the registry settled at ring/txservice/eloqstore only after that shuffle. Indices were therefore unusable for addressing a module. EloqModule now declares a ModuleType, and the enumerator is the slot: kRing = 0, kTxService = 1, kEloqStore = 2 Registration writes its own slot and unregistration clears it in place, so a slot always denotes the same kind of module -- across a module being absent (RingModule without io_uring) and across a module restarting. Slots may therefore be sparse, which is what makes them stable. On top of that, --module_visit_order takes module names giving the visit order of one ProcessModulesTask() pass, e.g. --module_visit_order=ring,eloqstore,txservice,eloqstore A name may repeat, which drives that module more than once per pass. The order is resolved once in TaskGroup::init() into an inline fixed array, because ProcessModulesTask() runs millions of times a second and must not pay a guard check or a heap indirection per pass. An unknown name aborts at startup rather than silently visiting the wrong module. Empty keeps the previous behavior: every registered module once, in slot order. Sparse slots also mean a module can no longer be located by counting, so ExtThdStart() for modules new to a worker moves into CheckAndUpdateModules(), which already diffs the module set and runs only when that set changes. The old loop indexed registered_modules_ with a population count and would skip -- or dereference past -- an empty slot; EloqStore binds its per-thread shard in ExtThdStart(), so skipping it left the thread-local shard null and io_uring initialization aborted. Co-Authored-By: Claude Opus 5 (1M context) --- src/bthread/eloq_module.cpp | 70 ++++++++++++++++++++++++--------- src/bthread/eloq_module.h | 37 ++++++++++++++++++ src/bthread/ring_module.h | 4 ++ src/bthread/task_group.cpp | 78 ++++++++++++++++++++++++++++++------- src/bthread/task_group.h | 15 ++++++- 5 files changed, 170 insertions(+), 34 deletions(-) diff --git a/src/bthread/eloq_module.cpp b/src/bthread/eloq_module.cpp index 90dac6e9..95633a7a 100644 --- a/src/bthread/eloq_module.cpp +++ b/src/bthread/eloq_module.cpp @@ -26,24 +26,60 @@ extern "C" { bthread::TaskControl *bthread_get_task_control(); } -extern std::array registered_modules; +extern std::array registered_modules; extern std::atomic registered_module_cnt; extern std::atomic 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(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(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 = diff --git a/src/bthread/eloq_module.h b/src/bthread/eloq_module.h index 97d31dd8..41dc8bc8 100644 --- a/src/bthread/eloq_module.h +++ b/src/bthread/eloq_module.h @@ -21,14 +21,51 @@ #define ELOQ_MODULE_H #include +#include #include +#include 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 diff --git a/src/bthread/ring_module.h b/src/bthread/ring_module.h index e12a75a4..f9b8d2b0 100644 --- a/src/bthread/ring_module.h +++ b/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; diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 4b0cc663..158de855 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -21,6 +21,8 @@ #include #include // size_t +#include // uint8_t +#include #include #include #include "butil/compat.h" // OS_MACOSX @@ -45,7 +47,7 @@ #include "bthread/ring_listener.h" #endif -std::array registered_modules; +std::array registered_modules; std::atomic registered_module_cnt; std::atomic registered_module_version; @@ -77,6 +79,13 @@ 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, e.g. \"ring,eloqstore,txservice," + "eloqstore\". 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. Empty " + "keeps the default: every module once, in module-type order."); BAIDU_VOLATILE_THREAD_LOCAL(TaskGroup*, tls_task_group, NULL); // Sync with TaskMeta::local_storage when a bthread is created or destroyed. @@ -286,7 +295,42 @@ 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 *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(type); + pos = comma + 1; + } + if (cnt == 0) { + for (size_t i = 0; i < eloq::kModuleTypeCount; ++i) { + (*order)[cnt++] = static_cast(i); + } + } + 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 +1313,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 +1352,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 +1393,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 +1420,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_); + } } } diff --git a/src/bthread/task_group.h b/src/bthread/task_group.h index f6126bd8..0cec1730 100644 --- a/src/bthread/task_group.h +++ b/src/bthread/task_group.h @@ -227,10 +227,19 @@ class TaskGroup { std::function override_shard_heap_{nullptr}; std::function has_tx_processor_work_{nullptr}; - std::array registered_modules_{}; + std::array 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 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 { From 3dc1a0c36f66d5488cdda418964d596cb5cd3a79 Mon Sep 17 00:00:00 2001 From: Liang Jeff Chen Date: Mon, 17 Aug 2026 03:29:54 -0700 Subject: [PATCH 2/2] feat(eloq): default the module visit order to Ring, EloqStore, TxService, EloqStore Benchmarking the previous default (every module once, in slot order) against driving EloqStore twice per pass showed the latter consistently ahead on read-heavy load -- separated from run-to-run spread, with the margin growing as connection count rises -- and no worse on a mixed read/write load, where only the far tail moves. A shard accumulates completed IO faster than one visit per pass can drain, so the second visit shortens the interval between an IO completing and the shard draining it. Make that the built-in default rather than something each deployment has to discover and opt into. --module_visit_order still overrides it, and "ring,txservice,eloqstore" restores one visit each. Naming a module that is not registered stays harmless: slots are keyed by module type, so an absent module leaves its slot empty rather than shifting the others, and the visit loop skips it. A deployment on another storage backend therefore visits Ring and TxService and pays a null check for the two EloqStore entries; the same holds for Ring without io_uring. Co-Authored-By: Claude Opus 5 (1M context) --- src/bthread/task_group.cpp | 32 +++++++++++++++++++++++++------- 1 file changed, 25 insertions(+), 7 deletions(-) diff --git a/src/bthread/task_group.cpp b/src/bthread/task_group.cpp index 158de855..7dff13d7 100644 --- a/src/bthread/task_group.cpp +++ b/src/bthread/task_group.cpp @@ -81,11 +81,12 @@ 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, e.g. \"ring,eloqstore,txservice," - "eloqstore\". 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. Empty " - "keeps the default: every module once, in module-type order."); + "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. @@ -322,8 +323,25 @@ static size_t ResolveModuleVisitOrder( pos = comma + 1; } if (cnt == 0) { - for (size_t i = 0; i < eloq::kModuleTypeCount; ++i) { - (*order)[cnt++] = static_cast(i); + // 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(type); } } return cnt;