diff --git a/docs/spec/concurrency_and_lifetimes.md b/docs/spec/concurrency_and_lifetimes.md index 050337fe9..0ce457e23 100644 --- a/docs/spec/concurrency_and_lifetimes.md +++ b/docs/spec/concurrency_and_lifetimes.md @@ -408,8 +408,10 @@ the "work" column below is for a synchronous handler unless it says otherwise. | `~Bridge` | **G2** for its calls | Calls waiting for a bind and the backend's pending calls end with `BridgeDestroyedError`, posted | **G4** with a `LocalBackend`: the backend is destroyed with the bridge, and drains its strands, so the destructor returns only once the running action has ended. A Task handler is asked to stop first | M | | `Bridge::switchBackend` | **G2** for the outgoing backend's calls | `BackendChangedError`, posted | The outgoing backend is destroyed inside the switch: with a `LocalBackend`, the switch waits for its running actions | M | | `Bridge::setExecuteDeadline` | **G0** for calls already made | Applies to calls made after it. When a call's deadline fires, the call ends as in G2, with `ClientTimeoutError` | A synchronous action runs to its end; a Task handler is asked to stop | M | +| A stop on `BridgeHandler::execute(action, stop)`'s token | **G2**, from any thread | `OperationCancelled` is posted to the call's `cbExec` by the thread that requested the stop; the call's own outcome, later, reaches nobody. A token already stopped rejects the call the same way without dispatching it | Asked to stop, not waited for: a Task handler on a `LocalBackend` sees the stop. A synchronous action runs to its end. **On a remote backend the call is only abandoned**: no `cancel` crosses the wire, so the server's handler runs to its end and its reply is dropped (measured with `SimulatedRemoteBackend`). The call stays counted as pending until that reply, as for a fired deadline | M | | `BridgeHandler::unsubscribe()` | **G1** | A delivery already posted still runs after it returns; no new one is scheduled. Pair it with a `CallbackScope` for G3 | n/a | M | | `CallbackScope::requestStop()`, `reset()`, `~CallbackScope` | **G3** on the delivery executor; **G1** from another thread | A gated callback posted but not yet run is refused when it runs. From another thread the check and the body are not atomic, so a callback that has passed its check still runs — advisory, and not measured as a race | Never waits for a callback in its body (measured). Stops every call a gated continuation is attached to | M | +| A stop on a coroutine suspended in `co_await completion` | **G2** for the await | The coroutine's resumption with `OperationCancelled` is posted to the executor it suspended on; the await's handlers are refused, so the call's own outcome, later, does not reach it | The call is asked to stop, through the awaiter's scope linked to its stop source: a Task handler on a `LocalBackend` sees the stop (measured). Not waited for. A synchronous action runs to its end; a remote server's handler runs on | M | | `TimeoutScheduler::cancel` | **G0** on return; **G3** for a callback not yet started, once the loop has run the cancel | The cancel is posted to the loop: on return the callback still holds its captures | Never waits for a callback already running | M | | `~TimeoutScheduler` | **G4** | Every pending callback is dropped unfired | Returns only once no callback of the scheduler is running | M | | `LimitPolicy::executeTimeout` (`RemoteServer`) | **G2** for the reply | `err "timeout"` is sent; the handler's own reply, later, is dropped — one reply per call | A synchronous handler runs to its end; a Task handler is asked to stop | M | @@ -425,14 +427,9 @@ the "work" column below is for a synchronous handler unless it says otherwise. - **No cancel crosses the wire.** A remote call is settled locally only; the server finishes the handler and its reply is dropped. There is no `cancel` envelope. -- **A `co_await` that is stopped withdraws the await, not the call.** The - coroutine resumes with `OperationCancelled`; the call it was waiting on runs - on, because the awaiter's stop is not linked to the call's stop source. A - continuation attached through a `CallbackScope` *is* linked: stopping the - scope stops the call. -- **A caller has no direct cancel verb on one call.** `Completion` has no - cancel API and `execute()` takes no stop token; the scope link above and the - execute deadline are the only per-call routes. +- **`Completion` has no cancel API of its own.** A call is cancelled through + the token given to `execute(action, stop)`, a stopped scope a continuation is + gated by, a stopped `co_await`, or the execute deadline. ## Gating callbacks on a receiver's lifetime — `CallbackScope` diff --git a/docs/spec/core/bridge.md b/docs/spec/core/bridge.md index 0d086d7f3..793a490ee 100644 --- a/docs/spec/core/bridge.md +++ b/docs/spec/core/bridge.md @@ -868,7 +868,7 @@ is tolerated, as above. | `registerHandler(binding)` | `void registerHandler(const shared_ptr&)` | Pre-built binding. Same bind rule. | | `switchBackend` | `void switchBackend(unique_ptr)` / `void switchBackend(shared_ptr)` | Tells the new backend its owner and pushes the default session, then issues a re-bind for every live binding. Binds settled before returning decide it: a failure releases what was acquired and rethrows with the old backend and every binding untouched; otherwise it commits — publishes settled ids, leaves bindings with a bind in flight holding their calls, swaps, notifies, moves the reconnect handler, and cancels the old backend's pending calls with `BackendChangedError` (stopping running Task handlers). The `unique_ptr` overload is a template on the concrete backend type and delegates to the `shared_ptr` one. | | `deregisterHandler` | `void deregisterHandler(const shared_ptr&)` | Supersedes a bind in flight (its reply releases its instance), rejects held calls with `HandlerDestroyedError`, deregisters from the active backend (if bound), resets `currentId` to 0, removes from tracking. | -| `executeVia` | `Completion executeVia(const shared_ptr&, Action, IExecutor*, function onResult = {})` | On the owner. Counts the call pending and arms the execute deadline when it is made; dispatches now when no bind is in flight, else holds it until the bind settles (a failed bind rejects it with the bind's error; a never-attached `AllowShared` binding with `"handler not bound"`). Attaches the default session. On `LocalBackend`, rejects an action whose `ActionValidator::ready` returns `false` with `morph::model::ValidationError` via `onError`, before `Model::execute` runs. Records a journal `LogEntry` for loggable actions on both success (`Outcome::Succeeded`) and a throwing `Model::execute` (`Outcome::Failed`, rethrown unchanged); a failure to serialise the result or to append the success entry happens after the mutation committed and rejects the completion with `morph::model::ActionRecordingError` instead. Dispatches through `IBackend::executeInto`, handing the backend a `detail::BridgeSink` that is simultaneously the caller's typed completion state and the backend's settle sink. The bridge-side work a result triggers (`onResult`, the subscription fan-out) runs on the owner, gated on the bridge's `CallbackToken`. A call whose handler returns a `Task` carries a `StopSource`, requested by the deadline, by a `CallbackScope` a callback was attached through, and by the backend's `cancelPending`. | +| `executeVia` | `Completion executeVia(const shared_ptr&, Action, IExecutor*, function onResult = {})` | On the owner. Counts the call pending and arms the execute deadline when it is made; dispatches now when no bind is in flight, else holds it until the bind settles (a failed bind rejects it with the bind's error; a never-attached `AllowShared` binding with `"handler not bound"`). Attaches the default session. On `LocalBackend`, rejects an action whose `ActionValidator::ready` returns `false` with `morph::model::ValidationError` via `onError`, before `Model::execute` runs. Records a journal `LogEntry` for loggable actions on both success (`Outcome::Succeeded`) and a throwing `Model::execute` (`Outcome::Failed`, rethrown unchanged); a failure to serialise the result or to append the success entry happens after the mutation committed and rejects the completion with `morph::model::ActionRecordingError` instead. Dispatches through `IBackend::executeInto`, handing the backend a `detail::BridgeSink` that is simultaneously the caller's typed completion state and the backend's settle sink. The bridge-side work a result triggers (`onResult`, the subscription fan-out) runs on the owner, gated on the bridge's `CallbackToken`. A call whose handler returns a `Task` carries a `StopSource`, requested by the deadline, by a `CallbackScope` a callback was attached through, by a `co_await` of its completion that is stopped, by the token given to `BridgeHandler::execute(action, stop)`, and by the backend's `cancelPending`. | | `setDefaultSession` | `void setDefaultSession(session::Context)` | Installs default session context; also pushes it to the active backend via `IBackend::setSession` so control envelopes (register/attach/assign/deregister) carry it too, not only `execute`. | | `defaultSession` | `session::Context defaultSession() const` | Returns snapshot of default session. | | `setExecuteDeadline` | `void setExecuteDeadline(std::chrono::milliseconds)` | Opt-in client-side execute deadline; `0` (the default) disables it. Lazily creates the backing `TimeoutScheduler`, on a private `exec::IoLoop` with one thread, on first enable. | @@ -886,6 +886,7 @@ is tolerated, as above. | ctor (custom binding) | `BridgeHandler(Bridge&, IExecutor*, shared_ptr)` | Registers the pre-built binding the same way. | | dtor | `~BridgeHandler()` | On the owner. Deregisters via `Bridge::deregisterHandler` (rejecting held calls with `HandlerDestroyedError`), but only if the bridge's `CallbackToken` is still active; a no-op if the `Bridge` was already destroyed. | | `execute` | `Completion execute(Action)` | Typed dispatch through the bridge; held until the handler's bind settles when one is in flight. For a shared handler, a payload-/result-keyed action's attach or promote step never throws synchronously — a backend refusal (e.g. `LimitPolicy::maxLiveModels`) resolves the returned `Completion` via `.onError(...)`. | +| `execute` (cancellable) | `Completion execute(Action, core::async::StopToken stop)` | As `execute(Action)`, with a caller's cancel. A stop on `stop` before the call settles rejects the `Completion` with `core::async::OperationCancelled` — posted from the thread that requested the stop — and requests stop on the call's `StopSource`, so a Task handler on a `LocalBackend` sees it; a synchronous handler runs to its end, and a remote server's handler runs on, since no cancel crosses the wire. The call stays pending until the backend settles it, as for a fired deadline. A token already stopped rejects the call without dispatching it. See the cancellation policy in [concurrency_and_lifetimes.md](../concurrency_and_lifetimes.md#cancellation-policy). | | `executeJson` | `Completion executeJson(string_view actionType, string_view bodyJson)` | Type-erased dispatch through `ActionExecuteRegistry`. | | `subscribe(cb)` | `void subscribe(function)` | Fire `cb` whenever an `R` is produced on the attached instance. | | `subscribe(scope, cb)` | `void subscribe(CallbackScope const&, function)` | As above, gated on the scope's liveness and stop state ([callback_scope.md](callback_scope.md)). Dead sinks are refused, not pruned. | diff --git a/docs/spec/core/completion.md b/docs/spec/core/completion.md index 17960b08d..f3ffe6fc1 100644 --- a/docs/spec/core/completion.md +++ b/docs/spec/core/completion.md @@ -66,7 +66,7 @@ by `std::make_shared`, which every site in the tree does | `onErrAttached` | `std::atomic` | `cbExec`, or a synchronous taker | Suppresses orphan logging; set by an `onError` attach on a state with an executor, and by `bridge::detail::takeSettled` | | `presumedOwnerThread`, `deliveredOn` | `std::atomic` | — | The debug check of an attach made outside every executor's task; see [Thread safety](#thread-safety) | | `cbExec` | `::morph::exec::IExecutor*` | set before publication | The owner; may be `nullptr` | -| `stopSource`, `stopLinks` | | `cbExec` | The call's stop source and the scope links relaying to it; released by `deliver()` | +| `stopSource`, `stopLinks` | | `cbExec` | The call's stop source, and the links relaying a stop to it — a scope's, an await's, or a caller's cancel token, which also settles the call; released by `deliver()` | **Setting a value or exception.** `setValue(T)` and `setException(exception_ptr)` are called by the producer, on any thread. The first claims `settled` with an @@ -664,16 +664,19 @@ future/promise or a monadic async type. Its scope is narrow by design: `T → U` mapping and no way to chain one asynchronous step onto another. To sequence work, the consumer must start a fresh operation from inside the handler. -- **No *work* cancellation.** There is no handle to cancel an outstanding - operation; once started, it runs to completion (or is abandoned). - `Bridge::setExecuteDeadline` (see - [Client-side execute deadline](#client-side-execute-deadline)) is not an - exception to this: it bounds how long the *caller* waits by resolving the - state early, and does nothing to the work still in flight underneath. - **Delivery**, by contrast, *can* be stopped — see - [Lifetime and stop gating](#lifetime-and-stop-gating). The distinction is - sharp and deliberate: a `CallbackScope` says "do not hand me this result", - never "stop producing it". Work-side cancellation is not offered at all. +- **No cancel API on `Completion` itself.** A `Completion` cannot cancel the + operation it reports on. The call's producer can link a cancel to it: a + `Bridge` call whose handler returns a `Task` carries a stop source, which a + stopped `CallbackScope` a callback was attached through, a stopped + `co_await`, the execute deadline and the token given to + `BridgeHandler::execute(action, stop)` all request. That token also settles + the call with `core::async::OperationCancelled`; the deadline settles it with + `ClientTimeoutError`. A synchronous handler, or a server across the wire, is + not reached by any of them: the token and the deadline settle the state + early, and the work runs on underneath (see + [Client-side execute deadline](#client-side-execute-deadline) and the + cancellation policy in + [concurrency_and_lifetimes.md](../concurrency_and_lifetimes.md#cancellation-policy)). - **Single consumer handle, but multiple handlers per outcome.** The `Completion` handle itself is move-only — only one owner at a time — but each state's `onOk`/`onErr` are vectors, so repeated `then()`/`onError()` @@ -694,9 +697,9 @@ state; the log is emitted only when the state itself is finally destroyed with a ## Out of scope -- Work cancellation — there is no mechanism to cancel an outstanding - *operation*. Stopping *delivery* of its result is - [`CallbackScope`](callback_scope.md); the two are different things. +- A cancel on the `Completion` handle — cancelling the operation belongs to + whoever produced it (see [Limitations](#limitations)). Stopping *delivery* + of its result is [`CallbackScope`](callback_scope.md). - Multiple values — `Completion` is a single-result primitive. - Synchronous blocking — there is no `wait()` or `get()`; the API is callback-only. diff --git a/docs/spec/core/coroutines.md b/docs/spec/core/coroutines.md index 26dc315db..4e06aa75c 100644 --- a/docs/spec/core/coroutines.md +++ b/docs/spec/core/coroutines.md @@ -167,10 +167,20 @@ a stop callback for the duration of the suspension: executor it suspended on, or on the completion's executor. - **A token already stopped at `co_await`** resumes with `OperationCancelled` without attaching anything to the completion. - -The completion itself is not cancelled: it is still settled by whatever -produces it, and its other handlers still run. Cancellation withdraws this one -await. +- **A stop also stops the call.** The await owns the call it waits on, the way + a continuation attached through a `CallbackScope` does: the awaiter's scope + is linked to the call's stop source when the handlers are attached + (`CompletionState::linkStop`), so the stop that withdraws the await — or the + frame destroyed while suspended — asks the call to stop too. A stop that + lands before the handlers are attached, a token already stopped included, + is relayed to the call on the completion's executor. Only a call that + carries a stop source can be stopped: a `Bridge` call whose handler returns + a `Task`, on a `LocalBackend`. Its handler sees `OperationCancelled` at its + next stop-aware `co_await`. + +The completion itself is still settled by whatever produces it — a stopped +Task handler rejects it with `OperationCancelled` — and its other handlers +still run; the withdrawn await's own handlers do not. ### `spawn` — the entry point from non-coroutine code diff --git a/include/morph/core/bridge.hpp b/include/morph/core/bridge.hpp index b803316f4..daff4afb8 100644 --- a/include/morph/core/bridge.hpp +++ b/include/morph/core/bridge.hpp @@ -7,6 +7,7 @@ #include #include #include +#include #include #include #include @@ -2226,6 +2227,43 @@ class BridgeHandler { } } + /// @brief Dispatches @p action as `execute(action)` does, cancellable through @p stop. + /// + /// A stop on @p stop before the call settles cancels it: the returned + /// `Completion` is rejected with `core::async::OperationCancelled`, posted + /// to the GUI executor from the thread that requested the stop, and the + /// call's stop source is asked to stop. A handler returning a `Task`, on a + /// `LocalBackend`, sees that stop at its next stop-aware `co_await`; a + /// synchronous handler runs to its end, and a remote server's handler runs + /// on, since no cancel crosses the wire. Either way the call's own outcome, + /// later, reaches nobody. A token already stopped rejects the call the same + /// way without dispatching it. + /// + /// The call stays counted as pending until the backend settles it, as a + /// call the execute deadline settled does. + /// + /// @tparam Action Concrete action type registered with `BRIDGE_REGISTER_ACTION`. + /// @param action Action to execute (moved into the dispatch). + /// @param stop The caller's cancel. Observed until the call settles; a + /// token that can never be stopped makes this `execute(action)`. + /// @return Completion that resolves on the GUI executor, or rejects with + /// `core::async::OperationCancelled` once @p stop is stopped. + template + ::morph::async::Completion::Result> execute( + Action action, ::core::async::StopToken stop) { + using R = ::morph::model::ActionTraits::Result; + if (stop.stop_requested()) { + auto [cancelled, promise] = ::morph::async::Completion::makeSettleable(_guiExec); + promise.reject(std::make_exception_ptr(::core::async::OperationCancelled{})); + return std::move(cancelled); + } + auto completion = execute(std::move(action)); + // Never empty: every path of `execute(action)` returns a completion + // built on a state, a rejected one included. + completion.state()->linkCancel(std::move(stop)); + return completion; + } + /// @brief Attaches (or re-points) this handler to the instance for @p key. /// /// Creates the instance if no live instance holds @p key, otherwise joins the diff --git a/include/morph/core/completion.hpp b/include/morph/core/completion.hpp index 18c8cffa3..7fecf43ba 100644 --- a/include/morph/core/completion.hpp +++ b/include/morph/core/completion.hpp @@ -3,6 +3,7 @@ #pragma once #include #include +#include #include #include #include @@ -145,6 +146,44 @@ struct CompletionState : std::enable_shared_from_this> { std::move(scopeStop), StopRelay{std::weak_ptr<::core::async::StopSource>{stopSource}})); } + // Relays a caller's cancel to the call: settles it with `OperationCancelled` + // first, so that outcome wins over the one the stopped work produces, then + // requests stop on its stop source. Holds both weakly. Runs on whichever + // thread requested the stop; a settle only posts its delivery. + struct CancelRelay { + std::weak_ptr state; + std::weak_ptr<::core::async::StopSource> source; + void operator()() const noexcept { + if (auto const strong = state.lock()) { + try { + strong->setException(std::make_exception_ptr(::core::async::OperationCancelled{})); + } catch (...) { // NOLINT(bugprone-empty-catch): a stop callback must not throw + // An executor that refuses the delivery leaves nothing to tell. + } + } + if (auto const strong = source.lock()) { + static_cast(strong->request_stop()); + } + } + }; + + // Links @p cancel to this call until the state is delivered: a stop on it + // settles the call with `OperationCancelled` and stops its stop source, if + // it has one. A no-op for a token that can never be stopped, a state + // already delivered, or one with no executor. A token already stopped + // cancels the call here. Runs on the owner. + void linkCancel(::core::async::StopToken cancel) { + if (cbExec == nullptr || delivered.load(std::memory_order_relaxed) || !cancel.stop_possible()) { + return; + } + checkOwner("Completion::linkCancel"); + stopLinks.push_back(std::make_shared<::core::async::StopCallback>( + std::move(cancel), CancelRelay{ + .state = this->weak_from_this(), + .source = std::weak_ptr<::core::async::StopSource>{stopSource}, + })); + } + void setValue(T val) { if (settled.exchange(true, std::memory_order_acq_rel)) { return; diff --git a/include/morph/core/detail/completion_awaiter.hpp b/include/morph/core/detail/completion_awaiter.hpp index 17b9d0940..f0fefd226 100644 --- a/include/morph/core/detail/completion_awaiter.hpp +++ b/include/morph/core/detail/completion_awaiter.hpp @@ -37,7 +37,8 @@ struct CompletionState; /// `CallbackToken`. With a stoppable token on the awaiting promise it also /// registers a stop callback. Exactly one of *the completion's handler* and /// *the stop callback* moves `Shared::outcome` off `Pending`, and only that one -/// resumes the coroutine. +/// resumes the coroutine. A stop that withdraws the await also stops the call +/// the completion reports on, when that call carries a stop source. /// /// The pair is attached on the completion's executor, which owns its handler /// lists: directly when the awaiting coroutine is running there, otherwise by @@ -169,6 +170,7 @@ class CompletionAwaiter { ::core::async::StopToken const token = awaiting.promise().stopToken(); if (token.stop_requested()) { shared->outcome.store(Outcome::Cancelled); + stopUnattached(state, token); return false; } if (token.stop_possible()) { @@ -177,6 +179,7 @@ class CompletionAwaiter { if (shared->outcome.load(std::memory_order_seq_cst) == Outcome::Cancelled) { // A stop landed while registering. Whoever claims answers // it; if onStop claimed first it is resuming us already. + stopUnattached(state, token); return shared->claimed.exchange(true, std::memory_order_acq_rel); } } @@ -241,8 +244,15 @@ class CompletionAwaiter { private: /// Attaches the await's handler pair to @p state; runs on its executor. + /// + /// The await owns the call it waits on, as a scope-gated continuation + /// does: the scope's stop -- a stop that withdraws the await, or the frame + /// destroyed while suspended -- is linked to the call's stop source, so + /// withdrawing the await also asks the call to stop. Linked after the + /// scope has stopped, it stops the call at once. static void attach(CompletionState& state, const std::shared_ptr& shared) { auto const token = shared->scope.token(); + state.linkStop(token.stopToken()); state.attachThen([token, shared](const T& settled) { if (token.active() && shared->decide(Outcome::Settled)) { shared->value.emplace(settled); @@ -257,6 +267,23 @@ class CompletionAwaiter { }); } + /// Stops the call @p state reports on, for an await a stop withdrew before + /// it attached, so no link carries that stop to the call. @p stopped, + /// already stopped, is linked on the completion's executor, which stops the + /// call there. Best effort: an executor that refuses the task leaves the + /// call running; the coroutine resumes with the cancellation either way. + static void stopUnattached(const std::shared_ptr>& state, + const ::core::async::StopToken& stopped) noexcept { + try { + if (::morph::exec::runningOn(*state->cbExec)) { + state->linkStop(stopped); + return; + } + state->cbExec->post([state, stopped] { state->linkStop(stopped); }); + } catch (...) { // NOLINT(bugprone-empty-catch): the await is answered already + } + } + std::shared_ptr> _state; std::shared_ptr _shared; }; diff --git a/scripts/branch_partial_allowlist.json b/scripts/branch_partial_allowlist.json index 19e1fa0f1..7f801172e 100644 --- a/scripts/branch_partial_allowlist.json +++ b/scripts/branch_partial_allowlist.json @@ -109,7 +109,7 @@ }, { "file": "include/morph/core/bridge.hpp", - "line": 1772, + "line": 1773, "source": "if (_executeDeadline.count() <= 0 || !_timeoutScheduler) {", "reason": "The `!_timeoutScheduler` disjunct is unreachable by construction. `setExecuteDeadline` is the only writer of both `_executeDeadline` and `_timeoutScheduler`, and it creates `_timeoutScheduler` in the same call that makes `_executeDeadline` positive; nothing resets `_timeoutScheduler` to null afterwards (setting the deadline back to 0 only stops new calls from arming it, per that method's doc comment). So `_executeDeadline > 0 && !_timeoutScheduler` cannot hold once any positive deadline has been set." }, diff --git a/tests/test_async_registration.cpp b/tests/test_async_registration.cpp index ce9f0f724..f8d4fa72f 100644 --- a/tests/test_async_registration.cpp +++ b/tests/test_async_registration.cpp @@ -913,20 +913,23 @@ TEST_CASE("Bridge::assignHandlerPrimary: a promotion reply after the handler or Owner owner; morph::exec::ThreadPoolExecutor pool{2}; auto async = std::make_shared(pool); + // The outcomes outlive every pump below: their handlers are not + // scope-gated, so a reply settling inside a later runFor() is delivered + // into them after the handler or the bridge that made the call is gone. + Outcome afterHandler; + Outcome afterBridge; { morph::bridge::Bridge bridge{std::make_unique(pool), owner}; bridge.switchBackend(std::static_pointer_cast(async)); { morph::bridge::BridgeHandler handler{bridge, &owner}; - Outcome outcome; - track(outcome, handler.execute(ARCreate{.initial = 3})); + track(afterHandler, handler.execute(ARCreate{.initial = 3})); REQUIRE(pumpOwnerUntil(owner, [&] { return async->pendingCount() == 1; })); } async->completeNext(); // the handler is gone owner.runFor(std::chrono::milliseconds{5}); morph::bridge::BridgeHandler handler{bridge, &owner}; - Outcome outcome; - track(outcome, handler.execute(ARCreate{.initial = 4})); + track(afterBridge, handler.execute(ARCreate{.initial = 4})); REQUIRE(pumpOwnerUntil(owner, [&] { return async->pendingCount() == 1; })); } async->completeNext(); // the bridge is gone diff --git a/tests/test_cancellation_policy.cpp b/tests/test_cancellation_policy.cpp index 072984589..47af78a47 100644 --- a/tests/test_cancellation_policy.cpp +++ b/tests/test_cancellation_policy.cpp @@ -12,15 +12,21 @@ // G3 no further callback will start // G4 G3, plus none is still running // -// "Work" is the model action or server handler the verb abandons. Whether a -// Task handler is asked to stop is measured in test_coroutine_model.cpp; these -// cases use a synchronous handler, which nothing can stop, so they measure -// what each verb does to the callbacks and whether it waits for that work. +// "Work" is the model action or server handler the verb abandons. Most cases +// use a synchronous handler, which nothing can stop, so they measure what each +// verb does to the callbacks and whether it waits for that work. The verbs +// whose point is to stop the work -- a stopped co_await, a stopped execute +// token -- are measured +// against a Task handler, which can observe a stop; the other verbs' Task +// handler stops are measured in test_coroutine_model.cpp. #include #include #include #include +#include +#include +#include #include #include #include @@ -30,6 +36,7 @@ #include #include #include +#include #include #include #include @@ -60,6 +67,10 @@ struct CPBlock { struct CPFast { int tag = 0; }; +/// Handled by a Task handler that sleeps until it is stopped. +struct CPSleep { + int ms = 0; +}; namespace { @@ -82,6 +93,39 @@ CPGate& gate() { return instance; } +/// The Task handler's progress, and the scheduler it sleeps on. +struct CPSleeper { + std::atomic started{0}; + std::atomic cancelled{0}; + std::atomic finished{0}; + morph::async::detail::TimeoutScheduler* scheduler = nullptr; +}; + +CPSleeper& sleeper() { + static CPSleeper instance; + return instance; +} + +/// Owns the scheduler `CPSleep` waits on for one test, and joins its thread +/// when the test ends rather than keeping it for the rest of the run. +class SleeperScope { +public: + SleeperScope() { + sleeper().started = 0; + sleeper().cancelled = 0; + sleeper().finished = 0; + sleeper().scheduler = &_scheduler; + } + SleeperScope(const SleeperScope&) = delete; + SleeperScope& operator=(const SleeperScope&) = delete; + SleeperScope(SleeperScope&&) = delete; + SleeperScope& operator=(SleeperScope&&) = delete; + ~SleeperScope() { sleeper().scheduler = nullptr; } + +private: + morph::async::detail::TimeoutScheduler _scheduler; +}; + template [[nodiscard]] bool holds(const std::exception_ptr& error) { try { @@ -116,12 +160,24 @@ struct CPModel { return action.tag; } int execute(CPFast action) { return action.tag; } + core::async::Task execute(CPSleep action) { + sleeper().started.fetch_add(1); + try { + co_await morph::async::delay(*sleeper().scheduler, std::chrono::milliseconds{action.ms}); + } catch (const core::async::OperationCancelled&) { + sleeper().cancelled.fetch_add(1); + throw; + } + sleeper().finished.fetch_add(1); + co_return action.ms; + } // NOLINTEND(readability-convert-member-functions-to-static) }; BRIDGE_REGISTER_MODEL(CPModel, "CP_Model") BRIDGE_REGISTER_ACTION(CPModel, CPBlock, "CP_Block") BRIDGE_REGISTER_ACTION(CPModel, CPFast, "CP_Fast") +BRIDGE_REGISTER_ACTION(CPModel, CPSleep, "CP_Sleep") namespace { @@ -403,6 +459,179 @@ TEST_CASE("Bridge::setExecuteDeadline is G0 for calls already made; a fired dead CHECK(after.ok == 0); } +TEST_CASE("execute with a stop token is G2 when stopped, and stops the Task handler running the call", + "[cancel-policy]") { + SleeperScope const sleeping; + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + Outcome outcome; + auto completion = handler.execute(CPSleep{.ms = 60'000}, stop.get_token()); + outcome.attach(completion); + REQUIRE(pumpUntil(owner, [] { return sleeper().started.load() == 1; })); + + static_cast(stop.request_stop()); + + // Not G3: the terminal callback has not run on return. + CHECK(outcome.err == 0); + // G2: OperationCancelled is posted, and runs on the next pump. + REQUIRE(pumpUntil(owner, [&] { return outcome.err == 1; })); + CHECK(holds(outcome.error)); + // The work: the Task handler is asked to stop. + REQUIRE(pumpUntil(owner, [] { return sleeper().cancelled.load() == 1; })); + CHECK(sleeper().finished.load() == 0); + // G1: the stopped handler's own outcome reaches nobody. + owner.runFor(30ms); + CHECK(outcome.ok == 0); + CHECK(outcome.err == 1); +} + +TEST_CASE("execute with a stop token stopped from another thread settles the call; a synchronous action runs on", + "[cancel-policy]") { + gate().reset(); + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + Outcome outcome; + auto completion = handler.execute(CPBlock{.tag = 1}, stop.get_token()); + outcome.attach(completion); + REQUIRE(pumpUntil(owner, [] { return gate().started.load() == 1; })); + + std::thread{[&stop] { static_cast(stop.request_stop()); }}.join(); + + CHECK(outcome.err == 0); + REQUIRE(pumpUntil(owner, [&] { return outcome.err == 1; })); + CHECK(holds(outcome.error)); + // No join, and nothing can stop a synchronous action: it is still running. + CHECK(gate().finished.load() == 0); + gate().released = true; + REQUIRE(morph::testing::waitUntil([] { return gate().finished.load() == 1; })); + owner.runFor(30ms); + CHECK(outcome.ok == 0); + CHECK(outcome.err == 1); +} + +TEST_CASE("execute with a stop token already stopped rejects the call without dispatching it", "[cancel-policy]") { + gate().reset(); + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + static_cast(stop.request_stop()); + Outcome outcome; + auto completion = handler.execute(CPBlock{.tag = 1}, stop.get_token()); + outcome.attach(completion); + + REQUIRE(pumpUntil(owner, [&] { return outcome.err == 1; })); + CHECK(holds(outcome.error)); + owner.runFor(30ms); + CHECK(gate().started.load() == 0); +} + +TEST_CASE("execute with a stop token on SimulatedRemoteBackend abandons the call; the server's handler runs on", + "[cancel-policy]") { + gate().reset(); + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + auto const server = std::make_shared(pool); + morph::bridge::Bridge bridge{std::make_unique(*server), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + REQUIRE(pumpUntil(owner, [&] { return handler.isBound(); })); + + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + Outcome outcome; + auto completion = handler.execute(CPBlock{.tag = 1}, stop.get_token()); + outcome.attach(completion); + REQUIRE(pumpUntil(owner, [] { return gate().started.load() == 1; })); + + static_cast(stop.request_stop()); + + CHECK(outcome.err == 0); + REQUIRE(pumpUntil(owner, [&] { return outcome.err == 1; })); + CHECK(holds(outcome.error)); + // Nothing crossed the wire: the server's handler finishes, and its reply + // is dropped. + CHECK(gate().finished.load() == 0); + gate().released = true; + REQUIRE(morph::testing::waitUntil([] { return gate().finished.load() == 1; })); + owner.runFor(30ms); + CHECK(outcome.ok == 0); + CHECK(outcome.err == 1); +} + +TEST_CASE("a cancel token links nothing to a call with no executor, one already delivered, or when it cannot stop", + "[cancel-policy]") { + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + + // No executor: nothing could ever deliver the cancellation. + auto const unowned = std::make_shared>(); + unowned->linkCancel(stop.get_token()); + CHECK(unowned->stopLinks.empty()); + + morph::exec::MainThreadExecutor owner; + auto [completion, promise] = morph::async::Completion::makeSettleable(&owner); + auto const state = completion.state(); + Outcome outcome; + outcome.attach(completion); + + // A token no source can stop: there is nothing to link. + state->linkCancel(core::async::StopToken{}); + CHECK(state->stopLinks.empty()); + + // Delivered: the call has nothing left to cancel, and its outcome stands. + promise.resolve(1); + REQUIRE(pumpUntil(owner, [&] { return outcome.ok == 1; })); + state->linkCancel(stop.get_token()); + CHECK(state->stopLinks.empty()); + static_cast(stop.request_stop()); + owner.runFor(20ms); + CHECK(outcome.ok == 1); + CHECK(outcome.err == 0); +} + +TEST_CASE("a cancel link that outlives its call still stops the call's stop source, and settles nothing", + "[cancel-policy]") { + morph::exec::MainThreadExecutor owner; + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource cancel; + auto const callStop = std::make_shared(); + std::shared_ptr link; + std::weak_ptr> gone; + { + auto [completion, promise] = morph::async::Completion::makeSettleable(&owner); + auto const state = completion.state(); + state->stopSource = callStop; + state->linkCancel(cancel.get_token()); + REQUIRE(state->stopLinks.size() == 1); + // Held past the state, as a stop callback running while the state + // is released would be. + link = state->stopLinks.front(); + gone = state; + } + REQUIRE(gone.expired()); + REQUIRE(link != nullptr); + + static_cast(cancel.request_stop()); + + // The relay holds the state weakly: with it gone there is nothing to + // settle, and the call's stop source is still asked to stop. + CHECK(callStop->stop_requested()); + owner.runFor(20ms); +} + namespace { /// The binding type `SubscriptionRegistry` matches on: only its instance id. @@ -478,6 +707,88 @@ TEST_CASE("CallbackScope stop is G3 on the delivery executor, and never waits fo } } +// ── A stopped co_await ────────────────────────────────────────────────────── + +namespace { + +/// What reached a coroutine awaiting one call, on the owner. +struct AwaitSeen { + /// Set in the step that suspends, just before the `co_await`. + std::atomic suspending{false}; + std::atomic finished{false}; + bool cancelled = false; + /// Anything other than the cancellation: a value, or another error. + bool settled = false; +}; + +core::async::Task awaitCall(morph::async::Completion call, std::shared_ptr seen) { + try { + seen->suspending = true; + static_cast(co_await std::move(call)); + seen->settled = true; + } catch (const core::async::OperationCancelled&) { + seen->cancelled = true; + } catch (...) { + seen->settled = true; + } + seen->finished = true; +} + +} // namespace + +TEST_CASE("A stopped co_await is G2 for the await, and stops the Task handler it was waiting on", "[cancel-policy]") { + SleeperScope const sleeping; + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + + auto seen = std::make_shared(); + auto task = awaitCall(handler.execute(CPSleep{.ms = 60'000}), seen); + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + task.handle().promise().setStopToken(stop.get_token()); + morph::async::spawn(owner, std::move(task)); + REQUIRE(pumpUntil(owner, [&] { return seen->suspending.load() && sleeper().started.load() == 1; })); + + static_cast(stop.request_stop()); + + // Not G3: the coroutine has not resumed on return. + CHECK_FALSE(seen->finished.load()); + // G2: its resumption with OperationCancelled is posted, and runs on a pump. + REQUIRE(pumpUntil(owner, [&] { return seen->finished.load(); })); + CHECK(seen->cancelled); + // The work: the handler the await was waiting on is asked to stop. + REQUIRE(pumpUntil(owner, [] { return sleeper().cancelled.load() == 1; })); + CHECK(sleeper().finished.load() == 0); + // G1: the call's own outcome, settled by the stopped handler, reaches + // nothing. + owner.runFor(30ms); + CHECK_FALSE(seen->settled); +} + +TEST_CASE("A co_await under a stop already requested stops the Task handler it would have waited on", + "[cancel-policy]") { + SleeperScope const sleeping; + morph::exec::ThreadPoolExecutor pool{2}; + morph::exec::MainThreadExecutor owner; + morph::bridge::Bridge bridge{std::make_unique(pool), owner}; + morph::bridge::BridgeHandler handler{bridge, &owner}; + + auto seen = std::make_shared(); + auto task = awaitCall(handler.execute(CPSleep{.ms = 60'000}), seen); + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + static_cast(stop.request_stop()); + task.handle().promise().setStopToken(stop.get_token()); + morph::async::spawn(owner, std::move(task)); + + REQUIRE(pumpUntil(owner, [&] { return seen->finished.load(); })); + CHECK(seen->cancelled); + REQUIRE(pumpUntil(owner, [] { return sleeper().cancelled.load() == 1; })); + CHECK(sleeper().finished.load() == 0); +} + // ── TimeoutScheduler ──────────────────────────────────────────────────────── TEST_CASE("TimeoutScheduler::cancel is G0 on return and G3 once the loop has run it; it never waits", diff --git a/tests/test_coroutine_client.cpp b/tests/test_coroutine_client.cpp index f412fd516..d31695242 100644 --- a/tests/test_coroutine_client.cpp +++ b/tests/test_coroutine_client.cpp @@ -191,6 +191,29 @@ TEST_CASE("co_await under a stop already requested does not suspend", "[coroutin REQUIRE_FALSE(seen->value.has_value()); } +TEST_CASE("a stop already requested at co_await off the completion's executor stops the call there", + "[coroutine][client]") { + morph::exec::MainThreadExecutor exec; + auto const state = std::make_shared>(); + state->stopSource = std::make_shared(); + auto const seen = std::make_shared(); + + auto const task = awaitInto(Completion{state, &exec}, seen); + // NOLINTNEXTLINE(misc-const-correctness): request_stop() is non-const + core::async::StopSource stop; + static_cast(stop.request_stop()); + task.handle().promise().setStopToken(stop.get_token()); + // Resumed by hand, outside every executor's task: the await answers at + // once, and the stop is relayed to the call through the completion's + // executor, not here. + task.handle().resume(); + REQUIRE(seen->finished.load()); + REQUIRE(seen->cancelled); + CHECK_FALSE(state->stopSource->stop_requested()); + + REQUIRE(pumpUntil(exec, [&] { return state->stopSource->stop_requested(); })); +} + TEST_CASE("a rejection after a stop withdrew the await reaches nothing", "[coroutine][client]") { morph::exec::MainThreadExecutor exec; auto [completion, promise] = Completion::makeSettleable(&exec);