diff --git a/CHANGELOG.md b/CHANGELOG.md index e55f0485..f6741bca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,27 @@ API surface). ### Changed +- **`LocalBackend::execute` no longer rescans the pending-completion list on + every dispatch.** `trackPending` used to `std::erase_if` the whole `_pending` + vector before each append, so admitting one call with *n* already in flight + cost *n* atomic `weak_ptr::expired()` loads under `_pendingMtx`, and a burst of + *n* cost O(n²) — before any model work started. Measured against one parked + model (clang 22.1.8, `-O2`, 8-core Linux), timing only the `execute()` calls: + 32,000 queued executes spent **362 ms** in admission alone, at **24.1 µs** per + admission over the last tenth; 16,000 spent 77.8 ms. The sweep is now + amortised — it runs only when the list reaches a threshold re-armed at twice + the surviving entry count after each sweep — which brings the same case to + **7.6 ms** total and **0.31 µs** per admission, flat in *n*, and within noise of + the 7.6 ms measured with the sweep deleted outright. + + The list is now bounded at twice the live count rather than exactly it, which + is the whole price; a new `LocalBackend::trackedPendingCount()` makes that + observable. `cancelPending` is unaffected and still fails every live + completion on a backend swap or `~Bridge`: it never saw dead entries in the + first place, because `weak.lock()` has always skipped them. See + `docs/spec/core/backend.md`, "The pending list and its amortised compaction", + and morph#528. + - **`Quantity::equation()` writes out at most 100 derivation steps by default, and takes the limit as an argument.** A derivation has no bound — a data-driven `total = total + row` loop records one step per iteration — and diff --git a/docs/spec/core/backend.md b/docs/spec/core/backend.md index 4c7ccbdc..0853aa4b 100644 --- a/docs/spec/core/backend.md +++ b/docs/spec/core/backend.md @@ -517,7 +517,8 @@ there, rather than once per backend. `beginSpan`/`endSpan` around `localOp` — see [observability.md](observability.md). Both `registerModel` and `deregisterModel` emit `registerCount`/`deregisterCount`. - `cancelPending` — snapshots the pending list under the pending mutex, delivers - `exc` to every still-live state. + `exc` to every still-live state, and re-arms the compaction threshold. See + [The pending list and its amortised compaction](#the-pending-list-and-its-amortised-compaction). - `notifyBackendChanged` — under `_regMtx`, looks up only the models recorded in `_changeAware` (populated at registration from `IModelHolder::isBackendChangeAware()` — a compile-time answer per model type, @@ -533,6 +534,58 @@ there, rather than once per backend. Each model instance gets its own strand so actions are serialised per-model without a global lock on the pool. +### The pending list and its amortised compaction + +`_pending` is a `vector>>>` guarded by +`_pendingMtx`. It exists for exactly one reader — `cancelPending`, which swaps it +out and fails everything still live on a backend swap or a `~Bridge`. Nothing +else consults it, and nothing unlinks from it when a completion settles: an entry +simply becomes a dead `weak_ptr` that `cancelPending`'s `weak.lock()` skips. + +Dead entries therefore have to be reclaimed by a sweep, and the question is only +how often. Sweeping on **every** `execute`, as `trackPending` did before +morph#528, makes admitting one call cost one atomic `weak_ptr::expired()` load +per entry already in the list, under the mutex, before any work starts — so a +burst of *n* costs O(n²). Measured against one parked model on an 8-core Linux +box (clang 22.1.8, `-O2`), timing only the `execute()` calls themselves: + +| queued executes | total admission time | mean per admission | mean over the last 10% | +|---|---|---|---| +| 1 000 | 0.59 ms | 0.59 µs | 0.82 µs | +| 4 000 | 5.01 ms | 1.25 µs | 2.19 µs | +| 16 000 | 77.8 ms | 4.86 µs | 9.57 µs | +| 32 000 | 362 ms | 11.3 µs | 24.1 µs | + +Total time quadruples per doubling of *n* and the per-admission cost doubles — +the O(n²)/O(n) pair, not an artefact of some constant. + +The sweep is now **amortised**: `trackPending` sweeps only when `_pending.size()` +reaches `_compactAt`, and each sweep re-arms `_compactAt` at twice the number of +entries that survived it (floor 32, below which sweeping costs more than it +reclaims). A sweep costs O(size) and at least `_compactAt / 2` appends must +happen before the next one, so admission is amortised O(1) at any depth. On the +same benchmark the 32 000-execute case drops from 362 ms to 7.6 ms — within noise +of the 7.6 ms measured with the sweep deleted outright, so what is left is the +`make_shared`, the registry lookup and the strand post, not the sweep. + +Two properties are the price and the guarantee: + +- **Memory.** The list is bounded at twice the live count plus the floor, rather + than at exactly the live count. `trackedPendingCount()` makes that observable; + the fixture in `tests/test_backend_extra.cpp` drives 3 072 settled-and-dropped + admissions past 48 parked ones and measures 112 entries left, against the 3 120 + an uncompacted list would hold. +- **`cancelPending` is unchanged.** It never saw dead entries in the first place + — `weak.lock()` has always skipped them — so carrying them for longer changes + nothing it observes. Every live state admitted across every sweep is still + reached, which is what the same fixture's second assertion checks. `cancelPending` + also resets `_compactAt` to the floor, since it has just emptied the list. + +What is **not** guarded by a test is the admission latency itself: a wall-clock +assertion on a shared CI runner would be a flake rather than evidence, so the +numbers above come from a benchmark and the test guards only the bound and the +cancellation. + ## `RemoteServer` — server-side message handler `RemoteServer` receives JSON envelopes (`morph::wire::Envelope`) from any @@ -1953,7 +2006,8 @@ inside the class calls `close()` — no thread it joins can be waiting on it. | `deregisterModel(mid)` | Releases one attachment through `_instances` under `_regMtx`; erases from `_changeAware` when that destroys the instance. | | `notifyBackendChanged()` | Looks up the models recorded in `_changeAware` under `_regMtx`, then posts `onBackendChanged()` (the `IModelHolder` base virtual — no `dynamic_cast`) onto each such model's strand (outside the lock). Cost is O(change-aware models). | | `execute(mid, call, cbExec)` | Posts `call.localOp` on the model's strand with `ScopedContext`. Returns a `Completion`. | -| `cancelPending(exc)` | Snapshots `_pending`, delivers `exc` to each live state. | +| `cancelPending(exc)` | Snapshots `_pending`, delivers `exc` to each live state, and re-arms the compaction threshold. | +| `trackedPendingCount()` | `[[nodiscard]] std::size_t trackedPendingCount() const` — size of `_pending` under `_pendingMtx`. **Not** the in-flight count: between sweeps the list also holds entries whose state is already destroyed. An upper bound on in-flight, and the observable that makes [the compaction policy](#the-pending-list-and-its-amortised-compaction)'s memory cost measurable. For in-flight *calls*, use `Bridge::pendingCalls()`. | ### `RemoteServer` @@ -2097,6 +2151,7 @@ not a behavior change to the existing loopback-only default. | `handleInline` | Synchronous; caller-restricted to control messages | Safe to call from a worker-pool thread (e.g. from a `BridgeHandler` constructor). It is meant for `register`/`deregister` only; an `execute` envelope is rejected with an `err` reply, because `dispatchExecute` posts to the strand and would reply after `handleInline` returns (writing into an already-destroyed reply buffer). The rejection is now enforced by the code, matching the documented intent. | | `SimulatedRemoteBackend` factory ignored | Model construction delegated to `RemoteServer`'s `ModelRegistryFactory` | The factory closure lives on the client side; the server owns the actual instances. | | `cancelPending` snapshots | Weak-ptr snapshot under lock, then resolves outside | Avoids holding the lock while delivering exceptions to each state, preventing deadlock if a callback re-enters the backend. | +| `_pending` compacted amortised, not intrusively | Sweep when `size() >= _compactAt`, re-arm at twice the survivors | The alternative considered in morph#528 was intrusive: give `CompletionState` a slot index and unlink on settle, making both registration and removal O(1) with no sweep at all. Rejected. It pushes a back-reference to the backend's table into a type shared by every backend, and puts a `_pendingMtx` acquisition on the settle path of every completion — turning a cost paid once per burst into contention paid by every strand thread on every result, on the exact path morph#579's value-handling contract just fixed. The amortised sweep buys the same O(1) admission for one `size_t` of state confined to `LocalBackend`, at the cost of a list bounded at 2× the live count instead of exactly it. | | `setReconnectHandler` | Default no-op | Only backends with a transport layer (e.g. `QtWebSocketBackend`) need to react to reconnects. `LocalBackend` and `SimulatedRemoteBackend` never invoke it. | | `setConnectHandler`/`setDisconnectHandler` on `IBackend`, not only `QtWebSocketBackend` | Same no-op-default pattern as `setReconnectHandler` | Connection state is a property of any transport-backed backend; a UI observing it shouldn't have to downcast to a concrete backend type. A purely local backend has no meaningful connection state, so the base-class hook is simply inert for it — no behavior change, matching the existing `setReconnectHandler` precedent exactly. | | `setDisconnectHandler` fires before reconnect scheduling | Ordering choice, not incidental | An instant successful reconnect must not look, from an observer's perspective, like nothing happened — the disconnected state must be visible even when the very next thing that happens is a fresh `connected`. | diff --git a/include/morph/core/backend.hpp b/include/morph/core/backend.hpp index c6032cee..5eaf4a17 100644 --- a/include/morph/core/backend.hpp +++ b/include/morph/core/backend.hpp @@ -1,9 +1,11 @@ // SPDX-License-Identifier: Apache-2.0 #pragma once +#include #include #include #include +#include #include #include #include @@ -1345,6 +1347,7 @@ class LocalBackend : public detail::IBackend { { std::scoped_lock const lock{_pendingMtx}; snapshot.swap(_pending); + _compactAt = kPendingCompactFloor; } for (auto& weak : snapshot) { if (auto state = weak.lock()) { @@ -1353,6 +1356,20 @@ class LocalBackend : public detail::IBackend { } } + /// @brief Number of entries currently held in the pending list. + /// + /// **Not** the number of calls in flight: the list is compacted amortised + /// (see `trackPending`), so it also carries entries whose completion has + /// already been destroyed and which `cancelPending` would skip. It is an + /// upper bound on the in-flight count and a direct measure of what the + /// compaction policy costs in memory — which is what it exists to make + /// observable. For a count of in-flight *calls*, use `Bridge::pendingCalls()`. + /// @return Size of the pending list, live and dead entries alike. + [[nodiscard]] std::size_t trackedPendingCount() const { + std::scoped_lock const lock{_pendingMtx}; + return _pending.size(); + } + private: /// @brief Builds a holder via @p factory, records it under a fresh id, and /// returns that id. Caller holds `_regMtx`. @@ -1389,9 +1406,34 @@ class LocalBackend : public detail::IBackend { return {mid, std::move(holder)}; } + /// @brief Records @p state in the pending list, compacting it amortised-O(1). + /// + /// The list is append-only between compactions; dead entries are swept only + /// when its size reaches `_compactAt`, which each sweep re-arms at twice the + /// number of entries that survived it. Doubling the threshold off the live + /// count is what makes the sweep amortised: a sweep costs O(size), and at + /// least `_compactAt / 2` appends must happen before the next one, so the + /// per-append cost is O(1) however deep the queue gets. The list is + /// correspondingly bounded at twice the live count (plus the floor), which + /// is the whole price of dropping the per-dispatch scan. + /// + /// Before morph#528 this swept on *every* append, so admitting one call with + /// `n` in flight cost `n` atomic `weak_ptr::expired()` loads under + /// `_pendingMtx` and a burst of `n` cost O(n²) — measured at 362ms of pure + /// admission time for 32k queued executes against one slow model, against + /// 7.6ms without the sweep. + /// + /// Purely a cost change: `cancelPending` still sees every live state, + /// because a dead entry is one whose `CompletionState` is already gone and + /// which `cancelPending`'s `weak.lock()` has always skipped. Carrying dead + /// entries for longer changes nothing it observes. + /// @param state Completion state to track until it expires or is cancelled. void trackPending(const std::shared_ptr<::morph::async::detail::CompletionState>>& state) { std::scoped_lock const lock{_pendingMtx}; - std::erase_if(_pending, [](const auto& weak) { return weak.expired(); }); + if (_pending.size() >= _compactAt) { + std::erase_if(_pending, [](const auto& weak) { return weak.expired(); }); + _compactAt = std::max(kPendingCompactFloor, _pending.size() * 2); + } _pending.emplace_back(state); } @@ -1414,8 +1456,16 @@ class LocalBackend : public detail::IBackend { // directory is the state the two backends genuinely share. std::unordered_set<::morph::exec::detail::ModelId, ::morph::exec::detail::ModelIdHash> _changeAware; std::atomic _nextId{0}; - std::mutex _pendingMtx; + // Smallest size at which `trackPending` will sweep. Below it the sweep costs + // more than the handful of dead `weak_ptr`s it could reclaim, and a backend + // that only ever has a few calls in flight never sweeps at all. + static constexpr std::size_t kPendingCompactFloor = 32; + mutable std::mutex _pendingMtx; std::vector>>> _pending; + // Size at which `trackPending` next sweeps `_pending` for expired entries; + // re-armed at twice the surviving count after each sweep. Guarded by + // `_pendingMtx` along with `_pending` itself. See `trackPending` (morph#528). + std::size_t _compactAt = kPendingCompactFloor; // Concurrent in-flight executes, for the executeInFlight metric. A // shared_ptr (not a plain atomic member) so strand tasks hold their own // reference instead of capturing `this` — see execute()'s comment and the diff --git a/scripts/branch_partial_allowlist.json b/scripts/branch_partial_allowlist.json index fd448a51..38b0257e 100644 --- a/scripts/branch_partial_allowlist.json +++ b/scripts/branch_partial_allowlist.json @@ -80,7 +80,7 @@ }, { "file": "include/morph/core/backend.hpp", - "line": 1228, + "line": 1230, "source": "if (const auto* inst = _instances.find(modelId)) {", "reason": "Unreachable by construction given the `_changeAware`/`_instances` invariant (core audit finding BK2). `_changeAware` is an index over the instance directory: an id enters it in `createHolder` (this file, when the holder answers `isBackendChangeAware()`) in the same `_regMtx`-held critical section that files the instance, and leaves it in `deregisterModel` only when `InstanceDirectory::release` reports the instance actually destroyed. `notifyBackendChanged()` (this function) holds the same `_regMtx` while walking `_changeAware` and looking each id up at this line, so every id it walks is still live -- the null arm cannot occur without a code change that breaks that subset invariant. Formerly keyed on `_models`, the map morph#523 replaced with the directory; the invariant and its reason are unchanged." }, diff --git a/tests/test_backend_extra.cpp b/tests/test_backend_extra.cpp index 10d8bbf5..d5859524 100644 --- a/tests/test_backend_extra.cpp +++ b/tests/test_backend_extra.cpp @@ -3,6 +3,9 @@ #include #include #include +#include +#include +#include #include #include #include @@ -299,3 +302,126 @@ TEST_CASE("morph::backend::LocalBackend: one execute produces exactly one beginS REQUIRE(beginCalls.load() == 1); REQUIRE(endCalls.load() == 1); } + +// ── morph::backend::LocalBackend: amortised pending-list compaction (morph#528) ──────────────── + +namespace { + +/// Builds an `ActionCall` whose local op is @p op — the two morph#528 cases +/// below differ only in that op. +morph::backend::detail::ActionCall pendingCall(std::function op) { + morph::backend::detail::ActionCall call; + call.modelTypeId = "BE_CounterModel"; + call.actionTypeId = "BE_CounterAction"; + call.localOp = [op = std::move(op)](morph::model::detail::IModelHolder&) -> std::shared_ptr { + op(); + return {}; + }; + return call; +} + +} // namespace + +// Two properties of `trackPending`'s amortised sweep, in one fixture because +// they are the two halves of the same trade: the sweep must run often enough to +// bound the list, and must never take a live entry with it. +// +// 1. **The list stays bounded.** Compaction reclaims dead entries, so a backend +// that has admitted thousands of since-settled calls does not carry +// thousands of dead `weak_ptr`s. Mutating `trackPending` to never sweep +// (drop the `_pending.size() >= _compactAt` branch) leaves ~3k entries +// against the bound asserted below, and fails here. +// 2. **`cancelPending` still reaches every live completion.** Dead entries +// linger between sweeps, so this is the property the optimisation could +// plausibly break. Mutating the sweep predicate to `true` (erase +// everything, not just the expired) drops the parked completions and fails +// here. +// +// What this does *not* assert is admission latency — the reason the sweep was +// made amortised in the first place. That is measured by a benchmark, not by a +// test; a wall-clock assertion on a shared CI runner would be a flake, not +// evidence. The morph#528 numbers are recorded in docs/spec/core/backend.md. +TEST_CASE("morph::backend::LocalBackend: amortised pending compaction bounds the list and keeps cancelPending whole", + "[backend][local][pending]") { + constexpr int kRounds = 48; + constexpr int kChurnPerRound = 64; // 3072 admissions that settle and are dropped + + // Declared *before* the pool and the backend, so they outlive them. Only the + // one parked task that is actually running has left the strand queue when + // this scope ends; the other 47 are still queued, and `~StrandExecutor` / + // `~ThreadPoolExecutor` run them during teardown — after any state declared + // below the pool has already been destroyed. Getting this backwards is a + // stack-use-after-scope on `gate`, which is exactly what ASan reported the + // first time round, not a theoretical one. + std::atomic gate{false}; + std::atomic churnSettled{0}; + std::atomic cancelled{0}; + std::atomic parkedRan{0}; + + morph::exec::ThreadPoolExecutor pool{4}; + SyncExecutor cbExec; + morph::backend::LocalBackend backend{pool}; + + // Two instances, so the parked one's strand cannot hold up the churning one. + auto parked = backend.registerModel("BE_CounterModel", morph::model::detail::ModelFactory::create); + auto churner = backend.registerModel("BE_CounterModel", morph::model::detail::ModelFactory::create); + + std::vector>> live; + + for (int round = 0; round < kRounds; ++round) { + // One completion that parks on the gate and so stays live in `_pending`. + live.push_back(backend.execute(parked, pendingCall([&gate, &parkedRan] { + while (!gate.load(std::memory_order_acquire)) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + parkedRan.fetch_add(1, std::memory_order_relaxed); + }), + &cbExec)); + live.back().onError([&cancelled](const std::exception_ptr& exc) { + try { + std::rethrow_exception(exc); + } catch (const morph::backend::BackendChangedError&) { + cancelled.fetch_add(1, std::memory_order_relaxed); + } catch (...) { // NOLINT(bugprone-empty-catch) + } + }); + + // A batch that settles and is then dropped, leaving dead entries behind + // for the next sweep to reclaim. + std::vector>> churn; + churn.reserve(kChurnPerRound); + int const target = churnSettled.load(std::memory_order_relaxed) + kChurnPerRound; + for (int i = 0; i < kChurnPerRound; ++i) { + churn.push_back(backend.execute( + churner, pendingCall([&churnSettled] { churnSettled.fetch_add(1, std::memory_order_relaxed); }), + nullptr)); + } + REQUIRE(morph::testing::waitUntil([&] { return churnSettled.load(std::memory_order_relaxed) >= target; }, + std::chrono::milliseconds{5000}, std::chrono::milliseconds{1})); + churn.clear(); + } + + // Property 1. The exact steady-state size depends on how promptly the strand + // releases each settled task's captured state, so the bound is deliberately + // loose — it only has to sit well below the 3120 entries an uncompacted list + // would hold, and it does. + auto const tracked = backend.trackedPendingCount(); + INFO("tracked=" << tracked << " admitted=" << (kRounds * (kChurnPerRound + 1))); + CHECK(tracked < 1024); + + // Property 2. Every parked completion, admitted across every sweep, is still + // reachable. `cbExec` is inline, so the handlers have all run by the time + // `cancelPending` returns. + backend.cancelPending(std::make_exception_ptr(morph::backend::BackendChangedError{})); + auto const cancelledCount = cancelled.load(std::memory_order_relaxed); + + // Release the parked ops and drain them before any assertion can abandon the + // fixture: `~StrandExecutor` blocks until the running task returns, and a + // `CHECK` that fires mid-teardown should not leave that to chance. + gate.store(true, std::memory_order_release); + REQUIRE(morph::testing::waitUntil([&] { return parkedRan.load(std::memory_order_relaxed) == kRounds; }, + std::chrono::milliseconds{10000})); + live.clear(); + + CHECK(cancelledCount == kRounds); +}