From 3da4cd6e91f5472d33c8344d960ea39beec22514 Mon Sep 17 00:00:00 2001 From: sami Date: Thu, 10 Sep 2026 15:50:18 +0400 Subject: [PATCH] core: rate/token_bucket + fold in DAEMON's engine-API review (stage 7) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit rate/token_bucket.hpp — a lazily-refilled TokenBucket (starts full: burst then throttle, IDM behaviour; rate 0 == unlimited; burst caps idle accumulation) and RateLimiter, the global -> per-queue -> per-task hierarchy (docs/04 §6). acquire(task, n) peeks every applicable level and commits on all-or-none so a blocked attempt never leaks tokens at a level that had them; held under one mutex so a concurrent detach can't dangle the bucket it's using. vdm/ids.hpp gains QueueId. Tests: burst/refill/cap/unlimited for the bucket; tightest-level-binds, no-partial-consumption, detach-safety, and an 8-thread aggregate-rate check for the hierarchy. Green under ASan/UBSan and TSan. Engine-API review (DAEMON signed off, no sched/ or dispatch rewrite): - Engine::rate_limiter() accessor added (limiter.set -> set_global_limit). - Checksum::Algo gains sha512 to match the wire Checksum set. - DownloadSpec: DAEMON creates save_path's parent dir before start(); missing dir -> Error::path_rejected (made explicit). - cancel(): documented to fire on_state(_, cancelled, nullopt) then on_finished(Err{canceled}), in that order; download.cancel == cancel(false), download.remove == cancel(true). - engine-api-m1.md: the five open questions resolved with DAEMON's answers (probe_hint optional, single cancel flag, {restart, keep_partial, abort} is the whole set, per-task 4 Hz is fine, refresh_url restarts all segments after a validating re-probe). Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01HPPSGhiArbvQgwC2DNiURS --- core/docs/engine-api-m1.md | 72 +++++++--- core/include/vdm/engine.hpp | 7 + core/include/vdm/ids.hpp | 16 +++ core/include/vdm/rate/token_bucket.hpp | 186 +++++++++++++++++++++++++ core/include/vdm/task/download.hpp | 13 +- core/tests/CMakeLists.txt | 1 + core/tests/rate/token_bucket_test.cpp | 145 +++++++++++++++++++ 7 files changed, 414 insertions(+), 26 deletions(-) create mode 100644 core/include/vdm/rate/token_bucket.hpp create mode 100644 core/tests/rate/token_bucket_test.cpp diff --git a/core/docs/engine-api-m1.md b/core/docs/engine-api-m1.md index b321b82..85ebb4a 100644 --- a/core/docs/engine-api-m1.md +++ b/core/docs/engine-api-m1.md @@ -1,6 +1,18 @@ -# CORE → DAEMON — the engine API for review (M1) +# CORE → DAEMON — the engine API (M1) -**Status: for review.** This is the `libveloxcore` public surface `veloxd` links and calls +**Status: reviewed and signed off** by DAEMON on 2026-09-10 +(`daemon/docs/engine-api-review.md`). Nothing forced a `sched/` or dispatch rewrite. The +five open questions are resolved at the bottom; the review's four confirms are folded into +the headers (`sha512` added to `Checksum::Algo`; the parent-directory contract made +explicit; `cancel()` ordering documented; `rate_limiter()` accessor added). + +Integration order: DAEMON wires this in after `daemon/src/sched/` lands (the scheduler is +what calls `start`/`pause`/`resume`/`cancel` and drives `set_task_order`); `sched/` builds +against these headers in parallel with CORE stage 8 and does not need the bodies. + +--- + +This is the `libveloxcore` public surface `veloxd` links and calls to actually run a download. It is the thing the AGENT-CORE brief asked for on day one and that slipped: DAEMON has an RPC surface and a store and, until this is agreed, nothing to call. Sketch headers: `core/include/vdm/engine.hpp`, `core/include/vdm/task/download.hpp` @@ -109,9 +121,15 @@ task waiting on credentials. - **`decide(Decision)`** — acts only while auto-paused for a decision. `restart` discards the partial and re-downloads; `keep_partial` continues against what is on disk (the user's stated risk); `abort` → `failed`. No-op otherwise. -- **`refresh_url(url, headers)`** — IDM's "Refresh Download Address": swap the URL on a - live or paused task without losing progress (a fresh signed URL). Maps to - `download.refreshUrl`. +- **`refresh_url(url, headers)`** — IDM's "Refresh Download Address": the engine re-probes + the new URL to validate it, then **restarts every segment on it** (the signed-URL-expiry + case the wire method exists for), keeping the bytes already on disk. Mirror rotation is + *not* this — that is `spec.mirrors` + the segmenter's requeue-to-a-different-host. +- **`on_decision_needed`** is only for the cases the engine cannot resolve itself. A + routine 416 / stale range is the engine's own re-probe + re-split loop; it escalates + `DecisionRequest{range_metadata_stale}` **only when that loop fails**, at which point + "retry the same range" is already exhausted — so `{restart, keep_partial, abort}` is the + whole choice set. ## 5. Progress @@ -138,21 +156,31 @@ source until the stream ends. --- -## Open questions for DAEMON +## Resolved (DAEMON review, 2026-09-10) -1. **`DownloadSpec.probe_hint`** — do you want to pass the File-Info `ProbeResult` in, or - would you rather the engine always probe (one code path, ~1 extra round trip)? The - sketch supports both; picking one simplifies stage 8. -2. **`cancel(discard_partial)`** vs a separate `handle.remove()` — the wire has - `download.cancel` and `download.remove {deleteFile}` as two methods. One call with a - flag, or two? -3. **`on_decision_needed` granularity** — is `{restart, keep_partial, abort}` the right - choice set for `server_file_changed`, or do you also need "retry the same range once - more" as a distinct option for the stale-416 case? -4. **Progress cadence** — 4 Hz per task matches the wire. With 20 active tasks that is 80 - callbacks/s on the engine's timer thread. Acceptable, or do you want a single - `on_progress_batch(span)` so the engine coalesces across tasks too? -5. **`refresh_url` while `downloading`** — should an in-flight segment finish on the old - URL and only new/retried segments use the new one (less disruption), or should the - engine restart all segments on the new URL immediately (guaranteed consistency)? The - signed-URL-expiry case wants the latter; a mirror swap wants the former. +1. **`probe_hint` stays optional.** DAEMON has a `ProbeResult` only on the File-Info path; + capture-take, `velox add`, `addBatch` and restart have none. The engine probes when + it's absent. +2. **One `cancel(discard_partial)`.** `download.cancel` = `cancel(false)`; + `download.remove` = `cancel(true)` for a live task (plus DAEMON's row/file cleanup), or + pure DAEMON-side for an already-terminal one. No `handle.remove()`. +3. **`{restart, keep_partial, abort}` is the whole set.** The engine owns the routine + 416 re-probe/re-split and only escalates `on_decision_needed` when that loop fails — + "retry the same range" is exhausted by then. +4. **Per-task 4 Hz is fine.** DAEMON coalesces across tasks for `event.task.progress` + regardless; `on_progress_batch` is a nice-to-have and must not block stage 8. +5. **`refresh_url` restarts all segments on the new URL** after a validating re-probe (the + signed-URL case). Mirror rotation is `spec.mirrors` + the segmenter, not this. + +## Review confirms, folded in + +- **(a)** `vdm::TaskId` is a cheap-copy hashable value; DAEMON never constructs one — it + only receives it from `start()` / callbacks and passes it back to `set_task_order()` + etc. ✔ (`vdm/ids.hpp`) +- **(b)** DAEMON `mkdir -p`s `save_path`'s parent before `start()`. The engine opens the + file and fails with `Error::path_rejected` if the directory is missing. ✔ (documented on + `DownloadSpec`) +- **(c)** `Checksum::Algo` now has `sha512`, matching the wire `Checksum` set. ✔ +- **(d)** `cancel()` always fires `on_state(_, cancelled, nullopt)` then + `on_finished(Err{Error::canceled})` (note: the taxonomy value is `canceled`), in that + order. ✔ (documented on `DownloadHandle::cancel`) diff --git a/core/include/vdm/engine.hpp b/core/include/vdm/engine.hpp index 85bdf80..fdc3628 100644 --- a/core/include/vdm/engine.hpp +++ b/core/include/vdm/engine.hpp @@ -13,6 +13,7 @@ #include #include +#include "vdm/rate/token_bucket.hpp" #include "vdm/segment/budget.hpp" #include "vdm/task/download.hpp" @@ -53,6 +54,12 @@ class Engine { // on_budget_changed). See ADR 0011. [[nodiscard]] segment::SegmentBudget &segment_budget() noexcept; + // The hierarchical speed limiter (docs/04 §6): global -> per-queue -> per-task token + // buckets. `limiter.set {globalBps, enabled}` -> rate_limiter().set_global_limit(); + // per-queue / per-task limits and the task<->queue attachment come from DAEMON too. + // The engine paces every segment read through it. + [[nodiscard]] rate::RateLimiter &rate_limiter() noexcept; + // Live settings (connection.* changes from settings.set). Each affects future work. void set_default_segments(std::uint32_t n); void set_default_buffer_bytes(std::uint64_t bytes); diff --git a/core/include/vdm/ids.hpp b/core/include/vdm/ids.hpp index 914b4a4..9304ba8 100644 --- a/core/include/vdm/ids.hpp +++ b/core/include/vdm/ids.hpp @@ -24,6 +24,15 @@ struct TaskId { friend constexpr auto operator<=>(const TaskId &, const TaskId &) = default; }; +// A scheduler queue. A task may belong to one — for the per-queue rate limit and +// per-queue concurrency. A task with no queue rate-limits against the global bucket only. +struct QueueId { + std::uint64_t value = 0; + + [[nodiscard]] constexpr bool valid() const noexcept { return value != 0; } + friend constexpr auto operator<=>(const QueueId &, const QueueId &) = default; +}; + using SteadyTime = std::chrono::steady_clock::time_point; } // namespace vdm @@ -35,4 +44,11 @@ struct std::hash { } }; +template <> +struct std::hash { + std::size_t operator()(vdm::QueueId id) const noexcept { + return std::hash{}(id.value); + } +}; + #endif // VDM_IDS_HPP diff --git a/core/include/vdm/rate/token_bucket.hpp b/core/include/vdm/rate/token_bucket.hpp new file mode 100644 index 0000000..49c7100 --- /dev/null +++ b/core/include/vdm/rate/token_bucket.hpp @@ -0,0 +1,186 @@ +// vdm/rate/token_bucket.hpp — a lazily-refilled token bucket, and the global -> queue -> +// task limiter hierarchy built on it (docs/04 §6). +// +// A segment worker calls RateLimiter::acquire(task, n) after receiving n body bytes. If +// every applicable level (task, its queue, global) has n tokens, it consumes n from each +// and returns 0. Otherwise it consumes nothing and returns how long to wait before +// retrying — the worker returns CURL_WRITEFUNC_PAUSE and schedules a curl_easy_pause +// resume after that delay (the "precision" layer on top of CURLOPT_MAX_RECV_SPEED_LARGE). +// +// This header compiles standalone. + +#ifndef VDM_RATE_TOKEN_BUCKET_HPP +#define VDM_RATE_TOKEN_BUCKET_HPP + +#include +#include +#include +#include +#include +#include + +#include "vdm/ids.hpp" + +namespace vdm::rate { + +// rate_bps == 0 means unlimited: consume() always succeeds and never waits. +class TokenBucket { + public: + TokenBucket() = default; + // `burst` caps how many tokens accumulate while idle; 0 => 1 second's worth. + explicit TokenBucket(std::uint64_t rate_bps, std::uint64_t burst = 0) { + set_rate(rate_bps, burst); + } + + void set_rate(std::uint64_t rate_bps, std::uint64_t burst = 0) { + std::lock_guard lk(mu_); + const bool was_unlimited = rate_ == 0; + rate_ = rate_bps; + cap_ = burst ? burst : rate_bps; // 1 s of burst by default + // A freshly-limited bucket starts full: you may transfer at burst speed + // immediately, then it throttles (classic token bucket / IDM behaviour). Lowering + // an existing limit only clamps down — it never hands out a fresh burst. + if (rate_bps > 0 && was_unlimited) + tokens_ = cap_; + else if (tokens_ > cap_) + tokens_ = cap_; + last_ = clock::now(); + } + [[nodiscard]] std::uint64_t rate() const { + std::lock_guard lk(mu_); + return rate_; + } + + // Consume `n` if available; otherwise consume nothing. Returns the wait until `n` + // tokens *would* be available (0 when it consumed). + [[nodiscard]] std::chrono::nanoseconds consume(std::uint64_t n) { + std::lock_guard lk(mu_); + if (rate_ == 0) + return {}; + refill_locked(); + if (tokens_ >= n) { + tokens_ -= n; + return {}; + } + const std::uint64_t deficit = n - tokens_; + // ns to earn `deficit` tokens at rate_ bytes/s + return std::chrono::nanoseconds{ + static_cast((deficit * 1'000'000'000ull + rate_ - 1) / rate_)}; + } + + // Two-phase for the hierarchy: check every level, then commit on all or none. + [[nodiscard]] std::chrono::nanoseconds peek(std::uint64_t n) { + std::lock_guard lk(mu_); + if (rate_ == 0) + return {}; + refill_locked(); + if (tokens_ >= n) + return {}; + const std::uint64_t deficit = n - tokens_; + return std::chrono::nanoseconds{ + static_cast((deficit * 1'000'000'000ull + rate_ - 1) / rate_)}; + } + void commit(std::uint64_t n) { + std::lock_guard lk(mu_); + if (rate_ == 0) + return; + tokens_ = tokens_ >= n ? tokens_ - n : 0; + } + + private: + using clock = std::chrono::steady_clock; + void refill_locked() { + auto now = clock::now(); + auto dt = std::chrono::duration_cast(now - last_).count(); + if (dt <= 0) + return; + last_ = now; + // added = rate_ * dt / 1e9, guarding overflow for very long idle gaps + long double added = + static_cast(rate_) * static_cast(dt) / 1'000'000'000.0L; + std::uint64_t add = + added >= static_cast(cap_) ? cap_ : static_cast(added); + tokens_ = tokens_ + add > cap_ ? cap_ : tokens_ + add; + } + + mutable std::mutex mu_; + std::uint64_t rate_ = 0; + std::uint64_t cap_ = 0; + std::uint64_t tokens_ = 0; + clock::time_point last_ = clock::now(); +}; + +// The hierarchy. All limits default to 0 (unlimited). A task with no queue is limited by +// task + global only. +class RateLimiter { + public: + void set_global_limit(std::uint64_t bps) { global_.set_rate(bps); } + [[nodiscard]] std::uint64_t global_limit() const { return global_.rate(); } + + void set_queue_limit(QueueId q, std::uint64_t bps) { + std::lock_guard lk(mu_); + queues_[q].set_rate(bps); + } + void set_task_limit(TaskId t, std::uint64_t bps) { + std::lock_guard lk(mu_); + tasks_[t].set_rate(bps); + } + + void attach_task(TaskId t, std::optional q) { + std::lock_guard lk(mu_); + tasks_.try_emplace(t); + if (q) { + task_queue_[t] = *q; + queues_.try_emplace(*q); + } else { + task_queue_.erase(t); + } + } + void detach_task(TaskId t) { + std::lock_guard lk(mu_); + tasks_.erase(t); + task_queue_.erase(t); + } + + // Consume `n` bytes against task, queue and global. 0 => consumed everywhere. > 0 => + // consumed nowhere; wait that long and retry. Held under mu_ for its whole duration + // so a concurrent detach_task() can't invalidate the bucket it is using. + [[nodiscard]] std::chrono::nanoseconds acquire(TaskId t, std::uint64_t n) { + std::lock_guard lk(mu_); + TokenBucket *tb = nullptr; + TokenBucket *qb = nullptr; + if (auto it = tasks_.find(t); it != tasks_.end()) + tb = &it->second; + if (auto qit = task_queue_.find(t); qit != task_queue_.end()) + if (auto q = queues_.find(qit->second); q != queues_.end()) + qb = &q->second; + + // peek all, then commit all or none — no level "leaks" tokens on a partial miss. + std::chrono::nanoseconds wait{}; + if (tb) + wait = std::max(wait, tb->peek(n)); + if (qb) + wait = std::max(wait, qb->peek(n)); + wait = std::max(wait, global_.peek(n)); + if (wait.count() > 0) + return wait; + + if (tb) + tb->commit(n); + if (qb) + qb->commit(n); + global_.commit(n); + return {}; + } + + private: + mutable std::mutex mu_; // guards the maps AND serialises acquire() + TokenBucket global_; + std::unordered_map queues_; + std::unordered_map tasks_; + std::unordered_map task_queue_; +}; + +} // namespace vdm::rate + +#endif // VDM_RATE_TOKEN_BUCKET_HPP diff --git a/core/include/vdm/task/download.hpp b/core/include/vdm/task/download.hpp index 6416b5b..4eb184b 100644 --- a/core/include/vdm/task/download.hpp +++ b/core/include/vdm/task/download.hpp @@ -34,15 +34,17 @@ namespace vdm::task { // --- input --------------------------------------------------------------------------- struct Checksum { - enum class Algo { md5, sha1, sha256 }; + enum class Algo { md5, sha1, sha256, sha512 }; // matches the wire Checksum set Algo algo = Algo::sha256; std::string hex; // lower-case, no separators }; // Everything the engine needs to run ONE download. DAEMON has already run the rules -// engine, canonicalised the path and checked it against the allowed roots, and resolved -// the filename — `save_path` is absolute and final. `.veloxpart` and -// `.veloxpart.meta` live beside it during the transfer. +// engine, canonicalised the path, checked it against the allowed roots, resolved the +// filename, and created the parent directory — `save_path` is absolute and final and its +// directory exists. `.veloxpart` and `.veloxpart.meta` live beside +// it during the transfer; on success the part file is renamed in place. If the directory +// is missing at open time the task fails with Error::path_rejected. struct DownloadSpec { std::string url; std::vector mirrors; // alternative URLs for the same bytes @@ -185,6 +187,9 @@ class DownloadHandle { void pause(); void resume(); // Idempotent, terminal. discard_partial also removes the .veloxpart[.meta] files. + // Always fires on_state(_, cancelled, nullopt) then on_finished(Err{Error::canceled}), + // in that order. `download.cancel` == cancel(false); `download.remove` == cancel(true) + // (plus DAEMON's own row/file cleanup). void cancel(bool discard_partial = false); // Only act while the task is awaiting the matching input (auto-paused for auth / diff --git a/core/tests/CMakeLists.txt b/core/tests/CMakeLists.txt index 2c6ac5e..ca11711 100644 --- a/core/tests/CMakeLists.txt +++ b/core/tests/CMakeLists.txt @@ -31,6 +31,7 @@ vdm_add_test(veloxcore_veloxpart_test meta/veloxpart_test.cpp) vdm_add_test(veloxcore_segmenter_test segment/segmenter_test.cpp) vdm_add_test(veloxcore_budget_test segment/budget_test.cpp) vdm_add_test(veloxcore_engine_api_test task/api_compiles_test.cpp) +vdm_add_test(veloxcore_token_bucket_test rate/token_bucket_test.cpp) set(_testserver ${CMAKE_SOURCE_DIR}/tools/testserver/testserver.py) foreach(net_it http_client probe) diff --git a/core/tests/rate/token_bucket_test.cpp b/core/tests/rate/token_bucket_test.cpp new file mode 100644 index 0000000..a0cf443 --- /dev/null +++ b/core/tests/rate/token_bucket_test.cpp @@ -0,0 +1,145 @@ +#include "vdm/rate/token_bucket.hpp" + +#include +#include +#include +#include + +#include "vtest.hpp" + +using namespace vdm; +using namespace vdm::rate; +using namespace std::chrono_literals; + +namespace { +TaskId tid(std::uint64_t v) { + return TaskId{v}; +} +QueueId qid(std::uint64_t v) { + return QueueId{v}; +} +} // namespace + +VT_TEST(tb_unlimited_never_waits) { + TokenBucket b(0); + for (int i = 0; i < 1000; ++i) + VT_CHECK_EQ(b.consume(1'000'000).count(), 0); +} + +VT_TEST(tb_burst_then_throttle) { + // 1000 B/s, default burst = 1 s = 1000 tokens. + TokenBucket b(1000); + VT_CHECK_EQ(b.consume(1000).count(), 0); // drains the burst + auto w = b.consume(1000); // empty now: must wait ~1 s + VT_CHECK(w >= 900ms && w <= 1100ms); +} + +VT_TEST(tb_refills_over_time) { + TokenBucket b(10'000, /*burst=*/10'000); + VT_CHECK_EQ(b.consume(10'000).count(), 0); + std::this_thread::sleep_for(120ms); // ~1200 tokens back + auto w = b.consume(1000); + VT_CHECK_EQ(w.count(), 0); // affordable from the refill + auto w2 = b.consume(5000); + VT_CHECK(w2.count() > 0); // not that much yet +} + +VT_TEST(tb_burst_caps_accumulation) { + TokenBucket b(1000, /*burst=*/2000); + std::this_thread::sleep_for(100ms); // idle far longer than burst/rate would fill + std::this_thread::sleep_for(100ms); + VT_CHECK_EQ(b.consume(2000).count(), 0); // at most the 2000 cap accumulated + VT_CHECK(b.consume(1).count() > 0); // and no more +} + +VT_TEST(tb_set_rate_zero_makes_unlimited) { + TokenBucket b(1000); + VT_CHECK_EQ(b.consume(1000).count(), 0); + VT_CHECK(b.consume(1000).count() > 0); + b.set_rate(0); + VT_CHECK_EQ(b.consume(1'000'000).count(), 0); +} + +// --- the hierarchy -------------------------------------------------------------------- + +VT_TEST(rl_all_unlimited_by_default) { + RateLimiter rl; + rl.attach_task(tid(1), std::nullopt); + for (int i = 0; i < 100; ++i) + VT_CHECK_EQ(rl.acquire(tid(1), 1'000'000).count(), 0); +} + +VT_TEST(rl_tightest_level_binds) { + RateLimiter rl; + rl.set_global_limit(100'000); + rl.set_queue_limit(qid(9), 20'000); + rl.set_task_limit(tid(1), 50'000); + rl.attach_task(tid(1), qid(9)); + + // burst: task 50k, queue 20k, global 100k -> the queue's 20k is the ceiling + VT_CHECK_EQ(rl.acquire(tid(1), 20'000).count(), 0); + auto w = rl.acquire(tid(1), 5'000); + VT_CHECK(w.count() > 0); // queue bucket is dry even though task & global aren't +} + +VT_TEST(rl_no_partial_consumption_on_miss) { + RateLimiter rl; + rl.set_global_limit(1'000'000); // plenty + rl.set_task_limit(tid(1), 1000); // 1 s burst + rl.attach_task(tid(1), std::nullopt); + + VT_CHECK_EQ(rl.acquire(tid(1), 1000).count(), 0); // drain the task bucket + for (int i = 0; i < 5; ++i) + VT_CHECK(rl.acquire(tid(1), 1000).count() > 0); // task bucket blocks, repeatedly + + // global must NOT have been charged for any of those blocked attempts: a fresh task + // limited only by the global bucket can still spend nearly its whole burst (only the + // one *successful* 1000-byte acquire above was charged). + rl.attach_task(tid(2), std::nullopt); + VT_CHECK_EQ(rl.acquire(tid(2), 990'000).count(), 0); +} + +VT_TEST(rl_detach_then_acquire_is_safe_and_unlimited) { + RateLimiter rl; + rl.set_task_limit(tid(1), 1000); + rl.attach_task(tid(1), std::nullopt); + VT_CHECK_EQ(rl.acquire(tid(1), 1000).count(), 0); + rl.detach_task(tid(1)); + // unknown task -> no task/queue bucket, only global (unlimited here) + VT_CHECK_EQ(rl.acquire(tid(1), 1'000'000).count(), 0); +} + +VT_TEST(rl_enforces_aggregate_rate_under_load) { + RateLimiter rl; + const std::uint64_t rate = 4'000'000; // 4 MB/s global + rl.set_global_limit(rate); + for (std::uint64_t i = 1; i <= 8; ++i) + rl.attach_task(tid(i), std::nullopt); + + std::atomic moved{0}; + auto t0 = std::chrono::steady_clock::now(); + std::vector ws; + for (std::uint64_t i = 1; i <= 8; ++i) { + ws.emplace_back([&, id = tid(i)] { + for (int k = 0; k < 400; ++k) { + std::uint64_t chunk = 16 * 1024; + for (;;) { + auto w = rl.acquire(id, chunk); + if (w.count() == 0) + break; + std::this_thread::sleep_for( + std::min(w, std::chrono::milliseconds(20))); + } + moved.fetch_add(chunk); + } + }); + } + ws.clear(); // join + auto secs = std::chrono::duration(std::chrono::steady_clock::now() - t0).count(); + + double effective = moved.load() / secs; + // Allow one burst's worth of slop plus scheduling noise: effective rate should sit + // within ~2x of the configured limit, never wildly above. + VT_CHECK(effective <= rate * 2.5); + VT_CHECK(moved.load() == 8u * 400u * 16u * 1024u); +}