diff --git a/docs/architecture/02-runtime-and-lifecycle.md b/docs/architecture/02-runtime-and-lifecycle.md index 171dd5c7..2ccd5e28 100644 --- a/docs/architecture/02-runtime-and-lifecycle.md +++ b/docs/architecture/02-runtime-and-lifecycle.md @@ -29,11 +29,18 @@ 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. 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 @@ -66,7 +73,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 +93,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 9af0ee1c..e8a74102 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 d09bce2b..6a57ebc5 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 e21d7240..a88883ea 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 fb374009..e9b18e62 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/batch_write_task.h b/include/tasks/batch_write_task.h index 4b8e8fca..5ac7d653 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 e80f7c7c..3d9a268e 100644 --- a/include/tasks/task.h +++ b/include/tasks/task.h @@ -26,8 +26,22 @@ 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, 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; + KvTask *ThdTask(); AsyncIoManager *IoMgr(); const KvOptions *Options(); diff --git a/include/tasks/task_manager.h b/include/tasks/task_manager.h index e87f2d60..9a3a08ef 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 5f2fbe15..d55191d8 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, @@ -750,7 +748,6 @@ KvError EloqStore::Start(std::string_view branch, standby_service_.reset(); } - eloq_store = this; // Initialize if (!options_.store_path.empty()) { @@ -868,8 +865,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_)); diff --git a/src/eloqstore_module.cpp b/src/eloqstore_module.cpp index bc2713dc..5b1b7af2 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 ed01eae8..417d1a6c 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 @@ -223,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 b19e9c0c..35c9b6cc 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 96ef8caf..d1626fd4 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) { diff --git a/tests/eloq_store_test.cpp b/tests/eloq_store_test.cpp index 391143a4..6425ae4a 100644 --- a/tests/eloq_store_test.cpp +++ b/tests/eloq_store_test.cpp @@ -355,6 +355,61 @@ 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"); + // 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); + + 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 f6bc1255..6faca450 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]") {