From 009a86fa11409f737068464bb912c7f825c0a937 Mon Sep 17 00:00:00 2001 From: Liang Jeff Chen Date: Fri, 16 Oct 2026 18:04:10 +0800 Subject: [PATCH 1/3] fix: run every EloqStore in a process under one brpc module A data store service runs one EloqStore per data shard it leads, so one process may run several stores. Each store registered a brpc module of its own, and task code found its store through one process-global pointer, so a second store replaced or cleared the first one's binding. - EloqStoreModule::Instance(): one module serves every store in the process, registered with brpc when the first store joins. Worker thd_id runs shard thd_id of each store in turn. A store joins at Start (AddStore) and leaves once its shards have stopped (RemoveStore), which waits until no worker is still running one of them; at most kMaxStores (8) run at once. - eloq_store is bound per thread, like shard: on a shard's own thread, on a brpc worker while the module runs one of the store's shards, and on the calling thread for the duration of EloqStore::Start and Stop (ScopedStoreBinding). - Tests: two stores run side by side in one process (eloq_store_test.cpp); persist.cpp no longer depends on the case order the global pointer forced. - Docs: architecture 02 and 10 describe the per-thread binding and the shared module. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/architecture/02-runtime-and-lifecycle.md | 23 ++-- .../architecture/10-bindings-and-embedding.md | 6 +- include/eloq_store.h | 4 +- include/eloqstore_module.h | 69 +++++++++-- include/storage/shard.h | 5 + include/tasks/task.h | 39 +++++- src/eloq_store.cpp | 21 ++-- src/eloqstore_module.cpp | 112 ++++++++++++++++-- src/storage/shard.cpp | 1 + tests/eloq_store_test.cpp | 47 ++++++++ tests/persist.cpp | 9 +- 11 files changed, 296 insertions(+), 40 deletions(-) diff --git a/docs/architecture/02-runtime-and-lifecycle.md b/docs/architecture/02-runtime-and-lifecycle.md index 171dd5c7e..6a6eeb83a 100644 --- a/docs/architecture/02-runtime-and-lifecycle.md +++ b/docs/architecture/02-runtime-and-lifecycle.md @@ -29,11 +29,17 @@ A comment in `eloq_store.h` records one non-obvious ordering constraint: `cloud_service_` is declared before `shards_` because shard teardown triggers async HTTP cleanup that calls back into `CloudStorageService`. -There is also a process-global `inline EloqStore *eloq_store` and a -thread-local `inline thread_local Shard *shard` (`include/tasks/task.h`). -Task code reaches its shard, options, and io manager through these — which is -why **one process hosts at most one running EloqStore** and why task code must -run on a shard thread. +Task code reaches its store, shard, options, and io manager through two +thread-locals, `eloq_store` and `shard` (`include/tasks/task.h`), which is why +task code must run on a shard thread. They are bound per thread, so one process +may run several stores (a data store service runs one per data shard it +leads): a shard's thread binds its shard and store when its loop starts; under +`ELOQ_MODULE_ENABLED` one module serves every store in the process, and worker +`thd_id` runs shard `thd_id` of each store in turn, so +`EloqStoreModule::Process` binds the shard it runs each time; and +`Start` and `Stop` bind their store on the calling thread for their duration +(`ScopedStoreBinding`), for the shard and io manager setup and teardown they +run there. ## Status state machine @@ -66,7 +72,8 @@ the order is load-bearing; later steps assume earlier ones: object; a stale term fails the whole `Start`). Or `StandbyService::Start()`. 8. `Shard::Start()` for each shard — spawns worker threads (normal build) or marks shards runnable (module build). -9. Module build only: register `EloqStoreModule` and notify workers. +9. Module build only: join the process's `EloqStoreModule` (`AddStore`; the + first store to join registers the module with brpc) and notify workers. 10. Start `ArchiveCrond` if `data_append_mode && num_retained_archives > 0 && archive_interval_secs > 0`. 11. Start `PrewarmService` if `prewarm_cloud_cache` and mode is Cloud. @@ -85,7 +92,9 @@ state: 2. Stop `PrewarmService`, then `ArchiveCrond` (background request *sources* stop before request *executors*). 3. Module build: flag shards `Stopping`, wait until each reports `Stopped`, - unregister the module. + then leave the module (`RemoveStore`, which waits until no worker is still + running one of the store's shards). The module stays registered for the + process's other stores. 4. `Shard::Stop()` for each shard — joins the worker thread; the thread's exit path runs `TaskManager::Shutdown()` then `PageManager::Shutdown()` then `io_mgr_->Stop()` *on the shard thread* (these structures assume diff --git a/docs/architecture/10-bindings-and-embedding.md b/docs/architecture/10-bindings-and-embedding.md index 9af0ee1cd..e8a741025 100644 --- a/docs/architecture/10-bindings-and-embedding.md +++ b/docs/architecture/10-bindings-and-embedding.md @@ -57,7 +57,11 @@ only when the embedding runtime (EloqKV's tx service, with bthread) defines - `EloqStoreModule : eloq::EloqModule` exposes `ExtThdStart/ExtThdEnd` (bind/unbind the thread-local `shard`), `Process(thd_id)` (run - `Shard::WorkOneRound`), and `HasTask(thd_id)`. + `Shard::WorkOneRound`), and `HasTask(thd_id)`. One module serves every + store in the process, registered with brpc once: worker `thd_id` runs shard + `thd_id` of each running store in turn, binding that shard and its store + first. Stores join at `Start` (`AddStore`) and leave after their shards + stop (`RemoveStore`, which waits for workers still inside one of them). - Shards own no threads; the host's workers drive shard progress. Shard count must match the host's worker topology (`thd_id` ↔ `shard_id`). - Request sync waits switch from `std::atomic` wait to bthread diff --git a/include/eloq_store.h b/include/eloq_store.h index d09bce2bc..6a57ebc5f 100644 --- a/include/eloq_store.h +++ b/include/eloq_store.h @@ -1050,12 +1050,14 @@ class EloqStore std::unique_ptr prewarm_service_{nullptr}; std::unique_ptr standby_service_{nullptr}; #ifdef ELOQ_MODULE_ENABLED - std::unique_ptr module_{nullptr}; + /** @brief Whether the process's EloqStoreModule runs this store. */ + bool in_module_{false}; #endif bool enable_eloqstore_metrics_{false}; std::atomic store_mode_{StoreMode::Local}; + friend class EloqStoreModule; friend class Shard; friend class KvRequest; friend class AsyncIoManager; diff --git a/include/eloqstore_module.h b/include/eloqstore_module.h index e21d72400..a88883ea1 100644 --- a/include/eloqstore_module.h +++ b/include/eloqstore_module.h @@ -2,22 +2,37 @@ #ifdef ELOQ_MODULE_ENABLED #include -#include +#include +#include +#include +#include #include "storage/shard.h" namespace eloqstore { +class EloqStore; +/** + * @brief The brpc module that runs the shards of every EloqStore in the + * process on brpc workers. + * + * One module serves all running stores: a process may run several (a data + * store service runs one per data shard it leads), and brpc registers few + * modules, each registration or removal waiting for every worker. Worker + * thd_id runs shard thd_id of each store in turn. A store joins at Start + * (AddStore) and leaves once its shards have stopped (RemoveStore), which + * waits until no worker is still running one of them. + */ class EloqStoreModule : public eloq::EloqModule { public: - EloqStoreModule() = default; - explicit EloqStoreModule(std::vector> *shards) - : shards_(shards) - { - } - ~EloqStoreModule() = default; + /** + * @brief The process's module, registered with brpc when the first store + * joins. + * @return The module. + */ + static EloqStoreModule &Instance(); eloq::ModuleType Type() const override { @@ -29,7 +44,45 @@ class EloqStoreModule : public eloq::EloqModule void Process(int thd_id) override; bool HasTask(int thd_id) const override; - std::vector> *shards_; + /** + * @brief Adds a store whose shards the workers start running. + * @param store The store, its shards started. + */ + void AddStore(EloqStore *store); + + /** + * @brief Removes a store, and waits until no worker is running one of its + * shards. Called once the shards have stopped: the workers run them until + * then. + * @param store The store. + */ + void RemoveStore(EloqStore *store); + + /** @brief How many stores may run in one process at once. */ + static constexpr size_t kMaxStores = 8; + +private: + EloqStoreModule() = default; + + /** @brief A store the workers run, and how many are using it now. */ + struct Slot + { + std::atomic store{nullptr}; ///< nullptr: free. + mutable std::atomic users{0}; ///< Workers inside the store now. + }; + + /** + * @brief Calls fn on shard thd_id of each store, until fn returns true. + * @param thd_id The worker, and the shard index. + * @param fn Called as fn(Shard *); true stops the walk. + * @return true if fn returned true. + */ + template + bool ForEachShard(int thd_id, Fn &&fn) const; + + std::array slots_; + std::mutex mux_; ///< Serializes AddStore and RemoveStore. + bool registered_{false}; ///< Registered with brpc; guarded by mux_. }; } // namespace eloqstore diff --git a/include/storage/shard.h b/include/storage/shard.h index fb3740093..e9b18e624 100644 --- a/include/storage/shard.h +++ b/include/storage/shard.h @@ -128,9 +128,14 @@ class Shard task_mgr_.NumActive() == 0 && io_mgr_->IsIdle() && !io_mgr_->NeedPrewarm() && delayed_requests_.empty(); } + /** + * @brief Binds this shard and its store to the calling thread (shard, + * eloq_store), for a brpc worker that runs it. + */ void BindExtThd() { shard = this; + eloq_store = store_; } #endif diff --git a/include/tasks/task.h b/include/tasks/task.h index e80f7c7c2..24966b238 100644 --- a/include/tasks/task.h +++ b/include/tasks/task.h @@ -26,8 +26,45 @@ class MappingSnapshot; class Shard; class EloqStore; -inline EloqStore *eloq_store; +/** + * @brief The store whose work this thread is doing. + * + * A process may run several stores (a data store service runs one per data + * shard it leads), so this is bound per thread, as shard is: on a shard's own + * thread, on a brpc worker each time a module runs one of its shards + * (EloqStoreModule::Process), and on the calling thread for the duration of + * EloqStore::Start and EloqStore::Stop (ScopedStoreBinding). + */ +inline thread_local EloqStore *eloq_store; +/** @brief The shard whose work this thread is doing (bound with eloq_store). */ inline thread_local Shard *shard; + +/** + * @brief Binds eloq_store to a store on this thread for a scope, then restores + * the binding it found. + */ +class ScopedStoreBinding +{ +public: + /** + * @brief Binds eloq_store to the store. + * @param store The store whose work the scope does. + */ + explicit ScopedStoreBinding(EloqStore *store) : prev_(eloq_store) + { + eloq_store = store; + } + ScopedStoreBinding(const ScopedStoreBinding &) = delete; + ScopedStoreBinding &operator=(const ScopedStoreBinding &) = delete; + /** @brief Restores the binding found at construction. */ + ~ScopedStoreBinding() + { + eloq_store = prev_; + } + +private: + EloqStore *prev_; ///< The binding to restore. +}; KvTask *ThdTask(); AsyncIoManager *IoMgr(); const KvOptions *Options(); diff --git a/src/eloq_store.cpp b/src/eloq_store.cpp index 5f2fbe15b..52d37800b 100644 --- a/src/eloq_store.cpp +++ b/src/eloq_store.cpp @@ -583,7 +583,7 @@ void EloqStore::CleanupRuntime(size_t started_shards) archive_crond_->Stop(); } #ifdef ELOQ_MODULE_ENABLED - if (module_ != nullptr) + if (in_module_) { for (size_t i = 0; i < shard_count; ++i) { @@ -614,7 +614,9 @@ void EloqStore::CleanupRuntime(size_t started_shards) } bthread_usleep(1000); } - eloq::unregister_module(module_.get()); + // Stopped: the workers no longer need to run the shards. + EloqStoreModule::Instance().RemoveStore(this); + in_module_ = false; } #endif @@ -649,10 +651,6 @@ void EloqStore::CleanupRuntime(size_t started_shards) { cloud_service_->Stop(); } - if (eloq_store == this) - { - eloq_store = nullptr; - } } KvError EloqStore::Start(std::string_view branch, @@ -660,6 +658,9 @@ KvError EloqStore::Start(std::string_view branch, PartitonGroupId partition_group_id) { LOG(INFO) << "===Start eloqstore, branch: " << branch << ", term: " << term; + // Code reached from here (shard and io manager setup) reads this store + // through the thread's binding. + ScopedStoreBinding store_binding(this); if (!ValidateOptions(options_)) { return KvError::InvalidArgs; @@ -750,7 +751,6 @@ KvError EloqStore::Start(std::string_view branch, standby_service_.reset(); } - eloq_store = this; // Initialize if (!options_.store_path.empty()) { @@ -868,8 +868,8 @@ KvError EloqStore::Start(std::string_view branch, } #ifdef ELOQ_MODULE_ENABLED - module_ = std::make_unique(&shards_); - eloq::register_module(module_.get()); + EloqStoreModule::Instance().AddStore(this); + in_module_ = true; for (auto &shard : shards_) { eloq::EloqModule::NotifyWorker(static_cast(shard->shard_id_)); @@ -2258,6 +2258,9 @@ void EloqStore::ExecSyncOnAnyShard(KvRequest *req) const void EloqStore::Stop() { LOG(INFO) << "EloqStore stopping."; + // Teardown reached from here reads this store through the thread's + // binding. + ScopedStoreBinding store_binding(this); while (true) { Status current = status_.load(std::memory_order_acquire); diff --git a/src/eloqstore_module.cpp b/src/eloqstore_module.cpp index bc2713dc5..5b1b7af26 100644 --- a/src/eloqstore_module.cpp +++ b/src/eloqstore_module.cpp @@ -1,13 +1,61 @@ #include "eloqstore_module.h" +#ifdef ELOQ_MODULE_ENABLED +#include +#include + +#include "eloq_store.h" +#endif + namespace eloqstore { #ifdef ELOQ_MODULE_ENABLED +EloqStoreModule &EloqStoreModule::Instance() +{ + // Never destroyed: brpc workers may visit the module until the process + // exits. + static EloqStoreModule *module = new EloqStoreModule(); + return *module; +} + +template +bool EloqStoreModule::ForEachShard(int thd_id, Fn &&fn) const +{ + for (const Slot &slot : slots_) + { + if (slot.store.load(std::memory_order_relaxed) == nullptr) + { + continue; + } + // Announce, then read the store: RemoveStore clears the slot, then + // waits for the users to leave, so it either sees this worker or this + // worker sees the slot cleared. + slot.users.fetch_add(1, std::memory_order_seq_cst); + EloqStore *store = slot.store.load(std::memory_order_seq_cst); + bool stop = false; + if (store != nullptr && + static_cast(thd_id) < store->shards_.size() && + store->shards_[thd_id] != nullptr) + { + stop = fn(store->shards_[thd_id].get()); + } + slot.users.fetch_sub(1, std::memory_order_release); + if (stop) + { + return true; + } + } + return false; +} + void EloqStoreModule::ExtThdStart(int thd_id) { - assert(static_cast(thd_id) < shards_->size()); - Shard *shard = shards_->at(thd_id).get(); - shard->BindExtThd(); + ForEachShard(thd_id, + [](Shard *shard) + { + shard->BindExtThd(); + return false; + }); } void EloqStoreModule::ExtThdEnd(int thd_id) @@ -17,16 +65,62 @@ void EloqStoreModule::ExtThdEnd(int thd_id) void EloqStoreModule::Process(int thd_id) { - // Process tasks on this shard - assert(static_cast(thd_id) < shards_->size()); - Shard *shard = shards_->at(thd_id).get(); - shard->WorkOneRound(); + ForEachShard(thd_id, + [](Shard *shard) + { + // The worker runs every store's shard in turn: bind this + // one's shard and store before running it. + shard->BindExtThd(); + shard->WorkOneRound(); + return false; + }); } bool EloqStoreModule::HasTask(int thd_id) const { - assert(static_cast(thd_id) < shards_->size()); - return !shards_->at(thd_id)->IsIdle() || shards_->at(thd_id)->NeedStop(); + return ForEachShard(thd_id, + [](Shard *shard) + { return !shard->IsIdle() || shard->NeedStop(); }); +} + +void EloqStoreModule::AddStore(EloqStore *store) +{ + std::lock_guard lk(mux_); + bool added = false; + for (Slot &slot : slots_) + { + if (slot.store.load(std::memory_order_relaxed) == nullptr) + { + slot.store.store(store, std::memory_order_seq_cst); + added = true; + break; + } + } + CHECK(added) << "More than " << kMaxStores + << " EloqStores running in one process"; + if (!registered_) + { + eloq::register_module(this); + registered_ = true; + } +} + +void EloqStoreModule::RemoveStore(EloqStore *store) +{ + std::lock_guard lk(mux_); + for (Slot &slot : slots_) + { + if (slot.store.load(std::memory_order_relaxed) != store) + { + continue; + } + slot.store.store(nullptr, std::memory_order_seq_cst); + while (slot.users.load(std::memory_order_seq_cst) != 0) + { + bthread_usleep(100); + } + return; + } } #endif diff --git a/src/storage/shard.cpp b/src/storage/shard.cpp index ed01eae8a..216b40ddb 100644 --- a/src/storage/shard.cpp +++ b/src/storage/shard.cpp @@ -124,6 +124,7 @@ void Shard::InitIoMgrAndPagePool() void Shard::WorkLoop() { shard = this; + eloq_store = store_; InitIoMgrAndPagePool(); // Get new requests from the queue, only blocked when there are no requests diff --git a/tests/eloq_store_test.cpp b/tests/eloq_store_test.cpp index 391143a4b..c1ff52253 100644 --- a/tests/eloq_store_test.cpp +++ b/tests/eloq_store_test.cpp @@ -355,6 +355,53 @@ TEST_CASE("EloqStore handles multiple stop calls", "[eloq_store]") CleanupTestDir(test_dir); } +// Several stores may run in one process (a data store service runs one per +// data shard it leads). Each must keep working whatever the others do: when the +// store was reached through a process-global pointer, stopping the store +// started last cleared it, and the next write on the other store crashed. +TEST_CASE("EloqStore runs alongside another store in the process", + "[eloq_store]") +{ + auto first_dir = CreateTestDir("_two_stores_first"); + auto second_dir = CreateTestDir("_two_stores_second"); + eloqstore::EloqStore first(CreateValidOptions(first_dir)); + eloqstore::EloqStore second(CreateValidOptions(second_dir)); + REQUIRE(first.Start("main", 0) == eloqstore::KvError::NoError); + REQUIRE(second.Start("main", 0) == eloqstore::KvError::NoError); + + auto write = [](eloqstore::EloqStore &store, std::string value) + { + std::vector entries; + entries.emplace_back( + "key", std::move(value), 1, eloqstore::WriteOp::Upsert); + eloqstore::BatchWriteRequest req; + req.SetArgs(eloqstore::TableIdent("two_stores", 0), std::move(entries)); + store.ExecSync(&req); + return req.Error(); + }; + auto read = [](eloqstore::EloqStore &store) + { + eloqstore::ReadRequest req; + req.SetArgs(eloqstore::TableIdent("two_stores", 0), "key"); + store.ExecSync(&req); + REQUIRE(req.Error() == eloqstore::KvError::NoError); + return req.value_; + }; + REQUIRE(write(first, "first") == eloqstore::KvError::NoError); + REQUIRE(write(second, "second") == eloqstore::KvError::NoError); + REQUIRE(read(first) == "first"); + REQUIRE(read(second) == "second"); + + second.Stop(); + REQUIRE(second.IsStopped()); + REQUIRE(write(first, "first again") == eloqstore::KvError::NoError); + REQUIRE(read(first) == "first again"); + + first.Stop(); + CleanupTestDir(first_dir); + CleanupTestDir(second_dir); +} + TEST_CASE("EloqStore handles requests when stopped", "[eloq_store]") { auto test_dir = CreateTestDir("_stopped_requests"); diff --git a/tests/persist.cpp b/tests/persist.cpp index f6bc12555..6faca4505 100644 --- a/tests/persist.cpp +++ b/tests/persist.cpp @@ -606,10 +606,11 @@ TEST_CASE("write task abort rolls back branch file-id high-water (cloud)", // CHECK(!IsPinned()) on the caller's legitimate pin -> SIGABRT. It must instead // surface the error to the client and leave the store usable. // -// Keep this case before the append-mode one below: that test drives its own -// EloqStore instances, which clears the global eloq_store pointer while the -// static InitStore store keeps running, so a later InitStore teardown of that -// store crashes in code reading the global (e.g. Prewarmer::Shutdown). +// This case ran before the append-mode one below while eloq_store was a +// process-global pointer: that test drives its own EloqStore instances, whose +// Stop cleared the global while the static InitStore store kept running. The +// binding is now per thread and Stop restores it, so the order no longer +// matters. TEST_CASE("failed index-page write surfaces error without aborting", "[persist][writepage_fail]") { From 01b804f623ee00e04cae2902d647f854ed11d1e8 Mon Sep 17 00:00:00 2001 From: Liang Jeff Chen Date: Tue, 20 Oct 2026 16:56:46 +0800 Subject: [PATCH 2/3] test: give the two-store case full-size coroutine stacks CreateValidOptions() sets 8 KB coroutine stacks, which suits the other cases of eloq_store_test.cpp: they validate options or start and stop stores without running a task. The two-store case writes and reads, and in a Debug build a write on such a stack overflowed it (SIGSEGV in the dynamic linker, resolving a symbol on the coroutine stack, with the fault in the stack's guard page). The case now uses the default coroutine stack size. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/eloq_store_test.cpp | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/tests/eloq_store_test.cpp b/tests/eloq_store_test.cpp index c1ff52253..6425ae4a1 100644 --- a/tests/eloq_store_test.cpp +++ b/tests/eloq_store_test.cpp @@ -364,8 +364,16 @@ TEST_CASE("EloqStore runs alongside another store in the process", { auto first_dir = CreateTestDir("_two_stores_first"); auto second_dir = CreateTestDir("_two_stores_second"); - eloqstore::EloqStore first(CreateValidOptions(first_dir)); - eloqstore::EloqStore second(CreateValidOptions(second_dir)); + // Unlike the other cases here, this one runs tasks: CreateValidOptions' + // 8 KB coroutine stacks are too small for a write in a Debug build. + auto options = [](const fs::path &dir) + { + eloqstore::KvOptions opts = CreateValidOptions(dir); + opts.coroutine_stack_size = eloqstore::KvOptions{}.coroutine_stack_size; + return opts; + }; + eloqstore::EloqStore first(options(first_dir)); + eloqstore::EloqStore second(options(second_dir)); REQUIRE(first.Start("main", 0) == eloqstore::KvError::NoError); REQUIRE(second.Start("main", 0) == eloqstore::KvError::NoError); From a7b4b3390dbbda2d72d4d767ae0a07eb57fde13a Mon Sep 17 00:00:00 2001 From: Liang Jeff Chen Date: Wed, 21 Oct 2026 01:05:22 +0800 Subject: [PATCH 3/3] fix: bind no store on the thread that calls Start or Stop Start and Stop run on whatever thread the embedding calls them from; in the data store service that is often an RPC handler's bthread, which can resume on another worker after a wait. They bound eloq_store on that thread for their whole duration, across those waits, so the binding could be set on one pthread and restored onto another. The only read of the binding on that thread was BatchWriteTask's constructor, which took the options through Options() while Start built each shard's task pools. It now takes the options explicitly, and TaskPool passes them to the task types that take them; its up-front tasks are kept in a deque, so they can be built with an argument and are never moved. - ScopedStoreBinding is gone: Start and Stop bind nothing. - Shard::Start no longer sets the caller's shard thread-local in module builds; a worker binds the shard each time it runs it. - The eloq_store comment and docs/architecture/02 say that only a thread doing a shard's work binds it. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/architecture/02-runtime-and-lifecycle.md | 9 +-- include/tasks/batch_write_task.h | 8 ++- include/tasks/task.h | 37 +++--------- include/tasks/task_manager.h | 57 +++++++++++++++---- src/eloq_store.cpp | 6 -- src/storage/shard.cpp | 3 +- src/tasks/batch_write_task.cpp | 6 +- src/tasks/task_manager.cpp | 17 +++--- 8 files changed, 79 insertions(+), 64 deletions(-) diff --git a/docs/architecture/02-runtime-and-lifecycle.md b/docs/architecture/02-runtime-and-lifecycle.md index 6a6eeb83a..2ccd5e28c 100644 --- a/docs/architecture/02-runtime-and-lifecycle.md +++ b/docs/architecture/02-runtime-and-lifecycle.md @@ -36,10 +36,11 @@ may run several stores (a data store service runs one per data shard it leads): a shard's thread binds its shard and store when its loop starts; under `ELOQ_MODULE_ENABLED` one module serves every store in the process, and worker `thd_id` runs shard `thd_id` of each store in turn, so -`EloqStoreModule::Process` binds the shard it runs each time; and -`Start` and `Stop` bind their store on the calling thread for their duration -(`ScopedStoreBinding`), for the shard and io manager setup and teardown they -run there. +`EloqStoreModule::Process` binds the shard it runs each time. Only a thread +doing a shard's work binds them. `Start` and `Stop` bind nothing: they run on +the caller's thread (in an embedding, often a bthread, which can resume on +another worker after a wait) and reach the store through `this`. `Start` adds +the store to the module, and `Stop` removes it once its shards have stopped. ## Status state machine diff --git a/include/tasks/batch_write_task.h b/include/tasks/batch_write_task.h index 4b8e8fca6..5ac7d653a 100644 --- a/include/tasks/batch_write_task.h +++ b/include/tasks/batch_write_task.h @@ -15,7 +15,13 @@ namespace eloqstore class BatchWriteTask : public WriteTask { public: - BatchWriteTask(); + /** + * @brief Builds an idle task. + * @param opts The store's options. Given rather than read through the + * thread's store binding: a shard builds its pooled tasks when it is + * constructed, on EloqStore::Start's caller, which binds no store. + */ + explicit BatchWriteTask(const KvOptions *opts); TaskType Type() const override { return TaskType::BatchWrite; diff --git a/include/tasks/task.h b/include/tasks/task.h index 24966b238..3d9a268e2 100644 --- a/include/tasks/task.h +++ b/include/tasks/task.h @@ -30,41 +30,18 @@ class EloqStore; * @brief The store whose work this thread is doing. * * A process may run several stores (a data store service runs one per data - * shard it leads), so this is bound per thread, as shard is: on a shard's own - * thread, on a brpc worker each time a module runs one of its shards - * (EloqStoreModule::Process), and on the calling thread for the duration of - * EloqStore::Start and EloqStore::Stop (ScopedStoreBinding). + * shard it leads), so this is bound per thread, as shard is, and only by a + * thread doing a shard's work: a shard's own thread, and a brpc worker each + * time a module runs one of its shards (EloqStoreModule::Process). + * + * @note Code that runs on any other thread, such as EloqStore::Start and + * EloqStore::Stop, reaches the store explicitly and must not read this: the + * caller may be a bthread, which can resume on another pthread after a wait. */ inline thread_local EloqStore *eloq_store; /** @brief The shard whose work this thread is doing (bound with eloq_store). */ inline thread_local Shard *shard; -/** - * @brief Binds eloq_store to a store on this thread for a scope, then restores - * the binding it found. - */ -class ScopedStoreBinding -{ -public: - /** - * @brief Binds eloq_store to the store. - * @param store The store whose work the scope does. - */ - explicit ScopedStoreBinding(EloqStore *store) : prev_(eloq_store) - { - eloq_store = store; - } - ScopedStoreBinding(const ScopedStoreBinding &) = delete; - ScopedStoreBinding &operator=(const ScopedStoreBinding &) = delete; - /** @brief Restores the binding found at construction. */ - ~ScopedStoreBinding() - { - eloq_store = prev_; - } - -private: - EloqStore *prev_; ///< The binding to restore. -}; KvTask *ThdTask(); AsyncIoManager *IoMgr(); const KvOptions *Options(); diff --git a/include/tasks/task_manager.h b/include/tasks/task_manager.h index e87f2d60a..9a3a08eff 100644 --- a/include/tasks/task_manager.h +++ b/include/tasks/task_manager.h @@ -2,8 +2,10 @@ #pragma once #include +#include #include #include +#include #include #include "tasks/background_write.h" @@ -52,16 +54,28 @@ class TaskManager class TaskPool { public: - TaskPool(uint32_t size, bool allow_growth = true) - : init_pool_size_(size), allow_growth_(allow_growth) + /** + * @brief Builds the pool's initial tasks. + * @param size Number of tasks built now. + * @param allow_growth Whether GetTask builds another task when none is + * free. + * @param opts The store's options, given to the task types built with + * them (see MakeTask). + */ + TaskPool(uint32_t size, bool allow_growth, const KvOptions *opts) + : opts_(opts), allow_growth_(allow_growth) { - if (size > 0) + for (uint32_t i = 0; i < size; i++) { - init_pool_ = std::make_unique(size); - for (uint32_t i = 0; i < size; i++) + if constexpr (std::is_constructible_v) { - FreeTask(&init_pool_[i]); + init_pool_.emplace_back(opts_); } + else + { + init_pool_.emplace_back(); + } + FreeTask(&init_pool_.back()); } }; @@ -80,7 +94,7 @@ class TaskManager { return nullptr; } - auto &task = ext_pool_.emplace_back(std::make_unique()); + auto &task = ext_pool_.emplace_back(MakeTask()); return task.get(); } @@ -94,9 +108,9 @@ class TaskManager template void ForEachTask(F &&visitor) { - for (uint32_t i = 0; i < init_pool_size_; ++i) + for (T &task : init_pool_) { - visitor(&init_pool_[i]); + visitor(&task); } for (auto &task : ext_pool_) { @@ -108,14 +122,33 @@ class TaskManager { free_head_ = nullptr; ext_pool_.clear(); - init_pool_.reset(); + init_pool_.clear(); } private: - std::unique_ptr init_pool_{nullptr}; + /** + * @brief Builds one task, with the store's options if its type takes + * them. + * @return The new task. + */ + std::unique_ptr MakeTask() const + { + if constexpr (std::is_constructible_v) + { + return std::make_unique(opts_); + } + else + { + return std::make_unique(); + } + } + + const KvOptions *opts_; ///< The store's options, for MakeTask. + /// The tasks built up front; a deque, so building one never moves + /// those already handed out. + std::deque init_pool_; std::vector> ext_pool_; T *free_head_{nullptr}; - uint32_t init_pool_size_{0}; bool allow_growth_{true}; }; diff --git a/src/eloq_store.cpp b/src/eloq_store.cpp index 52d37800b..d55191d83 100644 --- a/src/eloq_store.cpp +++ b/src/eloq_store.cpp @@ -658,9 +658,6 @@ KvError EloqStore::Start(std::string_view branch, PartitonGroupId partition_group_id) { LOG(INFO) << "===Start eloqstore, branch: " << branch << ", term: " << term; - // Code reached from here (shard and io manager setup) reads this store - // through the thread's binding. - ScopedStoreBinding store_binding(this); if (!ValidateOptions(options_)) { return KvError::InvalidArgs; @@ -2258,9 +2255,6 @@ void EloqStore::ExecSyncOnAnyShard(KvRequest *req) const void EloqStore::Stop() { LOG(INFO) << "EloqStore stopping."; - // Teardown reached from here reads this store through the thread's - // binding. - ScopedStoreBinding store_binding(this); while (true) { Status current = status_.load(std::memory_order_acquire); diff --git a/src/storage/shard.cpp b/src/storage/shard.cpp index 216b40ddb..417d1a6c2 100644 --- a/src/storage/shard.cpp +++ b/src/storage/shard.cpp @@ -224,7 +224,8 @@ void Shard::WorkLoop() void Shard::Start() { #ifdef ELOQ_MODULE_ENABLED - shard = this; + // No binding here: this runs on EloqStore::Start's caller, and a worker + // binds the shard each time it runs it (EloqStoreModule::Process). running_status_.store(ShardStatus::Running, std::memory_order_release); #else thd_ = std::thread([this] { WorkLoop(); }); diff --git a/src/tasks/batch_write_task.cpp b/src/tasks/batch_write_task.cpp index b19e9c0c7..35c9b6cc8 100644 --- a/src/tasks/batch_write_task.cpp +++ b/src/tasks/batch_write_task.cpp @@ -18,10 +18,10 @@ namespace eloqstore { -BatchWriteTask::BatchWriteTask() - : idx_page_builder_(Options()), data_page_builder_(Options()) +BatchWriteTask::BatchWriteTask(const KvOptions *opts) + : idx_page_builder_(opts), data_page_builder_(opts) { - overflow_ptrs_.reserve(Options()->overflow_pointers * 4); + overflow_ptrs_.reserve(opts->overflow_pointers * 4); } KvError BatchWriteTask::SeekStack(std::string_view search_key) diff --git a/src/tasks/task_manager.cpp b/src/tasks/task_manager.cpp index 96ef8caff..d1626fd4f 100644 --- a/src/tasks/task_manager.cpp +++ b/src/tasks/task_manager.cpp @@ -30,16 +30,19 @@ TaskManager::TaskManager(const KvOptions *opts) : batch_write_pool_(opts != nullptr && opts->max_write_concurrency > 0 ? opts->max_write_concurrency : batch_write_pool_size_default, - !(opts != nullptr && opts->max_write_concurrency > 0)), + !(opts != nullptr && opts->max_write_concurrency > 0), + opts), bg_write_pool_(opts != nullptr && opts->max_write_concurrency > 0 ? opts->max_write_concurrency : background_write_pool_size_default, - !(opts != nullptr && opts->max_write_concurrency > 0)), - read_pool_(read_pool_size_default), - scan_pool_(scan_pool_size_default), - list_object_pool_(list_object_pool_size_default), - list_standby_partition_pool_(list_standby_partition_pool_size_default), - reopen_pool_(reopen_pool_size_default) + !(opts != nullptr && opts->max_write_concurrency > 0), + opts), + read_pool_(read_pool_size_default, true, opts), + scan_pool_(scan_pool_size_default, true, opts), + list_object_pool_(list_object_pool_size_default, true, opts), + list_standby_partition_pool_( + list_standby_partition_pool_size_default, true, opts), + reopen_pool_(reopen_pool_size_default, true, opts) { if (opts != nullptr && opts->max_write_concurrency > 0) {