diff --git a/CMakeLists.txt b/CMakeLists.txt index 68d076ac..755f1804 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -151,6 +151,8 @@ set(ELOQ_STORE_SOURCES src/tasks/write_buffer_aggregator.cpp src/tasks/prewarm_task.cpp src/tasks/read_task.cpp + src/tasks/attach_shard_task.cpp + src/tasks/detach_shard_task.cpp src/tasks/reopen_task.cpp src/tasks/scan_task.cpp src/tasks/task.cpp diff --git a/docs/architecture/02-runtime-and-lifecycle.md b/docs/architecture/02-runtime-and-lifecycle.md index 171dd5c7..b853b969 100644 --- a/docs/architecture/02-runtime-and-lifecycle.md +++ b/docs/architecture/02-runtime-and-lifecycle.md @@ -33,17 +33,124 @@ 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. +run on a shard thread. One store is enough: it hosts every data store shard +the process serves (next sections). ## Status state machine `EloqStore::Status`: `Stopped → Starting → Running → Stopping → Stopped`. -`Start()` CAS-transitions Stopped→Starting and rejects concurrent or repeated -starts. `SendRequest` accepts work only while `Running`. - -## Start() sequence - -`EloqStore::Start(branch, term, partition_group_id)` — +`StartInstance()` CAS-transitions Stopped→Starting and rejects concurrent or +repeated starts. `SendRequest` accepts work only while `Running`, and only +for an attached data store shard (next section). + +## Data store shards + +The embedder's unit of ownership is a *data store shard*: a set of partitions +hosted as one unit, with its own term (in EloqDB, node group g's data is data +store shard g). One store hosts several at once, each attached and detached +on its own while the others keep serving, so one process runs one store and +one budget (buffer pool, fd limit, local cache space) however many node groups +its node leads. Only cloud mode hosts more than one: data on a node's own +disk has one owner. + +- **Table.** `ds_shards_[id]` holds each shard's state (`DsShardState`: + Detached, Attaching, Attached, Expired, Detaching) and term, read without a + lock on every request. `AttachShard`, `DetachShard` and `Stop` change it + under `ds_shard_mux_`, which keeps their multi-field changes consistent when + the embedder calls them concurrently (an attach's "no other shard is + hosted" check and its claim; a late attach against `Stop`'s reset). + `MarkShardExpired` flips one slot with a compare-and-swap. The mutex is + never held across the term-file wait or the detach's wait for the worker + shards: in module builds a second caller blocking on it would block a + host worker pthread that drives a worker shard. Ids are below + `kMaxDsShards` (1024). +- **Requests name their shard** (`KvRequest::SetDsShard`). `SendRequest` + resolves a request naming none to the default shard (the one `Start` + attached, else the first attached) and admits the request while the shard + is Attached or Expired; any other state refuses it (`NotRunning`). An + Expired shard's request goes on to its worker shard, which fails it with + `ExpiredTerm`: completing it inside `SendRequest` would run its callback + inside the submit call, which the fan-out sweeps do not expect. The worker + shard checks again when it + dequeues the request (`Shard::AdmitRequest`), so a detach that started in + between still refuses it. +- **A request has no term; its task does.** The sender cannot tell when, + during a request's life, the shard's term changes, so the request does not + carry one. A request still on its way to its worker shard when the shard is + detached (a submission not yet returned, or one not yet dequeued) is + refused if the shard is still detached when the worker shard gets to it, + and runs in the new term if the shard has been attached again by then. What + holds is that a task runs entirely in one term: a detach waits for the + shard's running tasks, and an attach needs the slot Detached, so the + shard's term never changes under a running task. +- **Terms travel with the task.** When a task starts (`Shard::StartTask`) it + copies the request's shard id and reads the shard's term from the table + once (`KvTask::ds_shard_id_`, `KvTask::term_`). Every switch to a task + (`Shard::SwitchToTask`) records both on the worker shard beside the running + task (`Shard::running_ds_shard_id_`, `Shard::running_term_`, read through + `ThdDsShard()` / `ThdTerm()` like `ThdTask()`), and + `AsyncIoManager::ProcessTerm()` returns that term: no lookup by partition + anywhere. The internal requests a task queues for + its partition (compaction, file GC, local GC, TTL cleaning, the reopen + driver behind parked waiters) are stamped with the task's shard, so they + are gated and counted like external ones and run in the same term. The fd + syncs that a file eviction runs from another shard's task name a manifest + by the fd's own term, as data files always were. +- **`AttachShard(id, term)`** (cloud mode): raises the shard's term file + `CURRENT_TERM__` with a compare-and-swap, fails with + `ExpiredTerm` if the file holds a newer term, marks the slot Attached, + starts the shard's archive schedule (below) and queues the shard's prewarm. + The swap is an ordinary task: `AttachShard` sends one `AttachShardRequest` + to worker shard `id % num_threads` (bypassing the admission gate, as the + slot is still Attaching) and waits for it with a bthread-aware wait. The + `AttachShardTask` takes the shard's term from the slot like any task + (`StartTask`) and runs `CloudStoreMgr::UpsertTermFile` through that worker's + object store: read the file and its ETag, then a conditional upload + (`If-Match` on the ETag, or `If-None-Match: *` when it is missing), reading + again when a concurrent writer wins (409/412, or 404 on an `If-Match`), up + to 10 attempts. No thread blocks on HTTP. + Each partition then takes over lazily on its first open: `GetManifest` + promotes its manifest to the shard's term, and a manifest of a newer term + fails that partition's request with `ExpiredTerm` and marks the whole shard + Expired (`Shard::OnTaskFinished`), since the term file is per shard. +- **`DetachShard(id)`** marks the slot Detaching (the gate), ends the shard's + archive schedule, then sends one + `DetachShardRequest` to every worker shard and waits for all of them. On + each worker the request runs as a `DetachShardTask`: it fails the requests + queued for the shard's partitions (pending writes, parked reopen waiters, + delayed reopen drivers), blocks until no task runs on them (the task that + brings `active_tasks_by_shard_` to 0 resumes it), then `EvictDsShard` drops + their root metas and cached pages, closes their files and drops their local + manifests (cloud mode). It finds the partitions without a lookup table: + of every partition with a root meta (`RootMetaMgr`) or an open file + (`IouringMgr::PartitionFiles`) on the worker, it drops those + `KvOptions::partition_filter` gives the detaching shard. The filter must + still give them when the detach runs, so the embedder detaches a shard + before reassigning its partitions; a leader change does not reassign + them. Hosting more than one shard needs the filter; `AttachShard` refuses + a second shard without it. The cached copies of data files stay: + a later term starts a new file, so no file of an older term changes, and a + later attach reuses them. +- **Archives are scheduled per shard.** With scheduled archiving on + (`archive_interval_secs`), `ArchiveCrond` keeps one schedule per attached + shard: `AttachShard` starts it, first due one interval later, and + `DetachShard` ends it before draining, so a shard attached again at a newer + term starts afresh. One thread waits for the earliest due shard, submits + that shard's `GlobalArchiveRequest` (named with the shard's term, limited to + its partitions by `partition_filter`) with a completion callback, and sets + the shard's next due time one interval after the archive completes. The + archives of different shards run independently; an Expired or Detaching + shard's turn is put off one interval. File GC keeps + `num_retained_archives` per partition, so retention is per shard too. +- **`Stop()`** drops everything wholesale and resets the table. +- **`Start(branch, term, id)`** is `StartInstance(branch)` followed by + `AttachShard(id, term)`, for embedders that host one shard; a failed attach + stops the store again. `Term()` and `PartitionGroupId()` report the default + shard's term and id. + +## StartInstance() sequence + +`EloqStore::StartInstance(branch)` — the order is load-bearing; later steps assume earlier ones: 1. `ValidateOptions(options_)` — sanity checks plus *rewrites* (e.g. in cloud @@ -54,23 +161,26 @@ the order is load-bearing; later steps assume earlier ones: 3. `InitStoreSpace()` — create/open every `store_path` root directory, hold `root_fds_`, build `store_path_lut` (partition→path placement honoring disk capacity or explicit weights). -4. Record identity: `term_` (forced 0 in Local mode), `branch_`, - `partition_group_id_` (cloud only). +4. Record `branch_`, which every hosted data store shard shares. Terms are + per shard (previous section). 5. Compute per-shard fd budget: `fd_limit` clamped to process `RLIMIT_NOFILE` headroom minus `num_reserved_fd` (100), divided by `num_threads`. Fails with `OpenFileLimit` if insufficient. 6. Construct + `Init()` every `Shard` (allocates queues; the heavy I/O bootstrap happens later on the shard's own thread). -7. Start remote service: `CloudStorageService::Start()` then **block** on its - bootstrap (verifies bucket, CAS-upserts the `CURRENT_TERM__` - object; a stale term fails the whole `Start`). Or `StandbyService::Start()`. +7. Start remote service: `CloudStorageService::Start()` or + `StandbyService::Start()`. No term file is touched here: each + `AttachShard` raises its shard's. 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. 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. -12. Publish `Status::Running`. + archive_interval_secs > 0`; it archives each attached shard on that + shard's own schedule (see "Data store shards"). +11. Start `PrewarmService` if `prewarm_cloud_cache` and mode is Cloud; it + warms a shard's partitions when the shard is attached. +12. Publish `Status::Running`. No request is admitted until a shard is + attached. Any failure routes through `fail_start` → `CleanupRuntime(started_shards)`, so partial startups unwind in the same order as full shutdown. @@ -117,8 +227,9 @@ state: `global_registered_memories` XOR `pinned_memory_chunks`, `gc_global_mem_size_per_shard`, `pinned_tail_scratch_slots`, `segments_per_file_shift`. -- `partition_filter` — predicate telling this instance which partitions it - owns (drives prewarm/global-op scoping in multi-instance deployments). +- `partition_filter(ds_shard_id, tbl_id)` — predicate telling which + partitions belong to a hosted data store shard; scopes prewarm and the + global archive, reopen and create-branch sweeps to the shard they run for. - `LoadFromIni(path)` parses the same fields from an `.ini` file (used by benchmarks/tools; see `benchmark/*.ini`). diff --git a/docs/architecture/03-request-api.md b/docs/architecture/03-request-api.md index ece82ebf..f7ccce16 100644 --- a/docs/architecture/03-request-api.md +++ b/docs/architecture/03-request-api.md @@ -99,4 +99,19 @@ Two retry loops live *inside* the engine (`Shard::StartTask` epilogue + up to `auto_oom_retry_times`, giving other tasks a chance to release pins. `ExpiredTerm` is terminal by design: it means another process owns a newer -term and this writer must stop (doc 06). +term and this writer must stop (doc 06). It is scoped to a data store shard: +the shard whose partition met the newer term is marked Expired, every later +request for it fails with `ExpiredTerm` on its worker shard, and the other +shards the store hosts keep serving (doc 02). + +## Data store shard of a request + +Every request works for one data store shard (`KvRequest::SetDsShard`), the +embedder's unit of ownership with its own term (doc 02). The shard decides +whether the request is admitted and which term its partition's files are +named with. A request naming no shard works for the store's default one: the +shard `Start` attached, so embedders hosting one shard never set it. The +global sweeps (`GlobalArchiveRequest`, `GlobalReopenRequest`, +`GlobalCreateBranchRequest`, `DropTableRequest`) hand their shard to the +per-partition requests they fan out, and keep only the partitions +`partition_filter` gives that shard. diff --git a/docs/architecture/06-persistence-and-recovery.md b/docs/architecture/06-persistence-and-recovery.md index a95e1b0a..79d2403a 100644 --- a/docs/architecture/06-persistence-and-recovery.md +++ b/docs/architecture/06-persistence-and-recovery.md @@ -96,10 +96,12 @@ the manifest and directory FDs. ## Terms -The embedding system passes a `term` to `Start()` (0 in Local mode). Files and -manifests are stamped with the writer's term. In cloud mode, -`CloudStorageService` bootstrap CAS-updates `CURRENT_TERM__` -(ETag-guarded); a writer that discovers a newer published term gets +The embedding system passes a `term` with each data store shard it attaches +(`AttachShard`; `Start()` attaches one), 0 in Local mode. Files and +manifests are stamped with the writer's term. In cloud mode the attach +CAS-updates that shard's `CURRENT_TERM__` (ETag-guarded), from +a task on a worker shard (`AttachShardTask` → `CloudStoreMgr::UpsertTermFile`, +doc 02); a writer that discovers a newer published term gets `KvError::ExpiredTerm` and must stop — this fences zombie writers after failover. Readers/GC treat *newer-term* files conservatively (see BranchGuard below). `ManifestTermFromFilename` etc. parse terms back out of names. diff --git a/docs/architecture/07-io-stack.md b/docs/architecture/07-io-stack.md index 89bfb23a..220862fe 100644 --- a/docs/architecture/07-io-stack.md +++ b/docs/architecture/07-io-stack.md @@ -89,7 +89,8 @@ Local disk = bounded LRU cache of cloud objects; object store = durable copy. - **Concurrency limits**: cloud ops take a slot (`AcquireCloudSlot`, `max_cloud_concurrency`) and a pooled `DirectIoBuffer` (`AcquireCloudBuffer`); both park the coroutine when exhausted. -- **Terms**: `ReadTermFile`/bootstrap CAS on `CURRENT_TERM_*` (doc 06); +- **Terms**: `ReadTermFile`, and `UpsertTermFile`, the attach-time CAS on + `CURRENT_TERM_*` run from a task (docs 02, 06); `process_term_`/`partition_group_id_` stamp uploads and validate manifests. ## Cloud transport @@ -101,8 +102,7 @@ Three layers, crossing the thread boundary exactly once each way: `KvTask` and shard. 2. **`CloudStorageService`** (per store) — `cloud_request_threads` worker threads, each running a curl multi loop; jobs from a given shard always go - to the same worker (curl handles are not thread-safe). Also runs the - bootstrap term-file CAS at Start. + to the same worker (curl handles are not thread-safe). 3. **`AsyncHttpManager`** (per shard, driven by the service worker) — builds signed HTTP requests via the pluggable `CloudBackend` (`include/storage/cloud_backend.h`; AWS S3 / GCS variants selected by diff --git a/docs/architecture/08-data-lifecycle.md b/docs/architecture/08-data-lifecycle.md index 7b4606ec..94b336fa 100644 --- a/docs/architecture/08-data-lifecycle.md +++ b/docs/architecture/08-data-lifecycle.md @@ -86,8 +86,12 @@ requests in each `PendingWriteQueue`. - **CreateArchive / CreateBranch / DeleteBranch** — manifest-level operations (doc 06) executed under the write lock for the partition. -`ArchiveCrond` (its own thread) periodically submits archive requests for all -owned partitions at `archive_interval_secs`, bounded by `max_archive_tasks`. +`ArchiveCrond` (its own thread) archives each attached data store shard on +that shard's own schedule: every `archive_interval_secs`, measured from the +end of its previous archive, starting one interval after the attach and +ending at its detach (doc 02, "Data store shards"). Each archive is one +`GlobalArchiveRequest` for the shard's partitions, bounded by +`max_archive_tasks`. ## File garbage collection (`include/file_gc.h`, `src/file_gc.cpp`) diff --git a/docs/architecture/10-bindings-and-embedding.md b/docs/architecture/10-bindings-and-embedding.md index 9af0ee1c..09528ab6 100644 --- a/docs/architecture/10-bindings-and-embedding.md +++ b/docs/architecture/10-bindings-and-embedding.md @@ -65,6 +65,17 @@ only when the embedding runtime (EloqKV's tx service, with bthread) defines - Shutdown is cooperative: shards are flagged `Stopping` and the host's workers run rounds until each reports `Stopped` (see `EloqStore::CleanupRuntime`). +- A host that leads several node groups hosts their data store shards on the + one store: `StartInstance` once, then `AttachShard`/`DetachShard` as + leadership comes and goes, each request stamped with its shard + (`KvRequest::SetDsShard`) and `partition_filter(ds_shard_id, tbl_id)` + answering from the host's bucket map (doc 02); the filter is required to + host more than one shard, since a detach drops the state of the + partitions it gives the detaching shard, and `DetachShard` must run before + the bucket map moves those partitions elsewhere. `AttachShard` and + `DetachShard` wait with bthread-aware waits, so an RPC handler's bthread + may call them; a module hook (`Process`) must not, since `DetachShard` + waits for every worker shard, its own included. When changing `Shard::WorkLoop`, mirror the change in `WorkOneRound` — they are the same scheduler in two drive modes. diff --git a/include/async_io_manager.h b/include/async_io_manager.h index 19a00292..5aa51c4a 100644 --- a/include/async_io_manager.h +++ b/include/async_io_manager.h @@ -512,9 +512,16 @@ class AsyncIoManager (void) pre_max_segment_file_id; } - virtual uint64_t ProcessTerm() const + /** + * @brief The term new files and manifests are named with and manifests + * are fenced against: the term of the running task's data store shard, + * read from the worker shard (ThdTerm), which records it at every switch + * to a task. 0 in local mode and outside a task. + * @return The term. + */ + uint64_t ProcessTerm() const { - return 0; + return ThdTerm(); } virtual std::string_view GetActiveBranch() const @@ -647,13 +654,6 @@ class IouringMgr : public AsyncIoManager FileId pre_max_file_id, FileId pre_max_segment_file_id) override; - // Process term management for term-aware file naming. - // Local mode always returns 0. - uint64_t ProcessTerm() const override - { - return 0; - } - void SetActiveBranch(std::string_view branch) override { active_branch_ = std::string(branch); @@ -670,6 +670,23 @@ class IouringMgr : public AsyncIoManager KvError DeleteFiles(const std::vector &file_paths); KvError CloseFiles(const TableIdent &tbl_id, const std::span file_ids); + /** + * @brief Closes every file this worker shard has open for a partition, + * flushing dirty ones. Run from a task: the closes are io_uring operations + * the task waits on. + * + * @param tbl_id The partition. + * @return NoError, or the first close error. + */ + KvError CloseTableFiles(const TableIdent &tbl_id); + /** + * @brief Collects the partitions that have a file open on this worker + * shard, so that a data store shard's detach can pick out the ones whose + * files it closes. + * @param[in,out] partitions Receives the partitions; what it already + * holds is kept, so several sources can collect into one set. + */ + void CollectPartitions(std::unordered_set &partitions) const; /** * @brief Get the number of currently open file descriptors. @@ -1286,22 +1303,6 @@ class CloudStoreMgr final : public IouringMgr return !prewarmers_.empty(); } - void SetProcessTerm(uint64_t term) - { - process_term_ = term; - } - void SetPartitionGroupId(PartitonGroupId partition_group_id) - { - partition_group_id_ = partition_group_id; - } - PartitonGroupId PartitionGroupId() const - { - return partition_group_id_; - } - uint64_t ProcessTerm() const override - { - return process_term_; - } void OnFileRangeWritePrepared(const TableIdent &tbl_id, TypedFileId file_id, std::string_view branch_name, @@ -1336,11 +1337,37 @@ class CloudStoreMgr final : public IouringMgr uint64_t offset = 0); KvError CloseFile(LruFD::Ref fd) override; - // Read partition-group CURRENT_TERM file from cloud, returns - // {term_value, etag, error}. If file doesn't exist (404), returns - // {0, "", NotFound}. + /** + * @brief Reads a data store shard's term file, CURRENT_TERM__, + * from the cloud. Run from a task. + * + * @param branch_name The branch. + * @param ds_shard_id The data store shard. + * @return {term, etag, error}; {0, "", NotFound} when the file does not + * exist. + */ std::tuple ReadTermFile( - std::string_view branch_name); + std::string_view branch_name, DsShardId ds_shard_id); + /** + * @brief Raises a data store shard's term file, CURRENT_TERM__, + * to a term with a compare-and-swap, or learns that a newer term holds + * it. Run from a task. + * + * Reads the file, then uploads the term only if the file is still the + * version read: with If-Match on its etag, or with If-None-Match "*" when + * it did not exist. A concurrent writer makes the upload fail, and the + * swap reads the file again, up to 10 attempts in all. + * + * @param branch_name The branch. + * @param ds_shard_id The data store shard. + * @param term The term to write. + * @return NoError, also when the file already holds @p term; ExpiredTerm + * if it holds a newer term; CloudErr when every attempt lost a race; else + * the read's or the upload's error. + */ + KvError UpsertTermFile(std::string_view branch_name, + DsShardId ds_shard_id, + uint64_t term); private: void WaitForCloudTasksToDrain(); @@ -1549,12 +1576,6 @@ class CloudStoreMgr final : public IouringMgr ObjectStore obj_store_; CloudStorageService *cloud_service_{nullptr}; - // Expected process term for this shard in cloud mode. - // 0 means unspecified/legacy; in that case term validation in GetManifest - // will be skipped and the latest manifest term will be used. - uint64_t process_term_{0}; - PartitonGroupId partition_group_id_{0}; - size_t inflight_cloud_slots_{0}; WaitingZone cloud_slot_waiting_; size_t inflight_cloud_buffers_{0}; @@ -1569,14 +1590,6 @@ class StandbyStoreMgr final : public IouringMgr public: StandbyStoreMgr(const KvOptions *opts, uint32_t fd_limit); void Stop() override; - void SetProcessTerm(uint64_t term) - { - process_term_ = term; - } - uint64_t ProcessTerm() const override - { - return process_term_; - } std::pair GetManifest( const TableIdent &tbl_id) override; @@ -1590,7 +1603,6 @@ class StandbyStoreMgr final : public IouringMgr std::string_view filename) const; int RunRsync(const std::string &remote, const std::string &dst); std::atomic inflight_standby_tasks_{0}; - uint64_t process_term_{0}; std::string remote_addr_; }; diff --git a/include/cloud_storage_service.h b/include/cloud_storage_service.h index 4c105fa7..88362316 100644 --- a/include/cloud_storage_service.h +++ b/include/cloud_storage_service.h @@ -7,11 +7,6 @@ #include #include -#ifdef ELOQ_MODULE_ENABLED -#include -#include -#endif - #include "concurrentqueue/blockingconcurrentqueue.h" #include "storage/object_store.h" @@ -33,47 +28,6 @@ class CloudStorageService void Submit(ObjectStore *store, ObjectStore::Task *task); void NotifyTaskFinished(ObjectStore::Task *task); - struct BootstrapTermSyncState - { - KvError err_{KvError::Busy}; -#ifdef ELOQ_MODULE_ENABLED - bthread::Mutex mutex_; - bthread::ConditionVariable cv_; - bool done_{false}; - - void Wait() - { - std::unique_lock lk(mutex_); - while (!done_) - { - cv_.wait(lk); - } - } - - void SetDone(KvError err) - { - std::lock_guard lk(mutex_); - err_ = err; - done_ = true; - cv_.notify_one(); - } -#else - std::atomic done_{false}; - - void Wait() - { - done_.wait(false, std::memory_order_acquire); - } - - void SetDone(KvError err) - { - err_ = err; - done_.store(true, std::memory_order_release); - done_.notify_one(); - } -#endif - }; - bool HasPendingJobs() const { return pending_jobs_.load(std::memory_order_acquire) > 0; @@ -117,7 +71,6 @@ class CloudStorageService std::atomic stopping_{true}; std::atomic accepting_jobs_{false}; std::atomic pending_jobs_{0}; - BootstrapTermSyncState bootstrap_state_; friend class EloqStore; }; diff --git a/include/eloq_store.h b/include/eloq_store.h index d09bce2b..b7c62393 100644 --- a/include/eloq_store.h +++ b/include/eloq_store.h @@ -5,12 +5,14 @@ #include #endif +#include #include #include #include #include #include #include +#include #include #include "error.h" @@ -57,7 +59,9 @@ enum class RequestType : uint8_t GlobalListArchiveTags, CreateBranch, DeleteBranch, - GlobalCreateBranch + GlobalCreateBranch, + AttachShard, + DetachShard }; inline const char *RequestTypeToString(RequestType type) @@ -106,6 +110,10 @@ inline const char *RequestTypeToString(RequestType type) return "delete_branch"; case RequestType::GlobalCreateBranch: return "global_create_branch"; + case RequestType::AttachShard: + return "attach_shard"; + case RequestType::DetachShard: + return "detach_shard"; default: return "unknown"; } @@ -141,6 +149,23 @@ class KvRequest void SetTableId(TableIdent tbl_id); const TableIdent &TableId() const; uint64_t UserData() const; + /** + * @brief Names the data store shard this request works for. + * + * The store runs the request's partition in that shard's term and admits + * the request only while the shard is attached (EloqStore::AttachShard). + * A request left unset works for the shard EloqStore::Start attached. + * @param ds_shard_id The shard's id. + */ + void SetDsShard(DsShardId ds_shard_id) + { + ds_shard_id_ = ds_shard_id; + } + /** @return The data store shard the request works for, or kUnsetDsShard. */ + DsShardId DsShard() const + { + return ds_shard_id_; + } /** * @brief Test if this request is done. @@ -163,6 +188,7 @@ class KvRequest void SetDone(KvError err); TableIdent tbl_id_; + DsShardId ds_shard_id_{kUnsetDsShard}; uint64_t user_data_{0}; std::function callback_{nullptr}; uint8_t reopen_retry_remaining_{0}; @@ -877,6 +903,33 @@ class GlobalCreateBranchRequest : public KvRequest friend class EloqStore; }; +/** + * @brief Internal: tells one worker shard to raise a data store shard's term + * file to the shard's attach term (EloqStore::AttachShard sends one, in cloud + * mode only). + */ +class AttachShardRequest : public KvRequest +{ +public: + RequestType Type() const override + { + return RequestType::AttachShard; + } +}; + +/** + * @brief Internal: tells one worker shard to let go of a data store shard's + * partitions (EloqStore::DetachShard sends one to every worker shard). + */ +class DetachShardRequest : public KvRequest +{ +public: + RequestType Type() const override + { + return RequestType::DetachShard; + } +}; + class ArchiveCrond; class ObjectStore; class EloqStoreModule; @@ -895,13 +948,139 @@ class EloqStore Stopped = 3, }; + /** @brief What a data store shard is to this store. */ + enum class DsShardState : uint8_t + { + Detached = 0, ///< Not hosted; its requests are refused (NotRunning). + Attaching, ///< AttachShard is checking its term; requests refused. + Attached, ///< Hosted: its requests run in its term. + Expired, ///< A newer term exists; requests fail (ExpiredTerm). + Detaching, ///< DetachShard is draining it; requests refused. + }; + + /** @brief The highest data store shard id plus one (ids index a table). */ + static constexpr size_t kMaxDsShards = 1024; + EloqStore(const KvOptions &opts); EloqStore(const EloqStore &) = delete; EloqStore(EloqStore &&) = delete; ~EloqStore(); + /** + * @brief Starts the store and attaches one data store shard: StartInstance + * followed by AttachShard. Kept for embedders that host one shard. + * + * @param branch The active branch. + * @param term The shard's term (0 in local mode). + * @param ds_shard_id The shard's id. Requests that name no shard + * (KvRequest::SetDsShard) work for this one. + * @return NoError, or the error of either step; after a failed attach the + * store is stopped again. + */ KvError Start(std::string_view branch = MainBranchName, uint64_t term = 0, - PartitonGroupId partition_group_id = 0); + DsShardId ds_shard_id = 0); + /** + * @brief Starts the store with no data store shard attached: opens the + * store paths, builds the worker shards and starts the services. Requests + * are admitted once a shard is attached. + * + * @param branch The active branch, which every hosted shard shares. + * @return NoError, or why the store could not start. + */ + KvError StartInstance(std::string_view branch = MainBranchName); + /** + * @brief Hosts a data store shard: its partitions are served in @p term + * from now on. In cloud mode the shard's term file, + * CURRENT_TERM__, is first raised to @p term with a + * compare-and-swap, and the attach fails with ExpiredTerm if the file + * holds a newer term. Each partition then takes over lazily, on its first + * open: its manifest is promoted to @p term, and a manifest of a newer + * term makes that partition's requests fail with ExpiredTerm and marks the + * shard Expired. + * + * Only cloud mode hosts more than one shard at a time: data on a node's + * own disk has one owner. Hosting more than one also needs + * KvOptions::partition_filter, which tells a detach which partitions the + * other shards keep. + * + * When scheduled archiving is on (KvOptions::archive_interval_secs), the + * shard is archived on its own schedule from the attach until its detach, + * every interval, the first time one interval after the attach. + * + * @note In cloud mode this waits for the term file's compare-and-swap, + * which a task on one worker shard runs (AttachShardRequest); in module + * builds the wait is a bthread condition variable. Call it from a + * bthread, such as an RPC handler's, or from a pthread that is not a brpc + * worker. Never call it from a brpc worker outside a bthread, such as a + * module hook (Process) or other work the embedder runs on the workers' + * own loop: the wait would block that worker's pthread, and with it the + * worker shard the pthread drives, which may be the one running the swap. + * + * @param ds_shard_id The shard's id, below kMaxDsShards. + * @param term The shard's term; ignored (0) in local mode. + * @return NoError; InvalidArgs if the id is out of range, the shard is not + * detached, or a second shard is attached outside cloud mode or without a + * partition filter; NotRunning if the store is not running; ExpiredTerm + * as described. + */ + KvError AttachShard(DsShardId ds_shard_id, uint64_t term); + /** + * @brief Stops hosting a data store shard. New requests for it are refused + * at once; the call returns after every worker shard has finished the + * tasks running on its partitions, failed the requests queued for them, + * and dropped the in-memory state, open files and local manifests of its + * partitions, those KvOptions::partition_filter gives it. The filter must + * still give the shard those partitions when this runs: detach a shard + * before reassigning them. The cached copies of sealed data files stay, + * since those never change. The shard's archive schedule ends first: no + * new scheduled archive of it starts. + * + * A request still on its way to a worker shard when the detach runs, one + * whose submission has not returned or that its worker shard has not + * dequeued yet, is refused if the shard is still detached when the worker + * shard gets to it, and runs in the new term if the shard has been + * attached again by then. Either way each task runs entirely in one term. + * + * Waits on the worker shards' completions (a bthread-aware wait in module + * builds), so it may be called from an RPC handler's bthread. + * + * @param ds_shard_id The shard's id. + * @return NoError, also when the shard is not attached; InvalidArgs if the + * id is out of range or the shard is being attached or detached by another + * caller. + */ + KvError DetachShard(DsShardId ds_shard_id); + /** + * @param ds_shard_id The shard's id. + * @return The shard's state; Detached for an id out of range. + */ + DsShardState ShardState(DsShardId ds_shard_id) const + { + return ds_shard_id < kMaxDsShards ? ds_shards_[ds_shard_id].state_.load( + std::memory_order_acquire) + : DsShardState::Detached; + } + /** + * @param ds_shard_id The shard's id. + * @return The term the shard was attached with; 0 if not attached or out + * of range. + */ + uint64_t ShardTerm(DsShardId ds_shard_id) const + { + return ds_shard_id < kMaxDsShards ? ds_shards_[ds_shard_id].term_.load( + std::memory_order_acquire) + : 0; + } + /** @return The ids of the shards that are attached or expired. */ + std::vector HostedShards() const; + /** + * @return The shard that requests naming none work for: the one Start + * attached, or the first one attached; kUnsetDsShard if none. + */ + DsShardId DefaultDsShard() const + { + return default_ds_shard_.load(std::memory_order_acquire); + } void Stop(); bool IsStopped() const; bool Inited() const; @@ -947,13 +1126,15 @@ class EloqStore size_t DataCacheMisses() const; void ResetDataCacheCounters(); + /** @return The default data store shard's term (see DefaultDsShard). */ uint64_t Term() const { - return term_; + return ShardTerm(DefaultDsShard()); } + /** @return The default data store shard's id (see DefaultDsShard). */ PartitonGroupId PartitionGroupId() const { - return partition_group_id_; + return DefaultDsShard(); } KvError UpdateStandbyMasterStorePaths(std::vector paths, @@ -1015,7 +1196,20 @@ class EloqStore } private: + /** @brief One data store shard's slot in the table indexed by id. */ + struct DsShardSlot + { + std::atomic state_{DsShardState::Detached}; + std::atomic term_{0}; + }; + bool SendRequest(KvRequest *req); + /** + * @brief Marks a data store shard Expired: a partition of it met a newer + * term. Called by a worker shard; a no-op unless the shard is Attached. + * @param ds_shard_id The shard's id. + */ + void MarkShardExpired(DsShardId ds_shard_id); // Submit @p req to an arbitrary shard and block until it completes. Used by // the global cloud-listing handlers, which are not tied to a partition, so // any shard can serve them. If the store is stopping the shard refuses the @@ -1043,8 +1237,42 @@ class EloqStore std::vector> metrics_meters_; #endif std::atomic status_{Status::Stopped}; - uint64_t term_{0}; - PartitonGroupId partition_group_id_{0}; + /** + * The data store shard table: one slot per shard id, holding the shard's + * state and term. Read without a lock on every request (SendRequest, + * Shard::AdmitRequest, Shard::StartTask). Changed by AttachShard, + * DetachShard and Stop under ds_shard_mux_, and by MarkShardExpired with + * a compare-and-swap on one slot's state. + */ + std::array ds_shards_; + /** + * Serializes the changes AttachShard, DetachShard and Stop make to the + * shard table (ds_shards_) and to default_ds_shard_. + * + * The embedder may call these concurrently from different threads or + * bthreads: two RPC handlers attaching or detaching different data store + * shards, or Stop while an attach is in flight. Each change spans more + * than one atomic, so the atomics alone do not keep it consistent: + * - AttachShard checks that no other slot is hosted (outside cloud mode, + * or without a partition filter) and then claims its own slot. Two + * concurrent attaches could otherwise both pass the check. + * - AttachShard and DetachShard check that the store is running and then + * publish a slot's state together with default_ds_shard_, while Stop + * resets every slot. An attach finishing late must not overwrite that + * reset. + * + * Never held across the long waits: AttachShard's wait for a worker + * shard to swap the term file, a cloud round trip, and DetachShard's wait + * for the worker shards to finish the shard's tasks. In module builds a + * caller is a bthread on one of the host's worker pthreads, and those + * pthreads also drive the worker shards. The waits park only the waiting + * bthread, but another caller that blocks on this std::mutex blocks its + * whole pthread, and so stops the worker shard that pthread drives. If the + * holder were a DetachShard waiting for that very worker shard, neither + * could proceed. + */ + std::mutex ds_shard_mux_; + std::atomic default_ds_shard_{kUnsetDsShard}; std::string branch_{MainBranchName}; std::unique_ptr archive_crond_{nullptr}; std::unique_ptr prewarm_service_{nullptr}; diff --git a/include/kv_options.h b/include/kv_options.h index e71d2863..ccc9102c 100644 --- a/include/kv_options.h +++ b/include/kv_options.h @@ -429,15 +429,20 @@ struct KvOptions uint64_t auto_reopen_pending_time_us = 10'000'000; /** - * @brief Filter function to determine which partitions belong to this - * instance. - * The filter returning true means that this table partition belongs to the - * current instance and should be included in operations like prewarming, - * snapshotting, etc. - * If not set (empty), all partitions are considered to belong to this - * instance. - */ - std::function partition_filter; + * @brief Tells which partitions belong to a data store shard this store + * hosts. + * + * The operations that sweep the storage (prewarm, global archive, reopen + * and create-branch) list every partition and keep those for which the + * filter returns true for the shard they run for. DetachShard drops the + * state of the partitions the filter gives the detaching shard, so it + * must still give them when the detach runs, and hosting more than one + * shard requires the filter: AttachShard refuses a second shard without + * it. The filter is called from the store's threads with the + * shard id a request was sent with (KvRequest::SetDsShard). If not set, + * every partition is taken to belong to every shard. + */ + std::function partition_filter; }; } // namespace eloqstore diff --git a/include/storage/object_store.h b/include/storage/object_store.h index 2ab5d7cb..233cae3b 100644 --- a/include/storage/object_store.h +++ b/include/storage/object_store.h @@ -51,8 +51,6 @@ class ObjectStore ~ObjectStore(); KvError EnsureBucketExists(); - KvError BootstrapUpsertTermFile(std::string_view remote_path, - uint64_t process_term); bool ParseListObjectsResponse( std::string_view payload, @@ -257,8 +255,6 @@ class AsyncHttpManager void Shutdown(); KvError EnsureBucketExists(); - KvError BootstrapUpsertTermFile(std::string_view remote_path, - uint64_t process_term); bool IsIdle() const { diff --git a/include/storage/root_meta.h b/include/storage/root_meta.h index f9b991d4..cb3084cf 100644 --- a/include/storage/root_meta.h +++ b/include/storage/root_meta.h @@ -9,6 +9,7 @@ #include #include #include +#include #include "absl/container/flat_hash_set.h" #include "common.h" @@ -193,6 +194,18 @@ class RootMetaMgr // (it can run inside Erase() and would re-enter the erase logic). Entry *FindNoErase(const TableIdent &tbl_id); void Erase(const TableIdent &tbl_id); + /** + * @brief Lists the partitions that have a cached root meta. + * + * This manager holds the cached RootMeta of every open partition, one + * Entry each. This inserts the TableIdent of every entry, including those + * awaiting a deferred erase, so that a data store shard's detach can pick + * out the partitions whose root metas it drops. + * + * @param[in,out] partitions Receives the partitions; what it already + * holds is kept, so several sources can collect into one set. + */ + void CollectPartitions(std::unordered_set &partitions) const; void Pin(Entry *entry); void Unpin(Entry *entry); diff --git a/include/storage/shard.h b/include/storage/shard.h index fb374009..d46658a6 100644 --- a/include/storage/shard.h +++ b/include/storage/shard.h @@ -91,6 +91,22 @@ class Shard // Drives CurResumeElapsedUs() for cooperative time-budgeted yielding. uint64_t cur_resume_start_us_{}; KvTask *running_{}; + /** + * The data store shard the running task works for, copied from the task + * at every switch to it (SwitchToTask) and cleared with running_ + * (LeaveTask), so that running_, this and running_term_ always describe + * the same task. The code a task runs reads it through ThdDsShard() + * instead of looking anything up: the request named the shard. + * kUnsetDsShard between tasks. + */ + DsShardId running_ds_shard_id_{kUnsetDsShard}; + /** + * The term of the running task's data store shard, kept like + * running_ds_shard_id_; the task took it once when it started. Read + * through ThdTerm(), down to the file naming in the IO manager + * (AsyncIoManager::ProcessTerm). 0 in local mode and between tasks. + */ + uint64_t running_term_{0}; CircularQueue ready_tasks_; CircularQueue low_priority_ready_tasks_; @@ -118,6 +134,41 @@ class Shard void TryStartPendingWrite(const TableIdent &tbl_id); void TryDispatchPendingWrites(); + /** + * @brief Decides whether a request may run: its data store shard must be + * attached. Completes the request with NotRunning (shard not attached) or + * ExpiredTerm (shard expired) otherwise. A request naming no shard, such + * as the internal listing requests, is always admitted. + * + * @param req The request, dequeued or taken from a pending queue. + * @return Whether the request may run. + */ + bool AdmitRequest(KvRequest *req); + /** + * @brief Fails every request queued for a data store shard that has not + * started: pending writes, parked reopen waiters and delayed reopen + * drivers. Each request names its shard. Running tasks are left to finish. + * @param ds_shard_id The shard. + */ + void FailPendingRequestsOf(DsShardId ds_shard_id); + /** + * @brief Drops what this worker shard holds for a data store shard once + * no task runs for it: the root metas (with their index pages and + * mappers) and open files of its partitions, their local manifests in + * cloud mode, and their pending queues. Its partitions are those with a + * root meta or an open file here that KvOptions::partition_filter gives + * it. Runs in the detach task. + * @param ds_shard_id The shard. + * @return NoError, or the first error met; the state is dropped either way. + */ + KvError EvictDsShard(DsShardId ds_shard_id); + /** + * @param ds_shard_id The shard. + * @return How many tasks run on the shard's partitions (the detach task + * itself not counted). + */ + uint32_t ActiveTasksOf(DsShardId ds_shard_id) const; + #ifdef ELOQ_MODULE_ENABLED void WorkOneRound(); bool IsIdle() const @@ -148,17 +199,48 @@ class Shard uint64_t DurationMicroseconds(uint64_t start_us); + /** + * @brief Makes a task the running one: records it with its data store + * shard and term (running_, running_ds_shard_id_, running_term_) and + * marks the resume start for CurResumeElapsedUs. Every switch to a task + * goes through here, so the code the task runs reads the shard and term + * from the worker shard (ThdDsShard, ThdTerm) instead of looking them up. + * @param task The task about to run. + */ + void SwitchToTask(KvTask *task) + { + running_ = task; + running_ds_shard_id_ = task->ds_shard_id_; + running_term_ = task->term_; + // Mark the resume start so a cooperative background loop measures this + // segment from now, not from the previously resumed task's timestamp + // (see CurResumeElapsedUs / MaybeYield). + cur_resume_start_us_ = ReadTimeMicroseconds(); + } + /** @brief No task runs: clears what SwitchToTask recorded. */ + void LeaveTask() + { + running_ = nullptr; + running_ds_shard_id_ = kUnsetDsShard; + running_term_ = 0; + } + template void StartTask(KvTask *task, KvRequest *req, F lbd) { task->req_ = req; task->result_err_ = KvError::NoError; task->status_ = TaskStatus::Ongoing; - running_ = task; - // Mark the resume start so a cooperative background loop measures this - // first segment from now, not from the previously resumed task's - // timestamp (see CurResumeElapsedUs / MaybeYield). - cur_resume_start_us_ = ReadTimeMicroseconds(); + // The request named its data store shard; the task takes the shard's + // term now, once, and carries both from here on. + task->ds_shard_id_ = req->DsShard(); + task->term_ = store_->ShardTerm(task->ds_shard_id_); + if (task->ds_shard_id_ != kUnsetDsShard && + task->Type() != TaskType::DetachShard) + { + ++active_tasks_by_shard_[task->ds_shard_id_]; + } + SwitchToTask(task); task->coro_ = boost::context::callcc( std::allocator_arg, stack_allocator_, @@ -188,7 +270,7 @@ class Shard task->status_ = TaskStatus::Finished; return std::move(shard->main_); }); - running_ = nullptr; + LeaveTask(); if (task->status_ == TaskStatus::Finished) { OnTaskFinished(task); @@ -213,6 +295,12 @@ class Shard metrics::TimePoint finished_request_start_{}; #endif + /** + * The write requests of one partition waiting for their turn. Writes to a + * partition run one at a time: TryStartPendingWrite starts the front + * request only while no write task of the partition is in flight, and + * OnTaskFinished starts the next one when that task ends. + */ class PendingWriteQueue { public: @@ -227,6 +315,15 @@ class Shard FileGcRequest file_gc_req_; LocalGcRequest local_gc_req_; CleanExpiredRequest expire_req_; + /** + * True while a write task of this partition is in flight: set by + * TryStartPendingWrite when it hands the front request to ProcessReq + * (reset at once if the task manager could not start a task) and + * cleared by OnTaskFinished. The running request is no longer in the + * queue, so an empty queue does not mean the partition is idle; this + * flag does. The shard detach relies on it too (FailPendingRequestsOf, + * EvictDsShard). + */ bool running_{false}; private: @@ -277,6 +374,18 @@ class Shard }; std::vector delayed_requests_; + /** + * Tasks running per data store shard (KvTask::ds_shard_id_), kept by + * StartTask and OnTaskFinished. A detach task waits for its shard's count + * to reach 0. + */ + std::unordered_map active_tasks_by_shard_; + /** + * Detach tasks blocked until their shard's running tasks finish; the task + * that brings a count to 0 resumes the matching one. + */ + std::unordered_map waiting_detaches_; + #ifdef ELOQ_MODULE_ENABLED std::atomic req_queue_size_{0}; #endif @@ -294,5 +403,6 @@ class Shard friend class EloqStoreModule; friend class EloqStore; friend class KvRequest; + friend class DetachShardTask; }; } // namespace eloqstore diff --git a/include/tasks/archive_crond.h b/include/tasks/archive_crond.h index e6343bb9..857df23f 100644 --- a/include/tasks/archive_crond.h +++ b/include/tasks/archive_crond.h @@ -1,28 +1,132 @@ #pragma once +#include +#include #include +#include +#include +#include +#include #include +#include #include "eloq_store.h" namespace eloqstore { +/** + * @brief Archives each attached data store shard on its own schedule, every + * KvOptions::archive_interval_secs. + * + * A shard's schedule belongs to its attachment: it starts when the shard is + * attached (AddShard, from EloqStore::AttachShard) and ends when the shard is + * detached (RemoveShard, from EloqStore::DetachShard), so a shard attached + * again at a newer term starts afresh. One thread serves every shard: it + * waits for the earliest due shard, submits that shard's GlobalArchiveRequest + * with a completion callback, and sets the shard's next due time one interval + * after the archive completes. The archives of different shards therefore run + * independently, and a slow one delays no other shard. + */ class ArchiveCrond { public: - ArchiveCrond(EloqStore *store); + /** @param store The store whose data store shards are archived. */ + explicit ArchiveCrond(EloqStore *store); + + /** + * @brief Starts the thread. The schedules a previous run of the store + * left are forgotten: the shards attached from now on bring their own. + */ void Start(); + /** + * @brief Stops the thread. An archive in flight is not waited for: it + * completes while the store stops its worker shards, and its completion + * only updates this object, which the store keeps. + * @note In module builds it yields while the thread finishes, never + * blocking a brpc worker: called from a bthread on the worker that drives + * the shard the thread waits on to list partitions, a blocking join would + * never return. + */ void Stop(); - bool StopRequested(); + /** + * @brief Starts a data store shard's schedule: its first archive is due + * one interval from now. + * @param ds_shard_id The shard, just attached. + */ + void AddShard(DsShardId ds_shard_id); + /** + * @brief Ends a data store shard's schedule: no new archive of it starts. + * An archive of it in flight finishes on its own, and its completion does + * not schedule another. + * @param ds_shard_id The shard, being detached. + */ + void RemoveShard(DsShardId ds_shard_id); private: + using Clock = std::chrono::steady_clock; +#ifdef ELOQ_MODULE_ENABLED + using Lock = std::unique_lock; +#else + using Lock = std::unique_lock; +#endif + + /** @brief One data store shard's archive schedule. */ + struct ShardSchedule + { + Clock::time_point next_due_{}; ///< When its next archive starts. + /** + * Changed by AddShard and RemoveShard, so that an archive submitted + * before them does not set the next due time when it completes. + */ + uint64_t generation_{0}; + bool attached_{false}; ///< Between AddShard and RemoveShard. + bool in_flight_{false}; ///< An archive of the shard is running. + /** + * The running, or last, archive's request. Replaced and freed only + * while no archive of the shard runs, so a completion never reads a + * freed request. + */ + std::unique_ptr req_; + }; + + /** @brief A due archive the thread took, submitted outside the lock. */ + struct DueArchive + { + DsShardId ds_shard_id_; ///< The shard to archive. + uint64_t generation_; ///< Its schedule's generation then. + GlobalArchiveRequest *req_; ///< Its request, owned by the entry. + }; + + /** @brief The thread: takes the due shards, submits them, waits. */ void Crond(); - void StartArchiving(); + /** + * @brief Submits one due archive, or puts it off if the shard is not + * attached now. + * @param due The archive. + */ + void Submit(const DueArchive &due); + /** + * @brief Records that a shard's archive ended: the next one is due one + * interval from now, unless the schedule ended or restarted meanwhile. + * @param ds_shard_id The shard. + * @param generation The schedule's generation when the archive started. + * @param err How the archive ended. + */ + void OnArchiveDone(DsShardId ds_shard_id, uint64_t generation, KvError err); + /** + * @brief Waits until notified, or until @p until if it has a value. + * @param lk Holds mu_; released while waiting. + * @param until The earliest due time, if any shard is scheduled. + */ + void Wait(Lock &lk, std::optional until); + /** @return KvOptions::archive_interval_secs as a duration. */ + Clock::duration Interval() const; EloqStore *store_{nullptr}; - uint64_t last_archive_ts_; std::thread thd_; + /** Set by the thread as it returns, so Stop can yield until then. */ + std::atomic exited_{false}; #ifdef ELOQ_MODULE_ENABLED bthread::Mutex mu_; @@ -31,7 +135,8 @@ class ArchiveCrond std::mutex mu_; std::condition_variable cond_var_; #endif - bool stop_requested_{true}; - std::atomic stopped_{false}; + bool stop_requested_{true}; ///< Guarded by mu_. + /** The attached shards' schedules, by shard id. Guarded by mu_. */ + std::unordered_map schedules_; }; -} // namespace eloqstore \ No newline at end of file +} // namespace eloqstore diff --git a/include/tasks/attach_shard_task.h b/include/tasks/attach_shard_task.h new file mode 100644 index 00000000..983d2b6c --- /dev/null +++ b/include/tasks/attach_shard_task.h @@ -0,0 +1,32 @@ +#pragma once + +#include "tasks/task.h" + +namespace eloqstore +{ +/** + * @brief Raises a data store shard's term file for EloqStore::AttachShard, + * which sends one AttachShardRequest to one worker shard in cloud mode. + * + * Runs as a task so that the compare-and-swap goes through the worker shard's + * object store like every other cloud request: the task waits on the + * asynchronous downloads and uploads instead of blocking a thread on them. + */ +class AttachShardTask : public KvTask +{ +public: + TaskType Type() const override + { + return TaskType::AttachShard; + } + /** + * @brief Raises the term file of the shard the task carries to the term + * it carries (CloudStoreMgr::UpsertTermFile). Shard::StartTask took that + * term from the shard's slot, where AttachShard set it. + * + * @return NoError; ExpiredTerm if the file holds a newer term; else the + * cloud error. + */ + KvError Attach(); +}; +} // namespace eloqstore diff --git a/include/tasks/detach_shard_task.h b/include/tasks/detach_shard_task.h new file mode 100644 index 00000000..f5af6a95 --- /dev/null +++ b/include/tasks/detach_shard_task.h @@ -0,0 +1,31 @@ +#pragma once + +#include "tasks/task.h" + +namespace eloqstore +{ +/** + * @brief Lets a worker shard go of a data store shard's partitions, for + * EloqStore::DetachShard, which sends one DetachShardRequest per worker shard. + * + * Runs as a task so that the eviction can wait on the io_uring closes of the + * partitions' files like any other task does. + */ +class DetachShardTask : public KvTask +{ +public: + TaskType Type() const override + { + return TaskType::DetachShard; + } + /** + * @brief Fails the requests queued for the shard's partitions, waits + * until no task runs on them, then drops the worker shard's state for + * them (Shard::EvictDsShard). + * + * @param ds_shard_id The data store shard. + * @return NoError, or the first error the eviction met. + */ + KvError Detach(DsShardId ds_shard_id); +}; +} // namespace eloqstore diff --git a/include/tasks/prewarm_task.h b/include/tasks/prewarm_task.h index 0ed16e21..f074411e 100644 --- a/include/tasks/prewarm_task.h +++ b/include/tasks/prewarm_task.h @@ -1,6 +1,7 @@ #pragma once #include #include +#include #include #include #include @@ -110,10 +111,17 @@ class PrewarmService void Stop(); /** - * @brief Signal prewarm service to do a new round of prewarm for those - * specified table partition. + * @brief Queues a round of prewarm for one partition. + * @param ds_shard_id The data store shard the partition is served for. + * @param table The partition. */ - void Prewarm(const TableIdent &table); + void Prewarm(DsShardId ds_shard_id, const TableIdent &table); + /** + * @brief Queues a round of prewarm for every partition of a data store + * shard (those the partition filter keeps for it); AttachShard calls it. + * @param ds_shard_id The data store shard. + */ + void PrewarmShard(DsShardId ds_shard_id); /** * @brief Wait for prewarm operation to complete @@ -131,7 +139,14 @@ class PrewarmService friend class EloqStore; friend class CloudStoreMgr; void AbortPrewarmWorkers(); - void PrewarmCloudCache(const std::string &remote_path); + /** + * @brief Lists and downloads the files under a cloud path for a data + * store shard, skipping partitions the filter does not give it. + * @param remote_path A partition's path, or empty for the whole bucket. + * @param ds_shard_id The data store shard. + */ + void PrewarmCloudCache(const std::string &remote_path, + DsShardId ds_shard_id); bool ListCloudObjects(const std::string &remote_path, std::vector &details, const std::string &continuation_token = std::string{}, @@ -148,7 +163,14 @@ class PrewarmService std::atomic stop_requested_{false}; std::atomic stopped_{true}; - // Standby node do prewarm when received ckpt signal from master node. - moodycamel::BlockingConcurrentQueue prewarm_tables_; + /** @brief One queued round: a partition, or a whole data store shard. */ + struct PrewarmJob + { + DsShardId ds_shard_id_{kUnsetDsShard}; + /** The partition to prewarm; no value means the whole shard. */ + std::optional table_; + }; + // Rounds queued by AttachShard (a shard) and by a partition's reopen. + moodycamel::BlockingConcurrentQueue prewarm_jobs_; }; } // namespace eloqstore diff --git a/include/tasks/task.h b/include/tasks/task.h index e80f7c7c..780808eb 100644 --- a/include/tasks/task.h +++ b/include/tasks/task.h @@ -28,7 +28,18 @@ class EloqStore; inline EloqStore *eloq_store; inline thread_local Shard *shard; +/** @return The task running on this thread's worker shard (Shard::running_). */ KvTask *ThdTask(); +/** + * @return The data store shard the running task works for + * (Shard::running_ds_shard_id_); kUnsetDsShard outside a task. + */ +DsShardId ThdDsShard(); +/** + * @return The term of the running task's data store shard + * (Shard::running_term_); 0 in local mode and outside a task. + */ +uint64_t ThdTerm(); AsyncIoManager *IoMgr(); const KvOptions *Options(); const Comparator *Comp(); @@ -60,6 +71,8 @@ enum struct TaskType Prewarm, ListObject, ListStandbyPartition, + AttachShard, + DetachShard, BatchWrite, Reopen, BackgroundWrite @@ -216,6 +229,18 @@ class KvTask KvRequest *req_{nullptr}; continuation coro_; KvTask *next_{nullptr}; + /** + * The data store shard this task works for: the one its request named. + * Set when the task starts (Shard::StartTask) and recorded on the worker + * shard at every switch to the task (Shard::running_ds_shard_id_). + */ + DsShardId ds_shard_id_{kUnsetDsShard}; + /** + * The term of that shard, read once from the store's shard table when the + * task starts; fixed while the shard stays attached. Recorded as + * Shard::running_term_, which AsyncIoManager::ProcessTerm returns. + */ + uint64_t term_{0}; }; class WaitingZone diff --git a/include/tasks/task_manager.h b/include/tasks/task_manager.h index e87f2d60..8b0a79f6 100644 --- a/include/tasks/task_manager.h +++ b/include/tasks/task_manager.h @@ -6,8 +6,10 @@ #include #include +#include "tasks/attach_shard_task.h" #include "tasks/background_write.h" #include "tasks/batch_write_task.h" +#include "tasks/detach_shard_task.h" #include "tasks/list_object_task.h" #include "tasks/list_standby_partition_task.h" #include "tasks/read_task.h" @@ -38,6 +40,10 @@ class TaskManager ListObjectTask *GetListObjectTask(); ListStandbyPartitionTask *GetListStandbyPartitionTask(); ReopenTask *GetReopenTask(const TableIdent &tbl_id); + /** @return An idle attach task (the pool grows; never null). */ + AttachShardTask *GetAttachShardTask(); + /** @return An idle detach task (the pool grows; never null). */ + DetachShardTask *GetDetachShardTask(); void FreeTask(KvTask *task); void AddExternalTask(); @@ -126,6 +132,8 @@ class TaskManager TaskPool list_object_pool_; TaskPool list_standby_partition_pool_; TaskPool reopen_pool_; + TaskPool attach_shard_pool_; + TaskPool detach_shard_pool_; size_t num_active_{0}; size_t num_active_write_{0}; size_t max_write_concurrency_{std::numeric_limits::max()}; diff --git a/include/types.h b/include/types.h index 01372c7b..4449cb33 100644 --- a/include/types.h +++ b/include/types.h @@ -35,7 +35,19 @@ constexpr FilePageId MaxFilePageId = UINT64_MAX; using FileId = uint64_t; static constexpr FileId MaxFileId = UINT64_MAX; -using PartitonGroupId = uint32_t; +/** + * @brief Identifies a data store shard: a set of partitions an embedder hosts + * on this store as one unit, with its own term (in EloqDB, node group g's data + * is data store shard g). + * + * One store hosts several of them at once (EloqStore::AttachShard). The id + * names the shard's cloud term file, CURRENT_TERM__. + */ +using DsShardId = uint32_t; +/** @brief The data store shard id's former name, kept for embedders. */ +using PartitonGroupId = DsShardId; +/** @brief A request that names no data store shard (see KvRequest). */ +constexpr DsShardId kUnsetDsShard = UINT32_MAX; constexpr char FileNameSeparator = '_'; static constexpr char FileNameData[] = "data"; diff --git a/rust/eloqstore-sys/vendor/CMakeLists.txt b/rust/eloqstore-sys/vendor/CMakeLists.txt index 9bc8cd67..4ac90a99 100644 --- a/rust/eloqstore-sys/vendor/CMakeLists.txt +++ b/rust/eloqstore-sys/vendor/CMakeLists.txt @@ -351,6 +351,8 @@ set(ELOQ_STORE_SOURCES src/tasks/archive_crond.cpp src/tasks/background_write.cpp src/tasks/batch_write_task.cpp + src/tasks/attach_shard_task.cpp + src/tasks/detach_shard_task.cpp src/tasks/write_buffer_aggregator.cpp src/tasks/prewarm_task.cpp src/tasks/read_task.cpp diff --git a/src/async_io_manager.cpp b/src/async_io_manager.cpp index 8e676ab9..9ea8b7cc 100644 --- a/src/async_io_manager.cpp +++ b/src/async_io_manager.cpp @@ -1385,6 +1385,32 @@ KvError IouringMgr::CloseFiles(const TableIdent &tbl_id, return CloseFiles(std::span(fd_refs.data(), fd_refs.size())); } +KvError IouringMgr::CloseTableFiles(const TableIdent &tbl_id) +{ + auto it_tbl = tables_.find(tbl_id); + if (it_tbl == tables_.end()) + { + return KvError::NoError; + } + // Collect the ids first: taking a Ref and closing mutate fds_. + std::vector file_ids; + file_ids.reserve(it_tbl->second.fds_.size()); + for (const auto &[file_id, _] : it_tbl->second.fds_) + { + file_ids.push_back(file_id); + } + return CloseFiles(tbl_id, std::span(file_ids)); +} + +void IouringMgr::CollectPartitions( + std::unordered_set &partitions) const +{ + for (const auto &[tbl_id, files] : tables_) + { + partitions.insert(tbl_id); + } +} + size_t IouringMgr::GetOpenFileCount() const { return lru_fd_count_; @@ -3038,8 +3064,8 @@ KvError IouringMgr::DeleteBranchFiles( FdIdx root_fd_idx{root_fd, false}; std::string current_term_filename = - CurrentTermFileNameForBranchAndPartitionGroup( - branch_name, eloq_store->PartitionGroupId()); + CurrentTermFileNameForBranchAndPartitionGroup(branch_name, + ThdDsShard()); // Read CURRENT_TERM__ from store root to find the highest // term written for this branch. If the file is missing, fall back to term @@ -4469,7 +4495,8 @@ std::pair CloudStoreMgr::GetManifest( // Use active branch for all local manifest filenames in this function. std::string active_br = std::string(GetActiveBranch()); - // Term file is updated at startup only (BootstrapUpsertTermFile). + // The term file is raised when the data store shard is attached + // (UpsertTermFile, from AttachShardTask), not here. KvError dl_err = DownloadFile(tbl_id, LruFD::kManifest, active_br, process_term, false); @@ -4623,7 +4650,7 @@ std::pair CloudStoreMgr::GetManifest( return {nullptr, dl_err}; } - // If ProcessTerm() is set and the selected term is older than + // If the shard's term is set and the selected term is older than // process_term, "promote" the manifest: copy its content into a new // manifest__ object (both locally and in cloud), so // subsequent readers can consistently use manifest__. @@ -5029,13 +5056,13 @@ std::pair CloudStoreMgr::RefreshManifest( } std::tuple CloudStoreMgr::ReadTermFile( - std::string_view branch_name) + std::string_view branch_name, DsShardId ds_shard_id) { KvTask *current_task = ThdTask(); - // Download CURRENT_TERM__ file from store root - std::string term_filename = CurrentTermFileNameForBranchAndPartitionGroup( - branch_name, partition_group_id_); + // Download CURRENT_TERM__ file from store root + std::string term_filename = + CurrentTermFileNameForBranchAndPartitionGroup(branch_name, ds_shard_id); ObjectStore::DownloadTask download_task(term_filename); download_task.SetKvTask(current_task); AcquireCloudSlot(current_task); @@ -5079,6 +5106,69 @@ std::tuple CloudStoreMgr::ReadTermFile( return {term, download_task.etag_, KvError::NoError}; } +KvError CloudStoreMgr::UpsertTermFile(std::string_view branch_name, + DsShardId ds_shard_id, + uint64_t term) +{ + KvTask *current_task = ThdTask(); + const std::string term_filename = + CurrentTermFileNameForBranchAndPartitionGroup(branch_name, ds_shard_id); + const std::string term_str = std::to_string(term); + constexpr int kMaxAttempts = 10; + for (int attempt = 0; attempt < kMaxAttempts; ++attempt) + { + auto [current_term, etag, read_err] = + ReadTermFile(branch_name, ds_shard_id); + const bool missing = read_err == KvError::NotFound; + if (!missing) + { + if (read_err != KvError::NoError) + { + return read_err; + } + if (current_term > term) + { + return KvError::ExpiredTerm; + } + if (current_term == term) + { + return KvError::NoError; + } + } + + ObjectStore::UploadTask upload_task(term_filename, true); + upload_task.data_buffer_.append(term_str); + upload_task.file_size_ = term_str.size(); + if (missing) + { + upload_task.if_none_match_ = "*"; + } + else + { + upload_task.if_match_ = etag; + } + upload_task.SetKvTask(current_task); + AcquireCloudSlot(current_task); + obj_store_.SubmitTask(&upload_task, shard); + current_task->WaitIo(); + if (upload_task.error_ == KvError::NoError) + { + return KvError::NoError; + } + + // Another writer changed the file since the read: 412 when the + // condition failed, 409 when a concurrent upload of the key is in + // progress, and 404 when the file read was deleted before the + // If-Match upload. Read it again. + const int64_t code = upload_task.response_code_; + if (code != 409 && code != 412 && (missing || code != 404)) + { + return upload_task.error_; + } + } + return KvError::CloudErr; +} + void CloudStoreMgr::ScheduleLocalFileCleanup( const TableIdent &tbl_id, const std::vector &filenames) { @@ -5415,7 +5505,7 @@ KvError CloudStoreMgr::DeleteBranchFiles(const TableIdent &tbl_id, std::string normalized_branch = NormalizeBranchName(branch_name); std::string current_term_filename = CurrentTermFileNameForBranchAndPartitionGroup(branch_name, - partition_group_id_); + ThdDsShard()); // Helper lambda: list all cloud objects under a given prefix and append // matching paths to paths_to_delete. The full object path stored in @@ -5662,8 +5752,11 @@ KvError CloudStoreMgr::SyncFile(LruFD::Ref fd) if (file_id == LruFD::kManifest) { // Use branch-aware manifest filename for both local read and cloud - // upload key, consistent with SwitchManifest. - filename = BranchManifestFileName(GetActiveBranch(), ProcessTerm()); + // upload key, consistent with SwitchManifest. The term is the + // fd's, as for data files: an fd eviction syncs here from a task + // that may work for another data store shard, whose term + // ProcessTerm() would give. + filename = BranchManifestFileName(GetActiveBranch(), term); } else { @@ -5766,8 +5859,8 @@ KvError CloudStoreMgr::SyncFiles(const TableIdent &tbl_id, std::string filename; if (file_id == LruFD::kManifest) { - filename = - BranchManifestFileName(GetActiveBranch(), ProcessTerm()); + // The fd's term, as in SyncFile. + filename = BranchManifestFileName(GetActiveBranch(), term); } else { diff --git a/src/cloud_storage_service.cpp b/src/cloud_storage_service.cpp index 27a75d8e..e5645d08 100644 --- a/src/cloud_storage_service.cpp +++ b/src/cloud_storage_service.cpp @@ -44,16 +44,6 @@ void CloudStorageService::Start() { return; } -#ifdef ELOQ_MODULE_ENABLED - { - std::lock_guard lk(bootstrap_state_.mutex_); - bootstrap_state_.err_ = KvError::Busy; - bootstrap_state_.done_ = false; - } -#else - bootstrap_state_.err_ = KvError::Busy; - bootstrap_state_.done_.store(false, std::memory_order_relaxed); -#endif accepting_jobs_.store(true, std::memory_order_release); workers_.reserve(worker_count_); for (size_t i = 0; i < worker_count_; ++i) @@ -149,29 +139,6 @@ void CloudStorageService::RunWorker(size_t worker_index) const int64_t wait_timeout_us = std::chrono::duration_cast(kIdleWait) .count(); - if (worker_index == 0) - { - KvError err = KvError::Busy; - if (stopping_.load(std::memory_order_acquire)) - { - err = KvError::NotRunning; - } - else - { - ObjectStore *bootstrap_store = nullptr; - { - std::shared_lock lk(shard_locks_.front().mutex); - bootstrap_store = shard_stores_.front(); - } - err = bootstrap_store == nullptr - ? KvError::CloudErr - : bootstrap_store->BootstrapUpsertTermFile( - CurrentTermFileNameForBranchAndPartitionGroup( - store_->Branch(), store_->PartitionGroupId()), - store_->Term()); - } - bootstrap_state_.SetDone(err); - } while (true) { diff --git a/src/eloq_store.cpp b/src/eloq_store.cpp index 5f2fbe15..2c6299e9 100644 --- a/src/eloq_store.cpp +++ b/src/eloq_store.cpp @@ -17,6 +17,7 @@ #include #include #include +#include #include #include #include @@ -56,6 +57,7 @@ constexpr uint64_t kStorePathWeightGranularity = 1ULL << 20; // 1 MiB constexpr size_t kMaxStorePathLutEntries = kDefaultStorePathLutEntries; KvError CollectLocalPartitions(const KvOptions &options, + DsShardId ds_shard_id, std::vector &partitions) { partitions.clear(); @@ -95,7 +97,8 @@ KvError CollectLocalPartitions(const KvOptions &options, LOG(WARNING) << "unexpected partition " << entry.path(); continue; } - if (options.partition_filter && !options.partition_filter(tbl_id)) + if (options.partition_filter && + !options.partition_filter(ds_shard_id, tbl_id)) { continue; } @@ -657,9 +660,32 @@ void EloqStore::CleanupRuntime(size_t started_shards) KvError EloqStore::Start(std::string_view branch, uint64_t term, - PartitonGroupId partition_group_id) + DsShardId ds_shard_id) { - LOG(INFO) << "===Start eloqstore, branch: " << branch << ", term: " << term; + if (status_.load(std::memory_order_acquire) == Status::Running) + { + // A running store keeps the shards it hosts. + LOG(ERROR) << "EloqStore started , do not start again"; + return KvError::NoError; + } + KvError err = StartInstance(branch); + if (err != KvError::NoError) + { + return err; + } + err = AttachShard(ds_shard_id, term); + if (err != KvError::NoError) + { + // A failed start leaves the store stopped, as it always has. + Stop(); + return err; + } + return KvError::NoError; +} + +KvError EloqStore::StartInstance(std::string_view branch) +{ + LOG(INFO) << "===Start eloqstore, branch: " << branch; if (!ValidateOptions(options_)) { return KvError::InvalidArgs; @@ -718,9 +744,7 @@ KvError EloqStore::Start(std::string_view branch, StoreMode mode = DeriveStoreMode(options_); const StoreMode prev_mode = store_mode_.exchange(mode, std::memory_order_acq_rel); - LOG(INFO) << "===Start eloqstore, term: " << term - << ", partition_group_id: " << partition_group_id - << ", mode: " << static_cast(mode); + LOG(INFO) << "===Start eloqstore, mode: " << static_cast(mode); if (prev_mode != mode) { LOG(INFO) << "EloqStore::Start update store mode, prev_mode=" @@ -761,9 +785,6 @@ KvError EloqStore::Start(std::string_view branch, } } - // local mode, set term to 0 - term_ = mode == StoreMode::Local ? 0 : term; - partition_group_id_ = mode == StoreMode::Cloud ? partition_group_id : 0; branch_ = std::string(branch); // EloqStore runs as a library and shares the process fd table with the @@ -848,12 +869,10 @@ KvError EloqStore::Start(std::string_view branch, if (cloud_service_) { + // Nothing to wait for here, unlike with one store-wide term: each data + // store shard's term file is raised by its AttachShard, which admits + // none of the shard's requests until the raise succeeds. cloud_service_->Start(); - cloud_service_->bootstrap_state_.Wait(); - if (cloud_service_->bootstrap_state_.err_ != KvError::NoError) - { - return fail_start(cloud_service_->bootstrap_state_.err_); - } } else if (standby_service_) { @@ -900,6 +919,262 @@ KvError EloqStore::Start(std::string_view branch, return KvError::NoError; } +KvError EloqStore::AttachShard(DsShardId ds_shard_id, uint64_t term) +{ + if (ds_shard_id >= kMaxDsShards) + { + LOG(ERROR) << "AttachShard: data store shard id " << ds_shard_id + << " is out of range"; + return KvError::InvalidArgs; + } + const StoreMode mode = Mode(); + // Local mode has no terms. + const uint64_t shard_term = mode == StoreMode::Local ? 0 : term; + DsShardSlot &slot = ds_shards_[ds_shard_id]; + { + // Concurrent AttachShard, DetachShard and Stop calls are serialized + // by ds_shard_mux_ (see its declaration). Here it makes the check + // that no other shard is hosted atomic with claiming this slot. It is + // released before the term-file wait below. + std::lock_guard lk(ds_shard_mux_); + if (status_.load(std::memory_order_acquire) != Status::Running) + { + return KvError::NotRunning; + } + // Would this attach leave the store hosting more than one data store + // shard? Find another shard whose slot is not Detached. Every other + // state counts: an Attached or Expired shard holds state until it is + // detached; an Attaching one is another AttachShard between its two + // critical sections; a Detaching one is still dropping its state on + // the worker shards, and without a partition filter that detach drops + // every partition, this shard's included. + std::optional other_shard; + for (size_t i = 0; i < kMaxDsShards; ++i) + { + if (i != ds_shard_id && + ds_shards_[i].state_.load(std::memory_order_acquire) != + DsShardState::Detached) + { + other_shard = static_cast(i); + break; + } + } + if (other_shard.has_value()) + { + if (mode != StoreMode::Cloud) + { + // Data on a node's own disk has one owner: one shard at a + // time. + LOG(ERROR) << "AttachShard: data store shard " << *other_shard + << " is hosted, and only cloud mode hosts several"; + return KvError::InvalidArgs; + } + if (!options_.partition_filter) + { + // Without the filter every partition belongs to every shard, + // and a detach could not tell the shards' state apart. + LOG(ERROR) << "AttachShard: data store shard " << *other_shard + << " is hosted, and hosting several needs " + "KvOptions::partition_filter"; + return KvError::InvalidArgs; + } + } + DsShardState expected = DsShardState::Detached; + if (!slot.state_.compare_exchange_strong( + expected, DsShardState::Attaching, std::memory_order_acq_rel)) + { + LOG(ERROR) << "AttachShard: data store shard " << ds_shard_id + << " is not detached (state " + << static_cast(expected) << ")"; + return KvError::InvalidArgs; + } + slot.term_.store(shard_term, std::memory_order_release); + } + LOG(INFO) << "Attaching data store shard " << ds_shard_id << " at term " + << shard_term; + auto give_up = [&slot]() + { + slot.term_.store(0, std::memory_order_release); + slot.state_.store(DsShardState::Detached, std::memory_order_release); + }; + if (mode == StoreMode::Cloud) + { + // Raise the shard's term file, or learn that a newer term holds it. + // A task on one worker shard runs the compare-and-swap with the + // shard's term set above (AttachShardTask); the id picks the worker + // shard, so attaches of different shards spread over them. The wait + // parks a bthread caller. A refused request means the store is + // stopping. + AttachShardRequest req; + req.SetDsShard(ds_shard_id); + req.err_ = KvError::NoError; +#ifdef ELOQ_MODULE_ENABLED + { + std::lock_guard req_lk(req.mutex_); + req.done_ = false; + } +#else + req.done_.store(false, std::memory_order_relaxed); +#endif + KvError err = KvError::NotRunning; + if (shards_[ds_shard_id % shards_.size()]->AddKvRequest(&req)) + { + req.Wait(); + err = req.Error(); + } + if (err != KvError::NoError) + { + LOG(ERROR) << "AttachShard: data store shard " << ds_shard_id + << " could not take term " << shard_term << ": " + << ErrorString(err); + give_up(); + return err; + } + } + { + std::lock_guard lk(ds_shard_mux_); + if (status_.load(std::memory_order_acquire) != Status::Running) + { + // Stop ran during the wait and reset the shard table. + give_up(); + return KvError::NotRunning; + } + DsShardId unset = kUnsetDsShard; + default_ds_shard_.compare_exchange_strong( + unset, ds_shard_id, std::memory_order_acq_rel); + slot.state_.store(DsShardState::Attached, std::memory_order_release); + } + if (archive_crond_ != nullptr) + { + // The shard is archived on its own schedule from now until its + // detach, the first time one interval from now. + archive_crond_->AddShard(ds_shard_id); + } + if (prewarm_service_ != nullptr) + { + prewarm_service_->PrewarmShard(ds_shard_id); + } + return KvError::NoError; +} + +KvError EloqStore::DetachShard(DsShardId ds_shard_id) +{ + if (ds_shard_id >= kMaxDsShards) + { + LOG(ERROR) << "DetachShard: data store shard id " << ds_shard_id + << " is out of range"; + return KvError::InvalidArgs; + } + DsShardSlot &slot = ds_shards_[ds_shard_id]; + bool running; + { + // Under ds_shard_mux_, as in AttachShard; released before waiting for + // the worker shards below. + std::lock_guard lk(ds_shard_mux_); + switch (slot.state_.load(std::memory_order_acquire)) + { + case DsShardState::Detached: + return KvError::NoError; + case DsShardState::Attaching: + case DsShardState::Detaching: + LOG(ERROR) << "DetachShard: data store shard " << ds_shard_id + << " is being attached or detached"; + return KvError::InvalidArgs; + default: + break; + } + // From here on the worker shards refuse the shard's requests. + slot.state_.store(DsShardState::Detaching, std::memory_order_release); + running = status_.load(std::memory_order_acquire) == Status::Running; + } + if (archive_crond_ != nullptr) + { + // No new archive of the shard starts. One in flight is refused or + // finished by the drain below, like any other request of the shard. + archive_crond_->RemoveShard(ds_shard_id); + } + LOG(INFO) << "Detaching data store shard " << ds_shard_id; + if (running) + { + // One request per worker shard, each completing once its shard holds + // nothing of this one. A refused request means the store is stopping, + // and its shards drain everything then. The vector is sized once and + // never grows, so the requests the worker shards hold stay in place. + std::vector reqs(shards_.size()); + std::vector sent(shards_.size(), false); + for (size_t i = 0; i < shards_.size(); ++i) + { + DetachShardRequest &req = reqs[i]; + req.SetDsShard(ds_shard_id); + req.err_ = KvError::NoError; +#ifdef ELOQ_MODULE_ENABLED + { + std::lock_guard req_lk(req.mutex_); + req.done_ = false; + } +#else + req.done_.store(false, std::memory_order_relaxed); +#endif + sent[i] = shards_[i]->AddKvRequest(&req); + } + for (size_t i = 0; i < shards_.size(); ++i) + { + if (sent[i]) + { + reqs[i].Wait(); + if (reqs[i].Error() != KvError::NoError) + { + LOG(WARNING) + << "DetachShard: worker shard " << i << " reported " + << reqs[i].ErrMessage() + << " while dropping data store shard " << ds_shard_id; + } + } + } + } + { + std::lock_guard lk(ds_shard_mux_); + DsShardId me = ds_shard_id; + default_ds_shard_.compare_exchange_strong( + me, kUnsetDsShard, std::memory_order_acq_rel); + slot.term_.store(0, std::memory_order_release); + slot.state_.store(DsShardState::Detached, std::memory_order_release); + } + LOG(INFO) << "Detached data store shard " << ds_shard_id; + return KvError::NoError; +} + +std::vector EloqStore::HostedShards() const +{ + std::vector hosted; + for (size_t i = 0; i < kMaxDsShards; ++i) + { + const DsShardState state = + ds_shards_[i].state_.load(std::memory_order_acquire); + if (state == DsShardState::Attached || state == DsShardState::Expired) + { + hosted.push_back(static_cast(i)); + } + } + return hosted; +} + +void EloqStore::MarkShardExpired(DsShardId ds_shard_id) +{ + if (ds_shard_id >= kMaxDsShards) + { + return; + } + DsShardState expected = DsShardState::Attached; + if (ds_shards_[ds_shard_id].state_.compare_exchange_strong( + expected, DsShardState::Expired, std::memory_order_acq_rel)) + { + LOG(WARNING) << "Data store shard " << ds_shard_id << " at term " + << ShardTerm(ds_shard_id) + << " has expired: a newer term owns a partition of it"; + } +} + KvError EloqStore::InitStoreSpace() { const bool cloud_store = !options_.cloud_store_path.empty(); @@ -1263,6 +1538,17 @@ void EloqStore::HandleDropTableRequest(DropTableRequest *req) req->SetDone(err); return; } + if (options_.partition_filter) + { + // The table's partitions are spread over the data store shards: drop + // only those of the shard the request works for, as the other + // fan-outs do. Without a filter the store hosts one shard, which has + // them all. + std::erase_if( + partitions, + [this, req](const TableIdent &tbl_id) + { return !options_.partition_filter(req->DsShard(), tbl_id); }); + } if (partitions.empty()) { @@ -1278,6 +1564,7 @@ void EloqStore::HandleDropTableRequest(DropTableRequest *req) { auto drop_req = std::make_unique(); drop_req->SetArgs(partition); + drop_req->SetDsShard(req->DsShard()); req->drop_reqs_.push_back(std::move(drop_req)); } @@ -1394,7 +1681,8 @@ void EloqStore::HandleGlobalArchiveRequest(GlobalArchiveRequest *req) : "delete") << " tag=" << tag << ", term=" << term; - req->result_archive_ = BranchArchiveName(branch_, term_, tag); + req->result_archive_ = + BranchArchiveName(branch_, ShardTerm(req->DsShard()), tag); std::vector all_partitions; if (options_.cloud_store_path.empty()) @@ -1438,7 +1726,7 @@ void EloqStore::HandleGlobalArchiveRequest(GlobalArchiveRequest *req) } if (options_.partition_filter && - !options_.partition_filter(tbl_id)) + !options_.partition_filter(req->DsShard(), tbl_id)) { continue; } @@ -1490,7 +1778,7 @@ void EloqStore::HandleGlobalArchiveRequest(GlobalArchiveRequest *req) } if (options_.partition_filter && - !options_.partition_filter(tbl_id)) + !options_.partition_filter(req->DsShard(), tbl_id)) { continue; } @@ -1527,6 +1815,7 @@ void EloqStore::HandleGlobalArchiveRequest(GlobalArchiveRequest *req) : ArchiveRequest::Action::Delete); archive_req->SetTag(tag); archive_req->SetTerm(term); + archive_req->SetDsShard(req->DsShard()); req->archive_reqs_.push_back(std::move(archive_req)); } @@ -1666,7 +1955,8 @@ void EloqStore::HandleGlobalReopenRequest(GlobalReopenRequest *req) { continue; } - if (options_.partition_filter && !options_.partition_filter(tbl_id)) + if (options_.partition_filter && + !options_.partition_filter(req->DsShard(), tbl_id)) { continue; } @@ -1675,7 +1965,8 @@ void EloqStore::HandleGlobalReopenRequest(GlobalReopenRequest *req) } std::vector local_partitions; - KvError local_err = CollectLocalPartitions(options_, local_partitions); + KvError local_err = + CollectLocalPartitions(options_, req->DsShard(), local_partitions); if (local_err != KvError::NoError) { req->SetDone(local_err); @@ -1692,7 +1983,8 @@ void EloqStore::HandleGlobalReopenRequest(GlobalReopenRequest *req) } else { - KvError local_err = CollectLocalPartitions(options_, partitions); + KvError local_err = + CollectLocalPartitions(options_, req->DsShard(), partitions); if (local_err != KvError::NoError) { req->SetDone(local_err); @@ -1718,6 +2010,7 @@ void EloqStore::HandleGlobalReopenRequest(GlobalReopenRequest *req) { auto reopen_req = std::make_unique(); reopen_req->SetArgs(partition); + reopen_req->SetDsShard(req->DsShard()); if (!req->Tag().empty()) { reopen_req->SetTag(req->Tag()); @@ -2009,7 +2302,7 @@ void EloqStore::HandleGlobalCreateBranchRequest(GlobalCreateBranchRequest *req) } if (options_.partition_filter && - !options_.partition_filter(tbl_id)) + !options_.partition_filter(req->DsShard(), tbl_id)) { continue; } @@ -2062,7 +2355,7 @@ void EloqStore::HandleGlobalCreateBranchRequest(GlobalCreateBranchRequest *req) } if (options_.partition_filter && - !options_.partition_filter(tbl_id)) + !options_.partition_filter(req->DsShard(), tbl_id)) { continue; } @@ -2094,6 +2387,7 @@ void EloqStore::HandleGlobalCreateBranchRequest(GlobalCreateBranchRequest *req) auto branch_req = std::make_unique(); branch_req->SetTableId(partition); branch_req->SetArgs(internal_name); + branch_req->SetDsShard(req->DsShard()); req->branch_reqs_.push_back(std::move(branch_req)); } @@ -2195,6 +2489,35 @@ bool EloqStore::SendRequest(KvRequest *req) return false; } + // Admission at submission; the request runs for a hosted data store shard, + // or not at all. + // - A request naming no shard is given the default one here. + // - For the store-level requests (DropTable and the Global* requests, + // handled below on this thread), this is the only check: they never + // reach a worker shard themselves, only the per-partition requests they + // fan out do. + // - A per-partition request is checked again when its worker shard + // dequeues it (Shard::AdmitRequest), which catches a detach that begins + // after this check. Refusing it here as well makes ExecAsyn return false + // at once rather than complete it through the worker. + if (req->ds_shard_id_ == kUnsetDsShard) + { + req->ds_shard_id_ = default_ds_shard_.load(std::memory_order_acquire); + } + switch (ShardState(req->ds_shard_id_)) + { + case DsShardState::Attached: + case DsShardState::Expired: + // An expired shard's request still goes to its worker shard, which + // fails it with ExpiredTerm there (Shard::AdmitRequest). Completing + // it here would run the caller's callback inside this call, which + // the fan-out schedulers do not expect: each would recurse into the + // next partition's submission. + break; + default: + return false; + } + req->err_ = KvError::NoError; #ifdef ELOQ_MODULE_ENABLED { @@ -2284,6 +2607,17 @@ void EloqStore::Stop() } } CleanupRuntime(shards_.size()); + { + // The worker shards have dropped everything: nothing is hosted now. + std::lock_guard lk(ds_shard_mux_); + for (DsShardSlot &slot : ds_shards_) + { + slot.state_.store(DsShardState::Detached, + std::memory_order_release); + slot.term_.store(0, std::memory_order_release); + } + default_ds_shard_.store(kUnsetDsShard, std::memory_order_release); + } status_.store(Status::Stopped, std::memory_order_release); LOG(INFO) << "EloqStore is stopped."; } @@ -2337,7 +2671,9 @@ void EloqStore::InitializeMetrics(metrics::MetricsRegistry *metrics_registry, "file_gc", "clean_expired", "create_branch", - "delete_branch"}}}); + "delete_branch", + "attach_shard", + "detach_shard"}}}); metrics_meters_[i]->Register(metrics::NAME_ELOQSTORE_REQUESTS_COMPLETED, metrics::Type::Counter, {{"request_type", @@ -2357,7 +2693,9 @@ void EloqStore::InitializeMetrics(metrics::MetricsRegistry *metrics_registry, "file_gc", "clean_expired", "create_branch", - "delete_branch"}}}); + "delete_branch", + "attach_shard", + "detach_shard"}}}); metrics_meters_[i]->Register( metrics::NAME_ELOQSTORE_INDEX_BUFFER_POOL_USED, metrics::Type::Gauge); diff --git a/src/file_gc.cpp b/src/file_gc.cpp index b674fa0f..a8c5ba0d 100644 --- a/src/file_gc.cpp +++ b/src/file_gc.cpp @@ -1496,19 +1496,20 @@ KvError ExecuteCloudGC(const TableIdent &tbl_id, // Check term file before proceeding uint64_t process_term = cloud_mgr->ProcessTerm(); auto [term_file_term, etag, err] = - cloud_mgr->ReadTermFile(cloud_mgr->GetActiveBranch()); + cloud_mgr->ReadTermFile(cloud_mgr->GetActiveBranch(), ThdDsShard()); if (err == KvError::NotFound) { - LOG(INFO) << "ExecuteCloudGC: term file not found for partition_group " - << cloud_mgr->PartitionGroupId(); + LOG(INFO) << "ExecuteCloudGC: term file not found for data store " + "shard " + << ThdDsShard(); return KvError::NoError; } else if (err != KvError::NoError) { LOG(ERROR) << "ExecuteCloudGC: failed to read term file for " - << "partition_group " << cloud_mgr->PartitionGroupId() - << " : " << ErrorString(err); + << "data store shard " << ThdDsShard() << " : " + << ErrorString(err); return err; } else @@ -1517,8 +1518,8 @@ KvError ExecuteCloudGC(const TableIdent &tbl_id, { LOG(WARNING) << "ExecuteCloudGC: term file term " << term_file_term << " != process_term " << process_term << " for table " - << tbl_id << ", partition_group " - << cloud_mgr->PartitionGroupId() << ", skipping GC"; + << tbl_id << ", data store shard " << ThdDsShard() + << ", skipping GC"; return KvError::ExpiredTerm; } } diff --git a/src/storage/object_store.cpp b/src/storage/object_store.cpp index 6f623325..d7df745b 100644 --- a/src/storage/object_store.cpp +++ b/src/storage/object_store.cpp @@ -173,16 +173,6 @@ KvError ObjectStore::EnsureBucketExists() return async_http_mgr_->EnsureBucketExists(); } -KvError ObjectStore::BootstrapUpsertTermFile(std::string_view remote_path, - uint64_t process_term) -{ - if (!async_http_mgr_) - { - return KvError::CloudErr; - } - return async_http_mgr_->BootstrapUpsertTermFile(remote_path, process_term); -} - void ObjectStore::SubmitTask(ObjectStore::Task *task, Shard *owner_shard) { CHECK(task != nullptr); @@ -382,194 +372,6 @@ KvError AsyncHttpManager::EnsureBucketExists() return result; } -KvError AsyncHttpManager::BootstrapUpsertTermFile(std::string_view remote_path, - uint64_t process_term) -{ - auto perform_download = - [this, - remote_path]() mutable -> std::tuple - { - ObjectStore::DownloadTask task{std::string(remote_path)}; - CURL *easy = curl_easy_init(); - if (!easy) - { - return {"", "", KvError::CloudErr}; - } - - curl_easy_setopt(easy, CURLOPT_PROXY, ""); - curl_easy_setopt(easy, CURLOPT_NOPROXY, "*"); - curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, WriteCallback); - curl_easy_setopt(easy, CURLOPT_WRITEDATA, &task); - curl_easy_setopt(easy, CURLOPT_HEADERFUNCTION, HeaderCallback); - curl_easy_setopt(easy, CURLOPT_HEADERDATA, &task); - curl_easy_setopt(easy, CURLOPT_TIMEOUT, 300L); - - if (!SetupDownloadRequest(&task, easy)) - { - CleanupTaskResources(&task); - curl_easy_cleanup(easy); - return {"", "", task.error_}; - } - - CURLcode res = curl_easy_perform(easy); - int64_t response_code = 0; - if (res == CURLE_OK) - { - curl_easy_getinfo(easy, CURLINFO_RESPONSE_CODE, &response_code); - } - curl_easy_cleanup(easy); - CleanupTaskResources(&task); - - if (res != CURLE_OK) - { - return {"", "", ClassifyCurlError(res)}; - } - if (response_code >= 200 && response_code < 300) - { - return {std::string(task.response_data_.view()), - std::move(task.etag_), - KvError::NoError}; - } - if (response_code == 404) - { - return {"", "", KvError::NotFound}; - } - return {"", "", ClassifyHttpError(response_code)}; - }; - - auto perform_upload = - [this, remote_path]( - std::string_view data, - std::string_view if_match, - std::string_view if_none_match) -> std::pair - { - ObjectStore::UploadTask task{std::string(remote_path), true}; - task.data_buffer_.append(data); - task.file_size_ = data.size(); - task.if_match_ = std::string(if_match); - task.if_none_match_ = std::string(if_none_match); - - CURL *easy = curl_easy_init(); - if (!easy) - { - return {KvError::CloudErr, 0}; - } - - curl_easy_setopt(easy, CURLOPT_PROXY, ""); - curl_easy_setopt(easy, CURLOPT_NOPROXY, "*"); - curl_easy_setopt(easy, CURLOPT_WRITEFUNCTION, WriteCallback); - curl_easy_setopt(easy, CURLOPT_WRITEDATA, &task); - curl_easy_setopt(easy, CURLOPT_HEADERFUNCTION, HeaderCallback); - curl_easy_setopt(easy, CURLOPT_HEADERDATA, &task); - curl_easy_setopt(easy, CURLOPT_TIMEOUT, 300L); - - if (!SetupUploadRequest(&task, easy)) - { - CleanupTaskResources(&task); - curl_easy_cleanup(easy); - return {task.error_, 0}; - } - - CURLcode res = curl_easy_perform(easy); - int64_t response_code = 0; - if (res == CURLE_OK) - { - curl_easy_getinfo(easy, CURLINFO_RESPONSE_CODE, &response_code); - } - curl_easy_cleanup(easy); - CleanupTaskResources(&task); - - if (res != CURLE_OK) - { - return {ClassifyCurlError(res), response_code}; - } - if (response_code >= 200 && response_code < 300) - { - return {KvError::NoError, response_code}; - } - if (response_code == 404) - { - return {KvError::NotFound, response_code}; - } - return {ClassifyHttpError(response_code), response_code}; - }; - - constexpr uint64_t kMaxAttempts = 10; - for (uint64_t attempt = 0; attempt < kMaxAttempts; ++attempt) - { - auto [body, etag, read_err] = perform_download(); - if (read_err == KvError::NotFound) - { - auto [create_err, response_code] = - perform_upload(std::to_string(process_term), "", "*"); - if (create_err == KvError::NoError) - { - return KvError::NoError; - } - if (create_err == KvError::CloudErr && - (response_code == 409 || response_code == 412)) - { - continue; - } - return create_err; - } - if (read_err != KvError::NoError) - { - return read_err; - } - - std::string_view term_str = body; - while (!term_str.empty() && - (term_str.front() == ' ' || term_str.front() == '\t' || - term_str.front() == '\r' || term_str.front() == '\n')) - { - term_str.remove_prefix(1); - } - while (!term_str.empty() && - (term_str.back() == ' ' || term_str.back() == '\t' || - term_str.back() == '\r' || term_str.back() == '\n')) - { - term_str.remove_suffix(1); - } - - uint64_t current_term = 0; - try - { - current_term = std::stoull(std::string(term_str)); - } - catch (const std::exception &) - { - return KvError::Corrupted; - } - - if (current_term > process_term) - { - return KvError::ExpiredTerm; - } - if (current_term == process_term) - { - return KvError::NoError; - } - - auto [update_err, response_code] = - perform_upload(std::to_string(process_term), etag, ""); - if (update_err == KvError::NoError) - { - return KvError::NoError; - } - if (update_err == KvError::NotFound || - (update_err == KvError::CloudErr && - (response_code == 404 || response_code == 409 || - response_code == 412))) - { - continue; - } - return update_err; - } - - return KvError::CloudErr; -} - void AsyncHttpManager::PerformRequests() { ProcessPendingRetries(); @@ -793,6 +595,12 @@ bool AsyncHttpManager::SetupUploadRequest(ObjectStore::UploadTask *task, "Content-Length: " + std::to_string(task->file_size_); task->headers_ = curl_slist_append(task->headers_, content_length.c_str()); task->headers_ = curl_slist_append(task->headers_, "Expect:"); + // The URL is presigned over the host header only. Left alone, curl adds + // "Content-Type: application/x-www-form-urlencoded" to a POSTFIELDS body, + // a header nothing signed, and an S3 front end that folds it into its + // signature check (s3proxy) rejects every upload. An empty value makes + // curl send no Content-Type at all. + task->headers_ = curl_slist_append(task->headers_, "Content-Type:"); // Add conditional headers if provided if (!task->if_match_.empty()) diff --git a/src/storage/root_meta.cpp b/src/storage/root_meta.cpp index 2d0413fb..936281a1 100644 --- a/src/storage/root_meta.cpp +++ b/src/storage/root_meta.cpp @@ -229,6 +229,15 @@ std::pair RootMetaMgr::GetOrCreate( return {entry, inserted}; } +void RootMetaMgr::CollectPartitions( + std::unordered_set &partitions) const +{ + for (const auto &[tbl_id, entry] : entries_) + { + partitions.insert(tbl_id); + } +} + RootMetaMgr::Entry *RootMetaMgr::FindNoErase(const TableIdent &tbl_id) { auto it = entries_.find(tbl_id); diff --git a/src/storage/shard.cpp b/src/storage/shard.cpp index ed01eae8..0c57303c 100644 --- a/src/storage/shard.cpp +++ b/src/storage/shard.cpp @@ -10,6 +10,7 @@ #include #include #include +#include #include #include @@ -21,6 +22,8 @@ #include "error.h" #include "global_registered_memory.h" #include "standby_service.h" +#include "tasks/attach_shard_task.h" +#include "tasks/detach_shard_task.h" #include "tasks/list_object_task.h" #include "tasks/list_standby_partition_task.h" #include "tasks/reopen_task.h" @@ -81,27 +84,15 @@ Shard::Shard(EloqStore *store, size_t shard_id, uint32_t fd_limit) KvError Shard::Init() { - // Inject process term and active branch into IO manager before any file - // operations. All IouringMgr instances support SetActiveBranch; only - // CloudStoreMgr additionally needs SetProcessTerm (local mode uses term=0). + // Inject the active branch into the IO manager before any file operation. + // Terms are per data store shard: each task carries its shard's term and + // the manager reads the running task's (AsyncIoManager::ProcessTerm). if (io_mgr_ != nullptr) { - uint64_t term = store_ != nullptr ? store_->Term() : 0; std::string_view branch = store_ != nullptr ? store_->Branch() : MainBranchName; assert(!branch.empty()); io_mgr_->SetActiveBranch(branch); - if (store_->Mode() == StoreMode::Cloud) - { - auto *cloud_mgr = static_cast(io_mgr_.get()); - cloud_mgr->SetProcessTerm(term); - cloud_mgr->SetPartitionGroupId(store_->PartitionGroupId()); - } - else if (store_->Mode() == StoreMode::StandbyReplica || - store_->Mode() == StoreMode::StandbyMaster) - { - static_cast(io_mgr_.get())->SetProcessTerm(term); - } } InitializeTscFrequency(); KvError res = io_mgr_->Init(this); @@ -323,6 +314,9 @@ void Shard::EnqueueReopenWaiter(KvRequest *req) ReopenRequest *reopen_req = state.DriverRequest(); reopen_req->SetArgs(tbl_id); + // The driver works for the waiter's data store shard, so it is counted + // and gated like the waiter. + reopen_req->SetDsShard(req->DsShard()); reopen_req->SetTag(""); reopen_req->SetClean(false); reopen_req->SetPendingTime(store_->Options().auto_reopen_pending_time_us); @@ -415,6 +409,12 @@ void Shard::AddPendingCompact(const TableIdent &tbl_id) PendingWriteQueue &pending_q = it->second; CompactRequest &req = pending_q.compact_req_; req.SetTableId(tbl_id); + // A write task of this partition calls this (WriteTask::CompactIfNeeded) + // to queue a compaction for the data store shard it runs for. Stamped + // with that shard, the compaction runs in the shard's term and, like the + // shard's other requests, is admitted only while the shard is attached + // and is failed when the shard is detached. + req.SetDsShard(ThdDsShard()); #ifdef ELOQ_MODULE_ENABLED { std::lock_guard lk(req.mutex_); @@ -463,6 +463,10 @@ bool Shard::TryAddFileGc(const TableIdent &tbl_id) } #endif req.SetTableId(tbl_id); + // A write task of this partition calls this (BatchWriteTask::Drop) to + // queue a file GC for the data store shard it runs for; stamped with that + // shard as in AddPendingCompact. + req.SetDsShard(ThdDsShard()); pending_q.PushBack(&req); return true; } @@ -476,6 +480,10 @@ void Shard::AddPendingTTL(const TableIdent &tbl_id) PendingWriteQueue &pending_q = it->second; CleanExpiredRequest &req = pending_q.expire_req_; req.SetTableId(tbl_id); + // A write task of this partition calls this (WriteTask::TriggerTTL) to + // queue a TTL cleanup for the data store shard it runs for; stamped with + // that shard as in AddPendingCompact. + req.SetDsShard(ThdDsShard()); #ifdef ELOQ_MODULE_ENABLED { std::lock_guard lk(req.mutex_); @@ -524,6 +532,10 @@ bool Shard::TryAddLocalGc(const TableIdent &tbl_id) } #endif req.SetTableId(tbl_id); + // A reopen task of this partition calls this (ReopenTask::Reopen) to + // queue a local file GC for the data store shard it runs for; stamped + // with that shard as in AddPendingCompact. + req.SetDsShard(ThdDsShard()); pending_q.PushBack(&req); return true; } @@ -563,6 +575,21 @@ GlobalRegisteredMemory *Shard::GlobalRegMem() void Shard::OnReceivedReq(KvRequest *req) { + if (req->Type() == RequestType::AttachShard || + req->Type() == RequestType::DetachShard) + { + // Internal, from EloqStore::AttachShard or DetachShard: it is for no + // partition, so it bypasses the pending write queues and the + // admission gate, which refuses a shard that is being attached or + // detached. + (void) ProcessReq(req); + return; + } + if (!AdmitRequest(req)) + { + return; + } + if (req->Reopen()) { req->SetReopen(false); @@ -601,19 +628,233 @@ void Shard::TryStartPendingWrite(const TableIdent &tbl_id) return; } PendingWriteQueue &pending_q = it->second; - if (pending_q.running_ || pending_q.Empty()) - { + while (!pending_q.running_ && !pending_q.Empty()) + { + WriteRequest *req = pending_q.PopFront(); + assert(req != nullptr); + // A request queued for a partition whose data store shard has been + // detached or has expired since is failed here, and the next one is + // tried. The embedded internal requests just become done again. + if (!AdmitRequest(req)) + { + continue; + } + pending_q.running_ = true; + if (!ProcessReq(req)) + { + pending_q.running_ = false; + pending_q.PushFront(req); + } return; } + if (!pending_q.running_ && pending_q.Empty()) + { + // Every queued request was refused, so no task will finish on this + // queue and erase it in OnTaskFinished; drop it here. + pending_queues_.erase(it); + } +} - WriteRequest *req = pending_q.PopFront(); - assert(req != nullptr); - pending_q.running_ = true; - if (!ProcessReq(req)) +bool Shard::AdmitRequest(KvRequest *req) +{ + const DsShardId ds_shard = req->DsShard(); + if (ds_shard == kUnsetDsShard) { - pending_q.running_ = false; - pending_q.PushFront(req); + return true; } + KvError err; + switch (store_->ShardState(ds_shard)) + { + case EloqStore::DsShardState::Attached: + return true; + case EloqStore::DsShardState::Expired: + err = KvError::ExpiredTerm; + break; + default: + err = KvError::NotRunning; + break; + } + if (req->Type() == RequestType::Reopen) + { + // The refused request may be the driver of its partition's pending + // reopen, the request that parked requests wait on. Then the waiters + // cannot be served either: take them out, drop the pending-reopen + // state and fail each of them with the same error. + // + // If the driver is the internal one, it is an object embedded in the + // state just dropped and no longer exists, so return without + // completing it. An external driver is the caller's object and is + // completed below like any other refused request. OnTaskFinished + // treats a finished reopen driver the same way. + auto it = pending_reopens_.find(req->TableId()); + if (it != pending_reopens_.end() && it->second.DriverRequest() == req) + { + const bool internal = it->second.IsCurrentInternalDriver( + static_cast(req)); + std::vector waiters = std::move(it->second.waiters_); + pending_reopens_.erase(it); + CompleteReopenWaiters(std::move(waiters), err); + if (internal) + { + return false; + } + } + } + req->SetDone(err); + return false; +} + +void Shard::FailPendingRequestsOf(DsShardId ds_shard_id) +{ + // Pending writes, including the embedded internal requests. A partition + // has one owner at a time, so its queue holds one shard's requests: the + // front one tells. A queue with a running task is left to that task's + // OnTaskFinished, whose dispatch refuses the rest. + std::vector tables; + for (auto &[tbl_id, pending_q] : pending_queues_) + { + if (!pending_q.running_ && !pending_q.Empty() && + pending_q.Front()->DsShard() == ds_shard_id) + { + tables.push_back(tbl_id); + } + } + for (const TableIdent &tbl_id : tables) + { + // TryStartPendingWrite refuses every queued request (AdmitRequest) + // and erases the emptied queue. + TryStartPendingWrite(tbl_id); + } + + // Reopen states whose driver is not running: fail the waiters and drop the + // state. A running driver keeps its state until OnTaskFinished, which + // fails the waiters then (the resubmission is refused by the gate). + std::unordered_set reopen_tables; + for (auto it = pending_reopens_.begin(); it != pending_reopens_.end();) + { + if (it->second.DriverRequest()->DsShard() != ds_shard_id) + { + ++it; + continue; + } + reopen_tables.insert(it->first); + auto queue_it = pending_queues_.find(it->first); + const bool driver_running = + queue_it != pending_queues_.end() && queue_it->second.running_; + std::vector waiters = std::move(it->second.waiters_); + CompleteReopenWaiters(std::move(waiters), KvError::NotRunning); + if (driver_running) + { + ++it; + } + else + { + it = pending_reopens_.erase(it); + } + } + + // Delayed internal drivers of those partitions: their states are gone. + // Matched by table, never through the request pointer: a delayed entry + // can outlive its driver (an external reopen replaces the embedded one, + // and the erase above destroys it), as PromoteReadyDelayedReopenRequests + // allows for too. + const size_t before = delayed_requests_.size(); + std::erase_if(delayed_requests_, + [&](const DelayedEntry &entry) + { return reopen_tables.contains(entry.tbl_id); }); + if (delayed_requests_.size() != before) + { + std::make_heap(delayed_requests_.begin(), + delayed_requests_.end(), + std::greater()); + } +} + +KvError Shard::EvictDsShard(DsShardId ds_shard_id) +{ + // A task that finished after the first pass may have parked a waiter or + // queued an internal request: clear those too, now that nothing runs. + FailPendingRequestsOf(ds_shard_id); + + KvError first_err = KvError::NoError; + auto note = [&first_err](KvError err) + { + if (err != KvError::NoError && first_err == KvError::NoError) + { + first_err = err; + } + }; + // The partitions this worker shard holds state for: those with a root + // meta or an open file. The root meta may have been evicted while files + // stayed open, or the other way round, so both are collected into one set. + std::unordered_set partitions; + index_mgr_.RootMetaManager()->CollectPartitions(partitions); + const bool has_files = !store_->Options().store_path.empty(); + if (has_files) + { + static_cast(io_mgr_.get())->CollectPartitions(partitions); + } + // Of those, drop the ones the partition filter gives the detaching shard. + // Without a filter every partition belongs to every shard; AttachShard + // then hosts only one shard, so all of them are its own. + const auto &filter = store_->Options().partition_filter; + for (const TableIdent &tbl_id : partitions) + { + if (filter && !filter(ds_shard_id, tbl_id)) + { + continue; + } + auto reopen_it = pending_reopens_.find(tbl_id); + if (reopen_it != pending_reopens_.end()) + { + CompleteReopenWaiters(std::move(reopen_it->second.waiters_), + KvError::NotRunning); + pending_reopens_.erase(reopen_it); + } + auto queue_it = pending_queues_.find(tbl_id); + if (queue_it != pending_queues_.end()) + { + CHECK(!queue_it->second.running_ && queue_it->second.Empty()) + << "EvictDsShard with a pending write on " << tbl_id; + pending_queues_.erase(queue_it); + } + if (has_files) + { + auto *io_mgr = static_cast(io_mgr_.get()); + // A partition served only from the cloud, or one that was never + // found, has no local directory: nothing to drop there. + std::error_code ec; + const bool has_dir = fs::is_directory( + tbl_id.StorePath(store_->Options().store_path, + store_->Options().store_path_lut), + ec); + if (ec) + { + // Not "no such directory", which clears ec: the manifest is + // left where it is, and the detach reports it. + LOG(WARNING) << "EvictDsShard: cannot stat the directory of " + << tbl_id << ": " << ec.message(); + note(ToKvError(ec.value())); + } + if (store_->Mode() == StoreMode::Cloud && has_dir) + { + // The manifest may be advanced by the next owner; the sealed + // data files never change and stay cached. + note(static_cast(io_mgr)->DropLocalManifest( + tbl_id)); + } + note(io_mgr->CloseTableFiles(tbl_id)); + } + // No task runs on the partition, so no root meta is pinned. + index_mgr_.RootMetaManager()->Erase(tbl_id); + } + return first_err; +} + +uint32_t Shard::ActiveTasksOf(DsShardId ds_shard_id) const +{ + auto it = active_tasks_by_shard_.find(ds_shard_id); + return it == active_tasks_by_shard_.end() ? 0 : it->second; } void Shard::TryDispatchPendingWrites() @@ -1015,6 +1256,23 @@ bool Shard::ProcessReq(KvRequest *req) req->SetDone(KvError::InvalidArgs); return true; } + case RequestType::AttachShard: + { + AttachShardTask *task = task_mgr_.GetAttachShardTask(); + // StartTask gives the task the request's shard and that shard's term. + auto lbd = [task]() -> KvError { return task->Attach(); }; + StartTask(task, req, lbd); + return true; + } + case RequestType::DetachShard: + { + DetachShardTask *task = task_mgr_.GetDetachShardTask(); + const DsShardId ds_shard = req->DsShard(); + auto lbd = [task, ds_shard]() -> KvError + { return task->Detach(ds_shard); }; + StartTask(task, req, lbd); + return true; + } } return true; } @@ -1037,10 +1295,7 @@ bool Shard::ExecuteReadyTasks() KvTask *task = ready_tasks_.Peek(); ready_tasks_.Dequeue(); assert(task->status_ == TaskStatus::Ongoing); - running_ = task; - // Mark the resume start so cooperative background loops can measure how - // long they have held the worker via CurResumeElapsedUs(). - cur_resume_start_us_ = ReadTimeMicroseconds(); + SwitchToTask(task); task->coro_ = task->coro_.resume(); if (task->status_ == TaskStatus::Finished) { @@ -1058,8 +1313,7 @@ bool Shard::ExecuteReadyTasks() KvTask *task = low_priority_ready_tasks_.Peek(); low_priority_ready_tasks_.Dequeue(); task->status_ = TaskStatus::Ongoing; - running_ = task; - cur_resume_start_us_ = ReadTimeMicroseconds(); + SwitchToTask(task); task->coro_ = task->coro_.resume(); if (task->status_ == TaskStatus::Finished) { @@ -1072,7 +1326,7 @@ bool Shard::ExecuteReadyTasks() } finish: - running_ = nullptr; + LeaveTask(); return busy; } @@ -1140,6 +1394,33 @@ void Shard::OnTaskFinished(KvTask *task) } } + const DsShardId req_shard = task->ds_shard_id_; + if (req_shard != kUnsetDsShard && task_type != TaskType::DetachShard) + { + auto active_it = active_tasks_by_shard_.find(req_shard); + CHECK(active_it != active_tasks_by_shard_.end() && + active_it->second > 0) + << "task finished for data store shard " << req_shard + << " that has no running task"; + if (--active_it->second == 0) + { + active_tasks_by_shard_.erase(active_it); + auto waiting_it = waiting_detaches_.find(req_shard); + if (waiting_it != waiting_detaches_.end()) + { + KvTask *detach_task = waiting_it->second; + waiting_detaches_.erase(waiting_it); + detach_task->Resume(); + } + } + if (err == KvError::ExpiredTerm) + { + // A newer term owns one of the shard's partitions: the whole + // shard is stale (the term file is per shard). + store_->MarkShardExpired(req_shard); + } + } + task->req_ = nullptr; task->result_err_ = KvError::NoError; diff --git a/src/tasks/archive_crond.cpp b/src/tasks/archive_crond.cpp index c7a14e52..236dc468 100644 --- a/src/tasks/archive_crond.cpp +++ b/src/tasks/archive_crond.cpp @@ -6,14 +6,12 @@ #include #include -#include -#include -#include -#include +#include +#include +#include #include #include "eloq_store.h" -#include "utils.h" namespace eloqstore { @@ -25,92 +23,239 @@ ArchiveCrond::ArchiveCrond(EloqStore *store) : store_(store) void ArchiveCrond::Start() { assert(!thd_.joinable()); - stop_requested_ = false; + { + Lock lk(mu_); + stop_requested_ = false; + // The store stopped since the last run: it drained its worker shards, + // so no archive of that run is in flight, and its shards are gone. + schedules_.clear(); + } + exited_.store(false, std::memory_order_relaxed); thd_ = std::thread(&ArchiveCrond::Crond, this); LOG(INFO) << "Archive crond started"; } void ArchiveCrond::Stop() { - mu_.lock(); - stop_requested_ = true; - mu_.unlock(); + { + Lock lk(mu_); + stop_requested_ = true; + } + cond_var_.notify_all(); #ifdef ELOQ_MODULE_ENABLED - while (!stopped_.load(std::memory_order_acquire)) + // The caller may be a bthread on a brpc worker while the thread waits + // for that worker's shard to list partitions: yield, never block. + while (thd_.joinable() && !exited_.load(std::memory_order_acquire)) { - cond_var_.notify_one(); bthread_usleep(1000); } +#endif if (thd_.joinable()) { thd_.join(); } -#else - if (thd_.joinable()) + LOG(INFO) << "Archive crond stopped"; +} + +void ArchiveCrond::AddShard(DsShardId ds_shard_id) +{ { - cond_var_.notify_one(); - thd_.join(); + Lock lk(mu_); + ShardSchedule &schedule = schedules_[ds_shard_id]; + schedule.attached_ = true; + ++schedule.generation_; + schedule.next_due_ = Clock::now() + Interval(); } - LOG(INFO) << "Archive crond stopped"; -#endif + cond_var_.notify_all(); +} + +void ArchiveCrond::RemoveShard(DsShardId ds_shard_id) +{ + Lock lk(mu_); + auto it = schedules_.find(ds_shard_id); + if (it == schedules_.end()) + { + return; + } + ShardSchedule &schedule = it->second; + schedule.attached_ = false; + ++schedule.generation_; + if (!schedule.in_flight_) + { + // No archive reads the request: drop the entry now. One in flight + // keeps it until it completes; the thread drops it then. + schedules_.erase(it); + } +} + +ArchiveCrond::Clock::duration ArchiveCrond::Interval() const +{ + return std::chrono::seconds(store_->Options().archive_interval_secs); } -bool ArchiveCrond::StopRequested() +void ArchiveCrond::Wait(Lock &lk, std::optional until) { - std::scoped_lock lk(mu_); - return stop_requested_; +#ifdef ELOQ_MODULE_ENABLED + // This thread is a plain pthread: the bthread condition variable blocks + // it, as a std one would. + if (!until.has_value()) + { + cond_var_.wait(lk); + return; + } + const auto remaining = + std::chrono::duration_cast(*until - + Clock::now()); + if (remaining.count() > 0) + { + cond_var_.wait_for(lk, remaining.count()); + } +#else + if (!until.has_value()) + { + cond_var_.wait(lk); + return; + } + cond_var_.wait_until(lk, *until); +#endif } void ArchiveCrond::Crond() { - const uint64_t interval_secs = store_->Options().archive_interval_secs; - last_archive_ts_ = utils::UnixTs(); - while (!StopRequested()) + Lock lk(mu_); + while (!stop_requested_) { - // Loop required to prevent spurious wakeups - auto elapsed = utils::UnixTs() - last_archive_ts_; - while (elapsed < interval_secs) + const Clock::time_point now = Clock::now(); + std::vector due; + std::optional earliest; + for (auto it = schedules_.begin(); it != schedules_.end();) { - std::unique_lock lk(mu_); -#ifdef ELOQ_MODULE_ENABLED - for (uint64_t sleeped_secs = 0; - sleeped_secs < interval_secs - elapsed && !stop_requested_; - ++sleeped_secs) + ShardSchedule &schedule = it->second; + if (schedule.in_flight_) { - bthread_usleep(1); + ++it; + continue; } -#else - auto wait_period = chrono::seconds(interval_secs - elapsed); - cond_var_.wait_for( - lk, wait_period, [this] { return stop_requested_; }); -#endif - if (stop_requested_) + if (!schedule.attached_) + { + // Removed while its archive ran, which has completed now. + it = schedules_.erase(it); + continue; + } + if (schedule.next_due_ <= now) + { + schedule.in_flight_ = true; + // The previous request is done: nothing reads it any more. + schedule.req_ = std::make_unique(); + due.push_back( + {it->first, schedule.generation_, schedule.req_.get()}); + } + else if (!earliest.has_value() || schedule.next_due_ < *earliest) { - // Stopped during wait. - return; + earliest = schedule.next_due_; } - elapsed = utils::UnixTs() - last_archive_ts_; + ++it; } - StartArchiving(); - last_archive_ts_ = utils::UnixTs(); + if (!due.empty()) + { + // Submitting may block while an archive lists its partitions, and + // a completion may run inline: neither holds the lock. + lk.unlock(); + for (const DueArchive &archive : due) + { + Submit(archive); + } + lk.lock(); + continue; + } + Wait(lk, earliest); } + lk.unlock(); + exited_.store(true, std::memory_order_release); } -void ArchiveCrond::StartArchiving() +void ArchiveCrond::Submit(const DueArchive &due) { - LOG(INFO) << "Start scheduled archiving"; + const EloqStore::DsShardState state = store_->ShardState(due.ds_shard_id_); + if (state == EloqStore::DsShardState::Detached) + { + // The shard went without RemoveShard: the store stopped, or an attach + // raced its own detach. Its schedule ends here. + Lock lk(mu_); + auto it = schedules_.find(due.ds_shard_id_); + if (it != schedules_.end() && it->second.generation_ == due.generation_) + { + it->second.attached_ = false; + } + if (it != schedules_.end()) + { + it->second.in_flight_ = false; + } + return; + } + if (state != EloqStore::DsShardState::Attached) + { + // Expired, or being detached: try again one interval later. A detach + // ends the schedule; a newer term comes with a new attach. + LOG(WARNING) << "Scheduled archiving of data store shard " + << due.ds_shard_id_ << " put off: the shard is in state " + << static_cast(state); + OnArchiveDone(due.ds_shard_id_, due.generation_, KvError::NotRunning); + return; + } + + LOG(INFO) << "Start scheduled archiving of data store shard " + << due.ds_shard_id_; + due.req_->SetDsShard(due.ds_shard_id_); + const DsShardId ds_shard_id = due.ds_shard_id_; + const uint64_t generation = due.generation_; + const bool sent = store_->ExecAsyn( + due.req_, + 0, + [this, ds_shard_id, generation](KvRequest *req) + { + // The request is not touched after this: the thread may free it + // as soon as OnArchiveDone clears in_flight_. + OnArchiveDone(ds_shard_id, generation, req->Error()); + }); + if (!sent) + { + // Refused before it ran (the store is stopping, or the shard left + // Attached just now): no callback comes. + OnArchiveDone(ds_shard_id, generation, KvError::NotRunning); + } +} - GlobalArchiveRequest global_req; - store_->ExecSync(&global_req); - if (global_req.Error() != KvError::NoError) +void ArchiveCrond::OnArchiveDone(DsShardId ds_shard_id, + uint64_t generation, + KvError err) +{ + if (err == KvError::NoError) + { + LOG(INFO) << "Scheduled archiving of data store shard " << ds_shard_id + << " completed successfully"; + } + else if (err != KvError::NotRunning) { - LOG(ERROR) << "Scheduled archiving failed: " - << static_cast(global_req.Error()); + LOG(ERROR) << "Scheduled archiving of data store shard " << ds_shard_id + << " failed: " << ErrorString(err); } - else { - LOG(INFO) << "Scheduled archiving completed successfully"; + Lock lk(mu_); + auto it = schedules_.find(ds_shard_id); + if (it != schedules_.end() && it->second.in_flight_) + { + ShardSchedule &schedule = it->second; + schedule.in_flight_ = false; + if (schedule.attached_ && schedule.generation_ == generation) + { + // Measured from the end of this archive, as the interval + // always was. + schedule.next_due_ = Clock::now() + Interval(); + } + } } + cond_var_.notify_all(); } } // namespace eloqstore diff --git a/src/tasks/attach_shard_task.cpp b/src/tasks/attach_shard_task.cpp new file mode 100644 index 00000000..31e559f8 --- /dev/null +++ b/src/tasks/attach_shard_task.cpp @@ -0,0 +1,17 @@ +#include "tasks/attach_shard_task.h" + +#include "async_io_manager.h" +#include "eloq_store.h" +#include "storage/shard.h" + +namespace eloqstore +{ +KvError AttachShardTask::Attach() +{ + // Only cloud mode has term files, and AttachShard sends this only then. + CHECK(shard->store_->Mode() == StoreMode::Cloud); + auto *cloud_mgr = static_cast(shard->IoManager()); + return cloud_mgr->UpsertTermFile( + cloud_mgr->GetActiveBranch(), ds_shard_id_, term_); +} +} // namespace eloqstore diff --git a/src/tasks/background_write.cpp b/src/tasks/background_write.cpp index 252cef39..0c92420b 100644 --- a/src/tasks/background_write.cpp +++ b/src/tasks/background_write.cpp @@ -682,9 +682,10 @@ KvError BackgroundWrite::CreateBranch(std::string_view branch_name) root, ttl_root, mapping, new_max_fp_id, dict_bytes, branch_metadata); // The CURRENT_TERM file is NOT created here. It will be created - // when a store instance starts with this branch as its active branch - // (via BootstrapUpsertTermFile in cloud mode). Branch deletion and - // GC already handle missing term files gracefully. + // when a data store shard is attached to a store instance with this + // branch as its active branch (CloudStoreMgr::UpsertTermFile in cloud + // mode). Branch deletion and GC already handle missing term files + // gracefully. return IoMgr()->WriteBranchManifest( tbl_ident_, normalized_branch, 0, snapshot); } diff --git a/src/tasks/detach_shard_task.cpp b/src/tasks/detach_shard_task.cpp new file mode 100644 index 00000000..4bbd9a00 --- /dev/null +++ b/src/tasks/detach_shard_task.cpp @@ -0,0 +1,23 @@ +#include "tasks/detach_shard_task.h" + +#include "storage/shard.h" + +namespace eloqstore +{ +KvError DetachShardTask::Detach(DsShardId ds_shard_id) +{ + // Nothing new is admitted for the shard: the store set it Detaching + // before sending this request. Fail what is queued, then wait for what + // runs. + shard->FailPendingRequestsOf(ds_shard_id); + while (shard->ActiveTasksOf(ds_shard_id) > 0) + { + // The task that brings the count to 0 resumes this one + // (Shard::OnTaskFinished). + shard->waiting_detaches_[ds_shard_id] = this; + status_ = TaskStatus::Blocked; + Yield(); + } + return shard->EvictDsShard(ds_shard_id); +} +} // namespace eloqstore diff --git a/src/tasks/prewarm_task.cpp b/src/tasks/prewarm_task.cpp index 72e876f2..905e9bc3 100644 --- a/src/tasks/prewarm_task.cpp +++ b/src/tasks/prewarm_task.cpp @@ -268,8 +268,7 @@ void PrewarmService::Start() thread_ = std::thread( [this] { - std::string prewarm_all(""); - PrewarmCloudCache(prewarm_all); + // Nothing to warm yet: AttachShard queues each shard's round. PrewarmLoop(); stopped_.store(true, std::memory_order_release); }); @@ -279,21 +278,19 @@ void PrewarmService::PrewarmLoop() { while (!stop_requested_.load(std::memory_order_acquire)) { - std::array prewarm_tables; + std::array jobs; size_t sz; do { - sz = prewarm_tables_.wait_dequeue_bulk_timed( - prewarm_tables.data(), - prewarm_tables.size(), - std::chrono::seconds(1)); + sz = prewarm_jobs_.wait_dequeue_bulk_timed( + jobs.data(), jobs.size(), std::chrono::seconds(1)); for (size_t i = 0; i < sz; ++i) { - const TableIdent &table = prewarm_tables[i]; - std::string remote_path = table.ToString(); - PrewarmCloudCache(remote_path); + const PrewarmJob &job = jobs[i]; + PrewarmCloudCache(job.table_ ? job.table_->ToString() : "", + job.ds_shard_id_); } - } while (sz == prewarm_tables.size()); + } while (sz == jobs.size()); } } @@ -314,9 +311,14 @@ void PrewarmService::Stop() } } -void PrewarmService::Prewarm(const TableIdent &table) +void PrewarmService::Prewarm(DsShardId ds_shard_id, const TableIdent &table) { - prewarm_tables_.enqueue(table); + prewarm_jobs_.enqueue(PrewarmJob{ds_shard_id, table}); +} + +void PrewarmService::PrewarmShard(DsShardId ds_shard_id) +{ + prewarm_jobs_.enqueue(PrewarmJob{ds_shard_id, std::nullopt}); } bool PrewarmService::ListCloudObjects( @@ -364,7 +366,8 @@ bool PrewarmService::ListCloudObjects( return request.Error() == KvError::NoError; } -void PrewarmService::PrewarmCloudCache(const std::string &remote_path) +void PrewarmService::PrewarmCloudCache(const std::string &remote_path, + DsShardId ds_shard_id) { const uint16_t num_threads = std::max(uint16_t{1}, store_->options_.num_threads); @@ -510,7 +513,7 @@ void PrewarmService::PrewarmCloudCache(const std::string &remote_path) continue; } if (store_->options_.partition_filter && - !store_->options_.partition_filter(tbl_id)) + !store_->options_.partition_filter(ds_shard_id, tbl_id)) { total_files_skipped++; continue; diff --git a/src/tasks/reopen_task.cpp b/src/tasks/reopen_task.cpp index 4a44df93..043d0261 100644 --- a/src/tasks/reopen_task.cpp +++ b/src/tasks/reopen_task.cpp @@ -118,7 +118,7 @@ KvError ReopenTask::Reopen(const TableIdent &tbl_id) CHECK(shard->store_ != nullptr); PrewarmService *prewarm_service = shard->store_->GetPrewarmService(); CHECK(prewarm_service != nullptr); - prewarm_service->Prewarm(tbl_id); + prewarm_service->Prewarm(ThdDsShard(), tbl_id); } // TODO(standby): for a non-cloud standby replica, nothing else removes diff --git a/src/tasks/task.cpp b/src/tasks/task.cpp index 1a33cfd8..436828fd 100644 --- a/src/tasks/task.cpp +++ b/src/tasks/task.cpp @@ -603,6 +603,16 @@ KvTask *ThdTask() return shard->running_; } +DsShardId ThdDsShard() +{ + return shard->running_ds_shard_id_; +} + +uint64_t ThdTerm() +{ + return shard->running_term_; +} + void MaybeYield() { // Null-safe defensive guard: only yield when a task is actually running on diff --git a/src/tasks/task_manager.cpp b/src/tasks/task_manager.cpp index 96ef8caf..fdf55d96 100644 --- a/src/tasks/task_manager.cpp +++ b/src/tasks/task_manager.cpp @@ -39,7 +39,11 @@ TaskManager::TaskManager(const KvOptions *opts) 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) + reopen_pool_(reopen_pool_size_default), + // One attach or detach runs per data store shard at a time; the pools + // grow if several shards attach or detach at once. + attach_shard_pool_(1), + detach_shard_pool_(1) { if (opts != nullptr && opts->max_write_concurrency > 0) { @@ -76,6 +80,8 @@ void TaskManager::Shutdown() list_object_pool_.ForEachTask(check_finished); list_standby_partition_pool_.ForEachTask(check_finished); reopen_pool_.ForEachTask(check_finished); + attach_shard_pool_.ForEachTask(check_finished); + detach_shard_pool_.ForEachTask(check_finished); CHECK_EQ(unfinished_tasks, 0) << "TaskManager::Shutdown requires a quiesced shard, unfinished_tasks=" << unfinished_tasks << ", unfinished_reqs=" << unfinished_reqs; @@ -93,6 +99,8 @@ void TaskManager::Shutdown() list_object_pool_.Clear(); list_standby_partition_pool_.Clear(); reopen_pool_.Clear(); + attach_shard_pool_.Clear(); + detach_shard_pool_.Clear(); } void TaskManager::SetPoolSizesForTest(uint32_t batch_write_pool_size, @@ -176,6 +184,18 @@ ReopenTask *TaskManager::GetReopenTask(const TableIdent &tbl_id) return task; } +AttachShardTask *TaskManager::GetAttachShardTask() +{ + num_active_++; + return attach_shard_pool_.GetTask(); +} + +DetachShardTask *TaskManager::GetDetachShardTask() +{ + num_active_++; + return detach_shard_pool_.GetTask(); +} + void TaskManager::FreeTask(KvTask *task) { CHECK(task->status_ == TaskStatus::Finished); @@ -212,6 +232,12 @@ void TaskManager::FreeTask(KvTask *task) static_cast(task)->ReleaseDirBusy(); reopen_pool_.FreeTask(static_cast(task)); break; + case TaskType::AttachShard: + attach_shard_pool_.FreeTask(static_cast(task)); + break; + case TaskType::DetachShard: + detach_shard_pool_.FreeTask(static_cast(task)); + break; case TaskType::EvictFile: assert(false && "EvictFile task should not be freed here"); break; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index 7262e7e9..0bb1a88e 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -27,6 +27,7 @@ set(UTEST_SOURCES concurrency.cpp cloud.cpp cloud_term.cpp + ds_shards.cpp gc.cpp chore.cpp eloq_store_test.cpp diff --git a/tests/branch_operations.cpp b/tests/branch_operations.cpp index 03c23820..46b06587 100644 --- a/tests/branch_operations.cpp +++ b/tests/branch_operations.cpp @@ -372,8 +372,9 @@ TEST_CASE("delete branch removes all term manifests", "[branch]") // Simulate the branch having been written to at higher terms (e.g. after a // failover). Write placeholder manifests for terms 1–3 and create // CURRENT_TERM_feature_0 with value "3" (branch creation does not write - // the term file; it is normally created at store startup via - // BootstrapUpsertTermFile, so we create it manually here). + // the term file; it is normally created when a data store shard is + // attached, via CloudStoreMgr::UpsertTermFile, so we create it manually + // here). // DeleteBranchFiles reads CURRENT_TERM to discover max_term, then unlinks // manifests 0..max_term; it never reads the manifest contents, so // placeholder content is fine. diff --git a/tests/cloud.cpp b/tests/cloud.cpp index 05ef8f19..d49df78a 100644 --- a/tests/cloud.cpp +++ b/tests/cloud.cpp @@ -663,8 +663,9 @@ TEST_CASE("cloud prewarm honors partition filter", "[cloud][prewarm]") partitions[0], partitions[1], }; - options.partition_filter = - [included](const eloqstore::TableIdent &tbl) -> bool + options.partition_filter = [included]( + eloqstore::DsShardId, + const eloqstore::TableIdent &tbl) -> bool { return included.count(tbl) != 0; }; eloqstore::EloqStore *store = InitStore(options); @@ -1178,8 +1179,9 @@ TEST_CASE("cloud global archive shares timestamp and filters partitions", first_partition_id, second_page_partition_id, }; - options.partition_filter = - [included_ids](const eloqstore::TableIdent &tbl) -> bool + options.partition_filter = [included_ids]( + eloqstore::DsShardId, + const eloqstore::TableIdent &tbl) -> bool { return included_ids.count(tbl.partition_id_) != 0; }; eloqstore::EloqStore *store = InitStore(options); @@ -1580,7 +1582,7 @@ TEST_CASE("cloud reopen triggers prewarm to download newer remote data files", options.prewarm_cloud_cache = true; options.prewarm_task_count = 1; options.allow_reuse_local_caches = true; - options.partition_filter = [&enable_partition_filter](const auto &) + options.partition_filter = [&enable_partition_filter](auto, const auto &) { return enable_partition_filter.load(std::memory_order_relaxed); }; CleanupStore(options); diff --git a/tests/ds_shards.cpp b/tests/ds_shards.cpp new file mode 100644 index 00000000..89321493 --- /dev/null +++ b/tests/ds_shards.cpp @@ -0,0 +1,614 @@ +/** + * @file ds_shards.cpp + * @brief One store hosting several data store shards: AttachShard, + * DetachShard, per-shard terms and term files, the admission gate, the + * detach drain, and store-level requests confined to their shard's + * partitions. + */ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "common.h" +#include "eloq_store.h" + +namespace +{ +using eloqstore::DsShardId; +using eloqstore::EloqStore; +using eloqstore::KvError; +using eloqstore::TableIdent; + +const TableIdent kPartitionA{"ds_shard_tbl", 1}; +const TableIdent kPartitionB{"ds_shard_tbl", 2}; + +/** @brief Options for a cloud store under its own cloud and local paths. */ +eloqstore::KvOptions ShardTestOptions(const std::string &local_path) +{ + eloqstore::KvOptions options = cloud_options; + options.store_path = {local_path}; + options.cloud_store_path += "/ds-shards"; + options.allow_reuse_local_caches = true; + // Partition N belongs to data store shard N: kPartitionA to shard 1, + // kPartitionB to shard 2. + options.partition_filter = + [](DsShardId ds_shard_id, const TableIdent &tbl_id) + { return static_cast(tbl_id.partition_id_) == ds_shard_id; }; + return options; +} + +/** @brief Upserts one key for a data store shard; returns the error. */ +KvError Write(EloqStore *store, + DsShardId ds_shard_id, + const TableIdent &tbl_id, + const std::string &key, + const std::string &value) +{ + eloqstore::BatchWriteRequest req; + req.SetDsShard(ds_shard_id); + req.SetTableId(tbl_id); + req.AddWrite(key, value, 1, eloqstore::WriteOp::Upsert); + store->ExecSync(&req); + return req.Error(); +} + +/** @brief Reads one key for a data store shard; returns the error. */ +KvError Read(EloqStore *store, + DsShardId ds_shard_id, + const TableIdent &tbl_id, + const std::string &key, + std::string *value = nullptr) +{ + eloqstore::ReadRequest req; + req.SetDsShard(ds_shard_id); + req.SetArgs(tbl_id, key); + store->ExecSync(&req); + if (value != nullptr && req.Error() == KvError::NoError) + { + *value = req.value_; + } + return req.Error(); +} + +bool Contains(const std::vector &names, const std::string &name) +{ + return std::find(names.begin(), names.end(), name) != names.end(); +} + +/** + * @brief Stops and joins a test's worker thread on every exit path, so that a + * failed REQUIRE unwinds past the thread instead of terminating the process + * on a joinable one. Declare it after the store, so it goes first. + */ +struct ThreadJoiner +{ + std::thread &thread_; + std::atomic *stop_; + ~ThreadJoiner() + { + if (stop_ != nullptr) + { + stop_->store(true); + } + if (thread_.joinable()) + { + thread_.join(); + } + } +}; + +/** @brief "k" followed by @p i zero-padded to three digits. */ +std::string PaddedKey(int i) +{ + std::string digits = std::to_string(i); + return "k" + std::string(3 - digits.size(), '0') + digits; +} + +/** + * @brief Counts a partition's archives of one term in the cloud: the files + * named manifest_main__, which the manifest itself, + * manifest_main_, does not match. + * @param options The store's options, for the cloud path. + * @param tbl_id The partition. + * @param term The term the archives are named with. + * @return How many there are. + */ +size_t CountArchives(const eloqstore::KvOptions &options, + const TableIdent &tbl_id, + uint64_t term) +{ + const std::string prefix = "manifest_main_" + std::to_string(term) + "_"; + size_t count = 0; + for (const std::string &name : + ListCloudFiles(options, options.cloud_store_path, tbl_id.ToString())) + { + if (name.rfind(prefix, 0) == 0) + { + ++count; + } + } + return count; +} + +/** + * @brief Polls a condition every 200 ms until it holds or a timeout passes. + * @param timeout How long to wait at most. + * @param pred The condition. + * @return Whether it held. + */ +template +bool WaitUntil(std::chrono::milliseconds timeout, Pred &&pred) +{ + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) + { + if (pred()) + { + return true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + } + return pred(); +} +} // namespace + +TEST_CASE("two data store shards are served in their own terms", + "[cloud][ds_shard]") +{ + eloqstore::KvOptions options = ShardTestOptions("/tmp/test-ds-shards"); + CleanupStore(options); + + { + // Hosting a second shard needs the partition filter. + eloqstore::KvOptions unfiltered = options; + unfiltered.partition_filter = nullptr; + EloqStore store(unfiltered); + REQUIRE(store.StartInstance() == KvError::NoError); + REQUIRE(store.AttachShard(1, 5) == KvError::NoError); + REQUIRE(store.AttachShard(2, 9) == KvError::InvalidArgs); + REQUIRE(store.HostedShards() == std::vector{1}); + store.Stop(); + } + + EloqStore store(options); + REQUIRE(store.StartInstance() == KvError::NoError); + REQUIRE(store.HostedShards().empty()); + // Nothing is attached: no request is admitted. + REQUIRE(Write(&store, 1, kPartitionA, "k", "v") == KvError::NotRunning); + + REQUIRE(store.AttachShard(1, 5) == KvError::NoError); + REQUIRE(store.AttachShard(2, 9) == KvError::NoError); + REQUIRE(store.HostedShards() == std::vector{1, 2}); + REQUIRE(store.ShardTerm(1) == 5); + REQUIRE(store.ShardTerm(2) == 9); + // Attaching again, or an id out of range, is refused. + REQUIRE(store.AttachShard(1, 6) == KvError::InvalidArgs); + REQUIRE(store.AttachShard(EloqStore::kMaxDsShards, 1) == + KvError::InvalidArgs); + + REQUIRE(Write(&store, 1, kPartitionA, "a", "va") == KvError::NoError); + REQUIRE(Write(&store, 2, kPartitionB, "b", "vb") == KvError::NoError); + std::string value; + REQUIRE(Read(&store, 1, kPartitionA, "a", &value) == KvError::NoError); + REQUIRE(value == "va"); + REQUIRE(Read(&store, 2, kPartitionB, "b", &value) == KvError::NoError); + REQUIRE(value == "vb"); + // A request naming no shard works for the first one attached. + REQUIRE(store.DefaultDsShard() == 1); + REQUIRE(Read(&store, eloqstore::kUnsetDsShard, kPartitionA, "a") == + KvError::NoError); + + // Each shard has its own term file, and each partition's manifest is + // named with its shard's term. + const auto root_files = ListCloudFiles(options, options.cloud_store_path); + REQUIRE(Contains(root_files, "CURRENT_TERM_main_1")); + REQUIRE(Contains(root_files, "CURRENT_TERM_main_2")); + REQUIRE( + Contains(ListCloudFiles( + options, options.cloud_store_path, kPartitionA.ToString()), + "manifest_main_5")); + REQUIRE( + Contains(ListCloudFiles( + options, options.cloud_store_path, kPartitionB.ToString()), + "manifest_main_9")); + + // Detaching one shard refuses its requests and leaves the other alone. + REQUIRE(store.DetachShard(1) == KvError::NoError); + REQUIRE(store.HostedShards() == std::vector{2}); + REQUIRE(store.ShardState(1) == EloqStore::DsShardState::Detached); + REQUIRE(Read(&store, 1, kPartitionA, "a") == KvError::NotRunning); + REQUIRE(Read(&store, 2, kPartitionB, "b", &value) == KvError::NoError); + REQUIRE(value == "vb"); + REQUIRE(store.DetachShard(1) == KvError::NoError); + + // Attached again at a newer term, the shard's partition is taken over: + // its data is there, and its manifest is promoted to the new term. + REQUIRE(store.AttachShard(1, 6) == KvError::NoError); + REQUIRE(Read(&store, 1, kPartitionA, "a", &value) == KvError::NoError); + REQUIRE(value == "va"); + REQUIRE(Write(&store, 1, kPartitionA, "a2", "va2") == KvError::NoError); + REQUIRE( + Contains(ListCloudFiles( + options, options.cloud_store_path, kPartitionA.ToString()), + "manifest_main_6")); + + store.Stop(); + REQUIRE(store.HostedShards().empty()); + CleanupStore(options); +} + +TEST_CASE("an expired term fails one data store shard only", + "[cloud][ds_shard]") +{ + eloqstore::KvOptions first_options = + ShardTestOptions("/tmp/test-ds-shards-first"); + eloqstore::KvOptions second_options = + ShardTestOptions("/tmp/test-ds-shards-second"); + CleanupStore(first_options); + CleanupLocalStore(second_options); + + // The first node hosts shards 1 and 2 at term 5. + { + EloqStore first(first_options); + REQUIRE(first.StartInstance() == KvError::NoError); + REQUIRE(first.AttachShard(1, 5) == KvError::NoError); + REQUIRE(first.AttachShard(2, 5) == KvError::NoError); + REQUIRE(Write(&first, 1, kPartitionA, "a", "va") == KvError::NoError); + REQUIRE(Write(&first, 2, kPartitionB, "b", "vb") == KvError::NoError); + first.Stop(); + } + + // A second node takes shard 1 over at term 7 and writes to it. + { + EloqStore second(second_options); + REQUIRE(second.StartInstance() == KvError::NoError); + REQUIRE(second.AttachShard(1, 7) == KvError::NoError); + std::string value; + REQUIRE(Read(&second, 1, kPartitionA, "a", &value) == KvError::NoError); + REQUIRE(value == "va"); + REQUIRE(Write(&second, 1, kPartitionA, "a7", "va7") == + KvError::NoError); + second.Stop(); + } + + // Back on the first node, shard 1 at term 5 is stale; shard 2 is not. + { + EloqStore first(first_options); + REQUIRE(first.StartInstance() == KvError::NoError); + REQUIRE(first.AttachShard(2, 5) == KvError::NoError); + REQUIRE(first.AttachShard(1, 5) == KvError::ExpiredTerm); + REQUIRE(first.HostedShards() == std::vector{2}); + REQUIRE(first.ShardState(1) == EloqStore::DsShardState::Detached); + REQUIRE(Read(&first, 1, kPartitionA, "a") == KvError::NotRunning); + std::string value; + REQUIRE(Read(&first, 2, kPartitionB, "b", &value) == KvError::NoError); + REQUIRE(value == "vb"); + // At a term above the second node's, shard 1 can be taken back. + REQUIRE(first.AttachShard(1, 8) == KvError::NoError); + REQUIRE(Read(&first, 1, kPartitionA, "a7", &value) == KvError::NoError); + REQUIRE(value == "va7"); + first.Stop(); + } + + CleanupStore(first_options); + CleanupLocalStore(second_options); +} + +TEST_CASE("detaching a data store shard drains its running requests", + "[cloud][ds_shard]") +{ + eloqstore::KvOptions options = + ShardTestOptions("/tmp/test-ds-shards-drain"); + CleanupStore(options); + + EloqStore store(options); + REQUIRE(store.StartInstance() == KvError::NoError); + REQUIRE(store.AttachShard(1, 5) == KvError::NoError); + REQUIRE(store.AttachShard(2, 5) == KvError::NoError); + for (int i = 0; i < 200; ++i) + { + REQUIRE(Write(&store, 1, kPartitionA, "k" + std::to_string(i), "v") == + KvError::NoError); + } + REQUIRE(Write(&store, 2, kPartitionB, "b", "vb") == KvError::NoError); + + // Reads of shard 1 keep arriving until the gate refuses one, which the + // detach below makes it do. Every one of them completes, either served + // or refused; none hangs. + constexpr int kMaxReads = 20000; + std::vector> reads; + reads.reserve(kMaxReads); + std::atomic submitted{0}; + std::atomic completed{0}; + std::atomic served{0}; + std::atomic refused{0}; + std::atomic other{0}; + auto on_done = [&](eloqstore::KvRequest *req) + { + switch (req->Error()) + { + case KvError::NoError: + ++served; + break; + case KvError::NotRunning: + ++refused; + break; + default: + ++other; + break; + } + ++completed; + }; + std::thread reader( + [&] + { + for (int i = 0; i < kMaxReads; ++i) + { + auto req = std::make_unique(); + req->SetDsShard(1); + req->SetArgs(kPartitionA, "k" + std::to_string(i % 200)); + ++submitted; + const bool accepted = store.ExecAsyn(req.get(), 0, on_done); + reads.push_back(std::move(req)); + if (!accepted) + { + // Refused at the gate: nothing will call back, and the + // detach has begun. + ++refused; + ++completed; + break; + } + } + }); + ThreadJoiner join_reader{reader, nullptr}; + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + REQUIRE(store.DetachShard(1) == KvError::NoError); + reader.join(); + const int total = submitted.load(); + const auto deadline = + std::chrono::steady_clock::now() + std::chrono::seconds(30); + while (completed.load() < total && + std::chrono::steady_clock::now() < deadline) + { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + REQUIRE(completed.load() == total); + REQUIRE(other.load() == 0); + REQUIRE(served.load() + refused.load() == total); + REQUIRE(refused.load() > 0); + + // The other shard was never disturbed, and the detached one comes back. + std::string value; + REQUIRE(Read(&store, 2, kPartitionB, "b", &value) == KvError::NoError); + REQUIRE(value == "vb"); + REQUIRE(store.AttachShard(1, 6) == KvError::NoError); + REQUIRE(Read(&store, 1, kPartitionA, "k0", &value) == KvError::NoError); + REQUIRE(value == "v"); + + store.Stop(); + CleanupStore(options); +} + +TEST_CASE("detaching a data store shard waits for its background work", + "[cloud][ds_shard]") +{ + // Overwriting the same keys round after round exceeds the file + // amplification, so the write tasks keep queuing compactions, and the + // sealed files queue file GC. That work is queued by shard 1's tasks and + // runs as shard 1's own, in shard 1's term: a detach fails what has not + // started and waits for what runs, instead of pulling the partition from + // under it, and nothing runs outside the shard's term. + eloqstore::KvOptions options = ShardTestOptions("/tmp/test-ds-shards-bg"); + CleanupStore(options); + + EloqStore store(options); + REQUIRE(store.StartInstance() == KvError::NoError); + REQUIRE(store.AttachShard(1, 5) == KvError::NoError); + REQUIRE(store.AttachShard(2, 5) == KvError::NoError); + REQUIRE(Write(&store, 2, kPartitionB, "b", "vb") == KvError::NoError); + + constexpr int kKeys = 200; + std::atomic stop{false}; + std::atomic served{0}; + std::atomic refused{0}; + std::atomic other{0}; + std::thread writer( + [&] + { + for (int round = 0; !stop.load(); ++round) + { + eloqstore::BatchWriteRequest req; + req.SetDsShard(1); + req.SetTableId(kPartitionA); + const std::string value(1000, + static_cast('a' + round % 26)); + for (int i = 0; i < kKeys; ++i) + { + // A batch's keys are sorted: zero-padded, so numeric + // order is key order. + req.AddWrite(PaddedKey(i), + value, + static_cast(round + 1), + eloqstore::WriteOp::Upsert); + } + store.ExecSync(&req); + switch (req.Error()) + { + case KvError::NoError: + ++served; + break; + case KvError::NotRunning: + ++refused; + // Do not spin while the shard is detached. + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + break; + default: + ++other; + break; + } + } + }); + + ThreadJoiner join_writer{writer, &stop}; + + // Detach and re-attach the shard under the storm, each time at a newer + // term: each detach meets writes, compactions and file GC queued or in + // flight. + uint64_t term = 5; + for (int cycle = 0; cycle < 5; ++cycle) + { + std::this_thread::sleep_for( + std::chrono::milliseconds(cycle == 0 ? 1000 : 400)); + REQUIRE(store.DetachShard(1) == KvError::NoError); + REQUIRE(Read(&store, 1, kPartitionA, PaddedKey(0)) == + KvError::NotRunning); + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + REQUIRE(store.AttachShard(1, ++term) == KvError::NoError); + } + stop.store(true); + writer.join(); + REQUIRE(other.load() == 0); + REQUIRE(served.load() > 0); + REQUIRE(refused.load() > 0); + + // The other shard was never disturbed, and the partition holds what its + // last served round wrote. + std::string value; + REQUIRE(Read(&store, 2, kPartitionB, "b", &value) == KvError::NoError); + REQUIRE(value == "vb"); + REQUIRE(Read(&store, 1, kPartitionA, PaddedKey(0), &value) == + KvError::NoError); + REQUIRE(value.size() == 1000); + REQUIRE(Read(&store, 1, kPartitionA, PaddedKey(kKeys - 1), &value) == + KvError::NoError); + REQUIRE(value.size() == 1000); + + // Every file the storm and its background work put in the cloud is named + // with one of the terms the shard held: the data and segment files, and + // the manifests, end in their term. + const auto files = ListCloudFiles( + options, options.cloud_store_path, kPartitionA.ToString()); + REQUIRE(!files.empty()); + for (const std::string &name : files) + { + const size_t sep = name.rfind('_'); + REQUIRE(sep != std::string::npos); + const uint64_t file_term = std::stoull(name.substr(sep + 1)); + INFO(name); + REQUIRE(file_term >= 5); + REQUIRE(file_term <= term); + } + + store.Stop(); + CleanupStore(options); +} + +TEST_CASE("each data store shard is archived on its own schedule", + "[cloud][ds_shard][archive]") +{ + eloqstore::KvOptions options = + ShardTestOptions("/tmp/test-ds-shards-archive"); + options.num_retained_archives = 100; + options.archive_interval_secs = 1; + CleanupStore(options); + const std::chrono::seconds timeout(20); + + EloqStore store(options); + REQUIRE(store.StartInstance() == KvError::NoError); + REQUIRE(store.AttachShard(1, 5) == KvError::NoError); + REQUIRE(store.AttachShard(2, 9) == KvError::NoError); + REQUIRE(Write(&store, 1, kPartitionA, "a", "va") == KvError::NoError); + REQUIRE(Write(&store, 2, kPartitionB, "b", "vb") == KvError::NoError); + + // Each shard's archives are named with its own term and cover only its + // own partitions. + REQUIRE(WaitUntil(timeout, + [&] + { + return CountArchives(options, kPartitionA, 5) > 0 && + CountArchives(options, kPartitionB, 9) > 0; + })); + REQUIRE(CountArchives(options, kPartitionA, 9) == 0); + REQUIRE(CountArchives(options, kPartitionB, 5) == 0); + + // A detach ends the shard's schedule; the other shard's goes on. The + // detach drains an archive of the shard in flight, so nothing of shard 1 + // lands after it returns. + REQUIRE(store.DetachShard(1) == KvError::NoError); + const size_t a_archives = CountArchives(options, kPartitionA, 5); + const size_t b_archives = CountArchives(options, kPartitionB, 9); + REQUIRE(WaitUntil( + timeout, + [&] + { return CountArchives(options, kPartitionB, 9) >= b_archives + 2; })); + REQUIRE(CountArchives(options, kPartitionA, 5) == a_archives); + + // Attached again at a newer term, the shard starts a fresh schedule, and + // its archives carry the new term. + REQUIRE(store.AttachShard(1, 6) == KvError::NoError); + REQUIRE(Write(&store, 1, kPartitionA, "a6", "va6") == KvError::NoError); + REQUIRE(WaitUntil( + timeout, [&] { return CountArchives(options, kPartitionA, 6) > 0; })); + + store.Stop(); + CleanupStore(options); +} + +TEST_CASE("dropping a table drops its data store shard's partitions only", + "[cloud][ds_shard]") +{ + eloqstore::KvOptions options = ShardTestOptions("/tmp/test-ds-shards-drop"); + CleanupStore(options); + + EloqStore store(options); + REQUIRE(store.StartInstance() == KvError::NoError); + REQUIRE(store.AttachShard(1, 5) == KvError::NoError); + REQUIRE(store.AttachShard(2, 5) == KvError::NoError); + REQUIRE(Write(&store, 1, kPartitionA, "a", "va") == KvError::NoError); + REQUIRE(Write(&store, 2, kPartitionB, "b", "vb") == KvError::NoError); + + // The table has a partition in each shard. Dropped for shard 1, it loses + // shard 1's partition only. + eloqstore::DropTableRequest drop; + drop.SetDsShard(1); + drop.SetArgs(kPartitionA.tbl_name_); + store.ExecSync(&drop); + REQUIRE(drop.Error() == KvError::NoError); + REQUIRE(Read(&store, 1, kPartitionA, "a") == KvError::NotFound); + std::string value; + REQUIRE(Read(&store, 2, kPartitionB, "b", &value) == KvError::NoError); + REQUIRE(value == "vb"); + + store.Stop(); + CleanupStore(options); +} + +TEST_CASE("only cloud mode hosts a second data store shard", "[ds_shard]") +{ + // Start attaches shard 0 with the default options' term 0. + EloqStore *store = InitStore(default_opts); + REQUIRE(store->HostedShards() == std::vector{0}); + REQUIRE(store->DefaultDsShard() == 0); + REQUIRE(store->AttachShard(1, 0) == KvError::InvalidArgs); + REQUIRE(Write(store, 0, test_tbl_id, "a", "va") == KvError::NoError); + + REQUIRE(store->DetachShard(0) == KvError::NoError); + REQUIRE(store->HostedShards().empty()); + REQUIRE(Read(store, 0, test_tbl_id, "a") == KvError::NotRunning); + // With the default shard gone, a request naming none is refused too. + REQUIRE(Read(store, eloqstore::kUnsetDsShard, test_tbl_id, "a") == + KvError::NotRunning); + + REQUIRE(store->AttachShard(0, 0) == KvError::NoError); + std::string value; + REQUIRE(Read(store, 0, test_tbl_id, "a", &value) == KvError::NoError); + REQUIRE(value == "va"); + store->Stop(); + CleanupStore(default_opts); +} diff --git a/tests/manifest.cpp b/tests/manifest.cpp index 9d42eb20..31e9ab45 100644 --- a/tests/manifest.cpp +++ b/tests/manifest.cpp @@ -197,8 +197,9 @@ TEST_CASE("global archive shares timestamp and filters partitions", partitions[0].partition_id_, partitions[2].partition_id_, }; - opts.partition_filter = - [included_ids](const eloqstore::TableIdent &tbl) -> bool + opts.partition_filter = [included_ids]( + eloqstore::DsShardId, + const eloqstore::TableIdent &tbl) -> bool { return included_ids.count(tbl.partition_id_) != 0; }; eloqstore::EloqStore *store = InitStore(opts);