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..7dff13d7 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,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 *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) { + // 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; +} + 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_); + } } } 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 {