From 0c0ac94f020c9b355e811189e25bbb9bef860784 Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Tue, 11 Aug 2026 01:23:14 -0300 Subject: [PATCH 1/6] Add `Cancel` method parameter to `proxy.capnp` Methods can declare a `cancel :Proxy.Cancel` parameter, which maps to a C++ `CancelToken` argument and lets the method detect when its request has been abandoned by the client, either because the client disconnected or because the promise was dropped. This commit only defines the schema type and makes `mpgen` reject a `Cancel` parameter if no preceding `Context` parameter exists. Later commits will enable cancellation at capnp layer and introduce the `CancelToken` type. --- include/mp/proxy.capnp | 9 +++++++++ src/mp/gen.cpp | 13 +++++++++++++ 2 files changed, 22 insertions(+) diff --git a/include/mp/proxy.capnp b/include/mp/proxy.capnp index e0a66fd9..346152f0 100644 --- a/include/mp/proxy.capnp +++ b/include/mp/proxy.capnp @@ -67,3 +67,12 @@ struct Context $count(0) { # Handle of the client thread that is calling the current method, and that # any callbacks made by the server thread should be made on. } + +struct Cancel { + # Method parameter mapping to a C++ CancelToken argument, which + # lets the method detect when its request was abandoned, either because + # the client disconnected or because the call’s promise was dropped. + # + # NOTE: Must be declared after a `mp.Context` parameter, which provides + # the worker thread the method will run on. +} diff --git a/src/mp/gen.cpp b/src/mp/gen.cpp index 7733eba6..d6d1e635 100644 --- a/src/mp/gen.cpp +++ b/src/mp/gen.cpp @@ -40,6 +40,8 @@ constexpr uint64_t INCLUDE_TYPES_ANNOTATION_ID = 0xbcec15648e8a0cf1ull; // From constexpr uint64_t WRAP_ANNOTATION_ID = 0xe6f46079b7b1405eull; // From proxy.capnp constexpr uint64_t COUNT_ANNOTATION_ID = 0xd02682b319f69b38ull; // From proxy.capnp constexpr uint64_t EXCEPTION_ANNOTATION_ID = 0x996a183200992f88ull; // From proxy.capnp +constexpr uint64_t CONTEXT_STRUCT_ID = 0x9c44e6645d0b22c6ull; // From proxy.capnp +constexpr uint64_t CANCEL_STRUCT_ID = 0xdb61673d0c7481c5ull; // From proxy.capnp constexpr uint64_t NAME_ANNOTATION_ID = 0xb594888f63f4dbb9ull; // From proxy.capnp constexpr uint64_t SKIP_ANNOTATION_ID = 0x824c08b82695d8ddull; // From proxy.capnp @@ -533,7 +535,18 @@ static void Generate(kj::StringPtr src_prefix, const bool is_destroy = method_name == kj::StringPtr{"destroy"}; FieldList fields; + bool has_context = false; for (const auto schema_field : method.getParamType().getFields()) { + if (const auto type = schema_field.getType(); type.isStruct()) { + const uint64_t type_id = type.asStruct().getProto().getId(); + if (type_id == CONTEXT_STRUCT_ID) has_context = true; + if (type_id == CANCEL_STRUCT_ID && !has_context) { + // `mp.Cancel` requires a preceding `mp.Context` so the method runs on + // a worker thread, not the event loop thread where cancellations are delivered. + throw std::runtime_error( + method_prefix + ": mp.Cancel parameter requires a preceding mp.Context parameter"); + } + } fields.addField(schema_field, true, false); } for (const auto schema_field : method.getResultType().getFields()) { From 3f6b324c6aae9722a947e514593a55bd85f4c7ef Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Tue, 11 Aug 2026 02:37:45 -0300 Subject: [PATCH 2/6] proxy: allow capnp to cancel calls to methods with an `mp.Cancel` parameter By default, the RPC system runs calls to completion even after the client abandons them. This commit overrides `dispatchCall` in `ProxyServerBase` to allow cancellation of calls to methods that declare an `mp.Cancel` parameter, detected from the interface schema. The lookup mirrors capnp's own dynamic dispatch: https://github.com/capnproto/capnproto/blob/7dbb95989721016f8b590245ec7528c6ff03d1fe/c++/src/capnp/dynamic-capability.c++#L61-L67 On Cap'n Proto v1.0+, this sets the `allowCancellation` flag on the dispatch result, older versions use `CallContext::allowCancellation()` (see the "Breaking change" note in https://capnproto.org/news/2023-07-28-capnproto-1.0.html). This is the automatic equivalent of the `$Cxx.allowCancellation` annotation, without requiring a schema annotation. --- include/mp/proxy-io.h | 37 +++++++++++++++++++++++++++++++++++++ include/mp/proxy.h | 23 +++++++++++++++++++++++ 2 files changed, 60 insertions(+) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index cda9064d..2125ab27 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -10,7 +10,9 @@ #include +#include #include +#include #include #include @@ -23,6 +25,7 @@ #include #include #include +#include #include #include @@ -632,6 +635,40 @@ ProxyServerBase::ProxyServerBase(std::shared_ptr impl, Co assert(m_impl); } +template +auto ProxyServerBase::dispatchCall( + uint64_t interface_id, uint16_t method_id, DispatchContext context) -> DispatchCallResult +{ + // A method allows cancellation if any of its parameters in the interface + // schema is an `mp.Cancel` struct. This lookup mirrors + // `capnp::DynamicCapability::Server::dispatchCall`. + bool allow_cancel{false}; + KJ_IF_MAYBE(iface, capnp::Schema::from().findSuperclass(interface_id)) { + auto methods{iface->getMethods()}; + if (method_id < methods.size()) { + for (const auto field : methods[method_id].getParamType().getFields()) { + const auto type{field.getType()}; + if (type.isStruct() && type.asStruct().getProto().getId() == capnp::typeId()) allow_cancel = true; + } + } + } + + // Cap'n Proto 1.0 replaced the dynamic `CallContext::allowCancellation()` + // opt-in with a static flag on the dispatch result, so the code path + // depends on the Cap'n Proto version. See the "Breaking change" note in + // https://capnproto.org/news/2023-07-28-capnproto-1.0.html +#if CAPNP_VERSION < 1000000 + if (allow_cancel) context.allowCancellation(); +#endif + + auto result{Interface::Server::dispatchCall(interface_id, method_id, kj::mv(context))}; + +#if CAPNP_VERSION >= 1000000 + if (allow_cancel) result.allowCancellation = true; +#endif + return result; +} + //! ProxyServer destructor, called from the EventLoop thread by Cap'n Proto //! garbage collection code after there are no more references to this object. //! This will typically happen when the corresponding ProxyClient object on the diff --git a/include/mp/proxy.h b/include/mp/proxy.h index 2144b571..07d6a994 100644 --- a/include/mp/proxy.h +++ b/include/mp/proxy.h @@ -12,14 +12,22 @@ #include #include #include +#include #include #include #include #include // IWYU pragma: keep +namespace capnp { +class AnyPointer; +template +class CallContext; +} // namespace capnp + namespace mp { class Connection; class EventLoop; + //! Mapping from capnp interface type to proxy client implementation (specializations are generated by //! proxy-codegen.cpp). template struct ProxyClient; // IWYU pragma: export @@ -157,6 +165,21 @@ struct ProxyServerBase : public virtual Interface_::Server void invokeDestroy(); using Interface_::Server::thisCap; + // Type aliases used by the `dispatchCall` signature below. + using DispatchCallResult = typename Interface_::Server::DispatchCallResult; + using DispatchContext = ::capnp::CallContext<::capnp::AnyPointer, ::capnp::AnyPointer>; + /** + * Override of the capnp-generated `dispatchCall` method, which the RPC + * system uses to route incoming requests to the corresponding method. + * + * By default, capnp runs each call to completion even after the + * client abandons it, so this override also permits cancellation for + * methods that declare an `mp.Cancel` parameter. + * + * Every generated `ProxyServer` inherits this class, so one override + * covers all interfaces. */ + DispatchCallResult dispatchCall(uint64_t interface_id, uint16_t method_id, DispatchContext context) override; + /** * Implementation pointer that may or may not be owned and deleted when this * capnp server goes out of scope. It is owned for servers created to wrap From fe1004c3c69cb21fd75d28b8e4d0cc870e87b861 Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Tue, 11 Aug 2026 15:36:32 -0300 Subject: [PATCH 3/6] util: add `CancelState`, `CancelToken`, and `OnCancel` classes `CancelState` holds a cancellation flag and callback registry shared between an executing IPC method and the thread canceling its request. `CancelToken` is the handle a method polls to detect cancellation, and `OnCancel` registers an RAII callback that can wake a method blocked in a wait. The three classes have the same semantics as `std::stop_source`, `std::stop_token`, and `std::stop_callback`, which are not available on all supported platforms. --- include/mp/util.h | 77 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 77 insertions(+) diff --git a/include/mp/util.h b/include/mp/util.h index 30742014..08f720e5 100644 --- a/include/mp/util.h +++ b/include/mp/util.h @@ -6,6 +6,7 @@ #define MP_UTIL_H #include +#include #include #include #include @@ -357,6 +358,82 @@ struct InterruptException final : std::exception { std::string m_message; }; +//! Cancellation state shared between a CancelToken held by an executing method +//! and the thread canceling the request. This class, together with the ones +//! below, has the same semantics as std::stop_source/stop_token/stop_callback, +//! which solve the same problem. +class CancelState +{ +public: + //! Whether the request has been canceled. + bool canceled() const { return m_canceled; } + + //! Mark the request as canceled and invoke registered callbacks. + void cancel() + { + if (m_canceled.exchange(true)) return; + const Lock lock{m_mutex}; + for (const std::function* fn : m_callbacks) + (*fn)(); + } + + //! Register a callback to run when cancel() is called. If the request has + //! already been canceled, invoke it immediately so cancellation is not missed. + void Register(const std::function& fn) + { + const Lock lock{m_mutex}; + if (m_canceled) { + fn(); + return; + } + m_callbacks.push_back(&fn); + } + + //! Remove a callback. + void Unregister(const std::function& fn) + { + const Lock lock{m_mutex}; + std::erase(m_callbacks, &fn); + } + + std::atomic m_canceled{false}; + Mutex m_mutex; + std::vector*> m_callbacks MP_GUARDED_BY(m_mutex); +}; + +//! Handle that a method polls to detect whether its request has been +//! abandoned. It does not own its `CancelState` and must not outlive it. +class CancelToken +{ +public: + CancelToken() = default; // Construct a token that is never canceled. + explicit CancelToken(CancelState* state) : m_state{state} {} + + bool canceled() const { return m_state && m_state->canceled(); } + + CancelState* m_state{nullptr}; +}; + +//! RAII registration for a cancellation callback. The callback runs on the +//! event loop thread when the request is canceled. +class OnCancel +{ +public: + OnCancel(CancelToken token, std::function fn) : m_state{token.m_state}, m_fn{std::move(fn)} + { + if (m_state) m_state->Register(m_fn); + } + ~OnCancel() + { + if (m_state) m_state->Unregister(m_fn); + } + OnCancel(const OnCancel&) = delete; + OnCancel& operator=(const OnCancel&) = delete; + + CancelState* m_state; + const std::function m_fn; +}; + class CancelProbe; //! Helper class that detects when a promise is canceled. Used to detect From d0543092c4e3ae4eb351e32533652ad88593d68b Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Tue, 11 Aug 2026 18:03:56 -0300 Subject: [PATCH 4/6] proxy: rename `cancel_lock` and `cancel_mutex` to `request_lock` and `request_mutex` The mutex guards the request's params and results structs, not the cancellation itself. The old names predate the CancelState class added in the previous commit and would be confusing next to it. Pure rename, no behavior change. --- include/mp/proxy-io.h | 4 ++-- include/mp/proxy-types.h | 6 +++--- include/mp/type-context.h | 20 ++++++++++---------- 3 files changed, 15 insertions(+), 15 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 2125ab27..2a7232a6 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -60,11 +60,11 @@ struct ServerInvokeContext : InvokeContext //! results structs if the request is canceled while the worker thread is //! reading params (`call_context.getParams()`) or writing results //! (`call_context.getResults()`). - Lock* cancel_lock{nullptr}; + Lock* request_lock{nullptr}; //! For IPC methods that execute asynchronously, not on the event-loop //! thread, this is set to true if the IPC call was canceled by the client //! or canceled by a disconnection. If the call runs on the event-loop - //! thread, it can't be canceled. This should be accessed with cancel_lock + //! thread, it can't be canceled. This should be accessed with request_lock //! held if it is not null, since in the asynchronous case it is accessed //! from multiple threads. bool request_canceled{false}; diff --git a/include/mp/proxy-types.h b/include/mp/proxy-types.h index 02e8eefb..a698d782 100644 --- a/include/mp/proxy-types.h +++ b/include/mp/proxy-types.h @@ -521,14 +521,14 @@ struct ServerCall template decltype(auto) invoke(ServerContext& server_context, TypeList<>, Args&&... args) const { - // If cancel_lock is set, release it while executing the method, and + // If request_lock is set, release it while executing the method, and // reacquire it afterwards. The lock is needed to prevent params and // response structs from being deleted by the event loop thread if the // request is canceled, so it is only needed before and after method // execution. It is important to release the lock during execution // because the method can take arbitrarily long to return and the event // loop will need the lock itself in on_cancel if the call is canceled. - if (server_context.cancel_lock) server_context.cancel_lock->m_lock.unlock(); + if (server_context.request_lock) server_context.request_lock->m_lock.unlock(); return TryFinally( [&]() -> decltype(auto) { return ProxyServerMethodTraits< @@ -536,7 +536,7 @@ struct ServerCall >::invoke(server_context, std::forward(args)...); }, [&] { - if (server_context.cancel_lock) server_context.cancel_lock->m_lock.lock(); + if (server_context.request_lock) server_context.request_lock->m_lock.lock(); // If the IPC request was canceled, throw InterruptException // because there is no point continuing and trying to fill the // call_context.getResults() struct. It's also important to stop diff --git a/include/mp/type-context.h b/include/mp/type-context.h index 54007207..538e93f6 100644 --- a/include/mp/type-context.h +++ b/include/mp/type-context.h @@ -97,9 +97,9 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& auto& request_threads = thread_context.request_threads; ConnThread request_thread; bool inserted{false}; - Mutex cancel_mutex; - Lock cancel_lock{cancel_mutex}; - server_context.cancel_lock = &cancel_lock; + Mutex request_mutex; + Lock request_lock{request_mutex}; + server_context.request_lock = &request_lock; loop.sync([&] { // Detect request being canceled before it executes. if (cancel_monitor.m_canceled) { @@ -108,9 +108,9 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& } // Detect request being canceled while it executes. assert(!cancel_monitor.m_on_cancel); - cancel_monitor.m_on_cancel = [&loop, &server_context, &cancel_mutex, req]() { + cancel_monitor.m_on_cancel = [&loop, &server_context, &request_mutex, req]() { MP_LOG(loop, Log::Info) << "IPC server request #" << req << " canceled while executing."; - // Lock cancel_mutex here to block the event loop + // Lock request_mutex here to block the event loop // thread and prevent it from deleting the request's // params and response structs while the execution // thread is accessing them. Because this lock is @@ -121,7 +121,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& // it. So in addition to locking the mutex, the // execution thread always checks request_canceled // as well before accessing the structs. - Lock cancel_lock{cancel_mutex}; + Lock request_lock{request_mutex}; server_context.request_canceled = true; }; // Update requests_threads map if not canceled. We know @@ -140,14 +140,14 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& // makes another IPC call), so avoid modifying the map. const bool erase_thread{inserted}; KJ_DEFER( - // Release the cancel lock before calling loop->sync and + // Release the request lock before calling loop->sync and // waiting for the event loop thread, because if a // cancellation happened, it needs to run the on_cancel - // callback above. It's safe to release cancel_lock at + // callback above. It's safe to release request_lock at // this point because the fn.invoke() call below will be // finished and no longer accessing the params or // results structs. - cancel_lock.m_lock.unlock(); + request_lock.m_lock.unlock(); // Erase the request_threads entry on the event loop // thread with loop->sync(), so if the connection is // broken there is not a race between this thread and @@ -160,7 +160,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& // cancellation happened. So we do not need to be // notified of cancellations after this point. Also // we do not want to be notified because - // cancel_mutex and server_context could be out of + // request_mutex and server_context could be out of // scope when it happens. cancel_monitor.m_on_cancel = nullptr; auto self_dispose{kj::mv(self)}; From f3d849c306c4720e93dd3be8e2aab42ab24e22a8 Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Tue, 11 Aug 2026 22:48:43 -0300 Subject: [PATCH 5/6] proxy: use `CancelState` for request cancellation Replace `CancelMonitor`'s `m_canceled` and `m_on_cancel` members with a `CancelState` member, and `ServerInvokeContext`'s `request_canceled` member with a `cancel_state` pointer and a `request_canceled()` helper that reads it. Behavior is unchanged. This prepares for the next commit, where methods with an `mp.Cancel` parameter observe the same state through a `CancelToken`. --- include/mp/proxy-io.h | 16 +++++++++------- include/mp/proxy-types.h | 2 +- include/mp/type-context.h | 32 ++++++++++++-------------------- include/mp/util.h | 16 ++++++---------- 4 files changed, 28 insertions(+), 38 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 2a7232a6..d82fd836 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -61,13 +61,15 @@ struct ServerInvokeContext : InvokeContext //! reading params (`call_context.getParams()`) or writing results //! (`call_context.getResults()`). Lock* request_lock{nullptr}; - //! For IPC methods that execute asynchronously, not on the event-loop - //! thread, this is set to true if the IPC call was canceled by the client - //! or canceled by a disconnection. If the call runs on the event-loop - //! thread, it can't be canceled. This should be accessed with request_lock - //! held if it is not null, since in the asynchronous case it is accessed - //! from multiple threads. - bool request_canceled{false}; + //! The request's cancellation state, owned by its `CancelMonitor`. Null + //! for methods executing on the event-loop thread. + CancelState* cancel_state{nullptr}; + //! Whether the IPC call was canceled by the client or by a disconnection. + //! Calls running on the event-loop thread cannot be canceled, so this + //! always returns false there. A false return is only reliable while + //! request_lock is held, which prevents the request's params and results + //! structs from being freed after the check. + bool request_canceled() const { return cancel_state && cancel_state->canceled(); } ServerInvokeContext(ProxyServer& proxy_server, CallContext& call_context, int req) : InvokeContext{*proxy_server.m_context.connection}, proxy_server{proxy_server}, call_context{call_context}, req{req} diff --git a/include/mp/proxy-types.h b/include/mp/proxy-types.h index a698d782..4726bc80 100644 --- a/include/mp/proxy-types.h +++ b/include/mp/proxy-types.h @@ -549,7 +549,7 @@ struct ServerCall // Since the call has been canceled that exception can't be // returned to the caller, so it needs to be discarded like // other result values. - if (server_context.request_canceled) throw InterruptException{"canceled"}; + if (server_context.request_canceled()) throw InterruptException{"canceled"}; }); } }; diff --git a/include/mp/type-context.h b/include/mp/type-context.h index 538e93f6..18c50693 100644 --- a/include/mp/type-context.h +++ b/include/mp/type-context.h @@ -10,6 +10,8 @@ #include +#include + namespace mp { template requires FieldTypeIs @@ -100,15 +102,15 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& Mutex request_mutex; Lock request_lock{request_mutex}; server_context.request_lock = &request_lock; + server_context.cancel_state = &cancel_monitor.m_state; + // Cancellation callback, registered in the loop.sync call below + // after checking that the request was not canceled early. + std::optional on_cancel; loop.sync([&] { // Detect request being canceled before it executes. - if (cancel_monitor.m_canceled) { - server_context.request_canceled = true; - return; - } + if (cancel_monitor.m_state.canceled()) return; // Detect request being canceled while it executes. - assert(!cancel_monitor.m_on_cancel); - cancel_monitor.m_on_cancel = [&loop, &server_context, &request_mutex, req]() { + on_cancel.emplace(CancelToken{&cancel_monitor.m_state}, [&loop, &request_mutex, req] { MP_LOG(loop, Log::Info) << "IPC server request #" << req << " canceled while executing."; // Lock request_mutex here to block the event loop // thread and prevent it from deleting the request's @@ -119,14 +121,13 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& // protection from the event loop deleting the // structs _before_ the execution thread acquires // it. So in addition to locking the mutex, the - // execution thread always checks request_canceled + // execution thread always checks request_canceled() // as well before accessing the structs. Lock request_lock{request_mutex}; - server_context.request_canceled = true; - }; + }); // Update requests_threads map if not canceled. We know // the request is not canceled currently because - // cancel_monitor.m_canceled was checked above and this + // `cancel_monitor.m_state.canceled()` was checked above and this // code is running on the event loop thread. std::tie(request_thread, inserted) = SetThread( GuardedRef{thread_context.waiter->m_mutex, request_threads}, server.m_context.connection, @@ -154,15 +155,6 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& // the disconnect handler trying to destroy the thread // client object. loop.sync([&] { - // Clear cancellation callback. At this point the - // method invocation finished and the result is - // either being returned, or discarded if a - // cancellation happened. So we do not need to be - // notified of cancellations after this point. Also - // we do not want to be notified because - // request_mutex and server_context could be out of - // scope when it happens. - cancel_monitor.m_on_cancel = nullptr; auto self_dispose{kj::mv(self)}; if (erase_thread) { // Look up the thread again without using existing @@ -179,7 +171,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& } }); ); - if (server_context.request_canceled) { + if (server_context.request_canceled()) { MP_LOG(loop, Log::Info) << "IPC server request #" << req << " canceled before it could be executed"; } else KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]{ try { diff --git a/include/mp/util.h b/include/mp/util.h index 08f720e5..058e52d2 100644 --- a/include/mp/util.h +++ b/include/mp/util.h @@ -436,20 +436,17 @@ class OnCancel class CancelProbe; -//! Helper class that detects when a promise is canceled. Used to detect -//! canceled requests and prevent potential crashes on unclean disconnects. -//! -//! In the future, this could also be used to support a way for wrapped C++ -//! methods to detect cancellation (like approach #4 in -//! https://github.com/bitcoin/bitcoin/issues/33575). +//! Helper class that reports request cancellation when the promise executing +//! its IPC method is destroyed. Cap'n Proto abandons a call by destroying +//! that promise so the paired `CancelProbe` is attached to it, and its +//! destructor notifies this class. class CancelMonitor { public: inline ~CancelMonitor(); inline void promiseDestroyed(CancelProbe& probe); - bool m_canceled{false}; - std::function m_on_cancel; + CancelState m_state; CancelProbe* m_probe{nullptr}; }; @@ -486,8 +483,7 @@ void CancelMonitor::promiseDestroyed(CancelProbe& probe) // case because the CancelMonitor class is meant to be used inside code // fulfilling or rejecting the promise and destroyed before doing so. assert(m_probe == &probe); - m_canceled = true; - if (m_on_cancel) m_on_cancel(); + m_state.cancel(); m_probe = nullptr; } } // namespace mp From 0986a13ea66a4fab1a8f64de5d37815efaaa1afa Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Wed, 12 Aug 2026 03:01:00 -0300 Subject: [PATCH 6/6] Add `mp.Cancel` parameter type passing a `CancelToken` to IPC methods Add type-cancel.h with serialization overloads for `Cancel` parameters: - The client-side overload sends an empty field and logs a warning if a live token is passed. - The server-side overload passes the method a `CancelToken` observing the request's cancellation state. Additionally, add a test that checks dropping the client promise cancels an executing method, and document the feature in design.md. --- CMakeLists.txt | 1 + doc/design.md | 31 ++++++++++++++++++++++++++ include/mp/type-cancel.h | 42 +++++++++++++++++++++++++++++++++++ test/mp/test/foo-types.h | 1 + test/mp/test/foo.capnp | 1 + test/mp/test/foo.h | 8 +++++++ test/mp/test/test.cpp | 48 ++++++++++++++++++++++++++++++++++++++++ 7 files changed, 132 insertions(+) create mode 100644 include/mp/type-cancel.h diff --git a/CMakeLists.txt b/CMakeLists.txt index bf50018a..3d9cf2b0 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -155,6 +155,7 @@ set(MP_PUBLIC_HEADERS include/mp/proxy-io.h include/mp/proxy-types.h include/mp/proxy.h + include/mp/type-cancel.h include/mp/type-char.h include/mp/type-chrono.h include/mp/type-context.h diff --git a/doc/design.md b/doc/design.md index 319ee3de..6ad08091 100644 --- a/doc/design.md +++ b/doc/design.md @@ -210,6 +210,37 @@ Subsequent requests will reuse the existing thread capabilities held in `callbac - Posts the work lambda to that thread's queue via `waiter->post(invoke)` - Cleans up the `request_threads` entry +### Cancellation with Cancel + +Adding a `Cancel` parameter to a method that already has a `Context` parameter passes the C++ method a `CancelToken` argument that reports whether the request was abandoned, because the client disconnected, exited, or dropped the call's promise: + +```capnp +waitForValue @25 (context :Proxy.Context, minValue :Int32, cancel :Proxy.Cancel) -> (result :Int32); +``` + +```cpp +int FooImplementation::waitForValue(int min_value, CancelToken cancel) +{ + // Construct the callback before locking the mutex it uses. + const OnCancel wake{cancel, [this] { + const std::lock_guard lock{m_wait_mutex}; + m_wait_cv.notify_all(); + }}; + std::unique_lock lock{m_wait_mutex}; + m_wait_cv.wait(lock, [&] { return m_value >= min_value || cancel.canceled(); }); + return m_value; +} +``` + +Without this, a method blocked in a long wait keeps waiting after its client is gone, pinning its worker thread and delaying shutdown. libmultiprocess already discards results of canceled requests, so a method that observes `canceled()` can simply return. + +Notes: + +- The parameter carries no data. The client-side `CustomBuildField` sends an empty struct and ignores the client's `CancelToken` argument, which can be a default-constructed token (is never canceled). The server-side `PassField` constructs a token backed by the request's `CancelMonitor`. +- `Cancel` must be declared after a `Context` parameter. Without a worker thread the method would run on the event loop thread, which is also the thread that delivers cancellations, so the token could never fire while the method runs. The code generator reports an error for this. +- The server allows Cap'n Proto to cancel these calls when a client abandons them, equivalent to annotating the method with `$Cxx.allowCancellation`, but without requiring an annotation and while working on Cap'n Proto versions that predate it. +- `OnCancel` callbacks run on the event loop thread and must not block. Construct an `OnCancel` before locking any mutex its callback locks, and do not destroy it from inside its own callback. + ## Interface Definitions As explained in the [usage](usage.md) document, interface descriptions need to be consumed both by the _libmultiprocess_ code generator, and by C++ code that calls and implements the interfaces. The C++ code only needs to know about C++ arguments and return types, while the code generator only needs to know about capnp arguments and return types, but both need to know class and method names, so the corresponding `.h` and `.capnp` source files contain some of the same information, and have to be kept in sync manually when methods or parameters change. Despite the redundancy, reconciling the interface definitions is designed to be _straightforward_ and _safe_. _Straightforward_ because there is no need to write manual serialization code or use awkward intermediate types like [`UniValue`](https://github.com/bitcoin/bitcoin/blob/master/src/univalue/include/univalue.h) instead of native types. _Safe_ because if there are any inconsistencies between API and data definitions (even minor ones like using a narrow int data type for a wider int API input), there are errors at build time instead of errors or bugs at runtime. diff --git a/include/mp/type-cancel.h b/include/mp/type-cancel.h new file mode 100644 index 00000000..437dd7ff --- /dev/null +++ b/include/mp/type-cancel.h @@ -0,0 +1,42 @@ +// Copyright (c) The Bitcoin Core developers +// Distributed under the MIT software license, see the accompanying +// file COPYING or http://www.opensource.org/licenses/mit-license.php. + +#ifndef MP_PROXY_TYPE_CANCEL_H +#define MP_PROXY_TYPE_CANCEL_H + +#include +#include + +namespace mp { +//! CustomBuildField override for mp.Cancel arguments. The field is left +//! empty because the server constructs the token from request state. A live +//! token is ignored, so passing one probably means the caller wrongly +//! expects cancellation to be forwarded. +template + requires FieldTypeIs +void CustomBuildField(TypeList, + Priority<1>, + InvokeContext& invoke_context, + Value&& value, + Output&& output) +{ + if (value.m_state) { + MP_LOG(*invoke_context.connection.m_loop, Log::Warning) + << "Ignoring live CancelToken argument. Cancellation is not forwarded to the server."; + } + output.init(); +} + +//! PassField override for mp.Cancel arguments. Ignores the field and +//! passes the method a `CancelToken` that observes the request's cancellation +//! state, owned by its `CancelMonitor` and published in +//! `server_context.cancel_state` by the Context's PassField (see type-context.h). +template +void PassField(Priority<1>, TypeList, ServerContext& server_context, Fn&& fn, Args&&... args) +{ + fn.invoke(server_context, std::forward(args)..., CancelToken{server_context.cancel_state}); +} +} // namespace mp + +#endif // MP_PROXY_TYPE_CANCEL_H diff --git a/test/mp/test/foo-types.h b/test/mp/test/foo-types.h index 1bc6c523..20361b16 100644 --- a/test/mp/test/foo-types.h +++ b/test/mp/test/foo-types.h @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include diff --git a/test/mp/test/foo.capnp b/test/mp/test/foo.capnp index 18df85aa..fea067cc 100644 --- a/test/mp/test/foo.capnp +++ b/test/mp/test/foo.capnp @@ -37,6 +37,7 @@ interface FooInterface $Proxy.wrap("mp::test::FooImplementation") { callFnAsync @18 (context :Proxy.Context) -> (); callIntFnAsync @21 (context :Proxy.Context, arg :Int32) -> (result :Int32); passDataPointers @22 (arg :List(Data)) -> (result :List(Data)); + callCancelFnAsync @25 (context :Proxy.Context, cancel :Proxy.Cancel) -> (); } interface FooInit $Proxy.wrap("mp::test::FooInit") { diff --git a/test/mp/test/foo.h b/test/mp/test/foo.h index 01566c58..67cbcf6a 100644 --- a/test/mp/test/foo.h +++ b/test/mp/test/foo.h @@ -5,6 +5,8 @@ #ifndef MP_TEST_FOO_H #define MP_TEST_FOO_H +#include + #include #include #include @@ -100,8 +102,14 @@ class FooImplementation void callFn() { assert(m_fn); m_fn(); } void callFnAsync() { assert(m_fn); m_fn(); } int callIntFnAsync(int arg) { assert(m_int_fn); return m_int_fn(arg); } + void callCancelFnAsync(CancelToken cancel) + { + assert(m_cancel_fn); + m_cancel_fn(cancel); + } std::function m_fn; std::function m_int_fn; + std::function m_cancel_fn; }; } // namespace test diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 7af1365e..bd5abc83 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -31,6 +31,7 @@ #include #include #include +#include #include #include #include @@ -651,5 +652,52 @@ KJ_TEST("Call async IPC method without thread or pool errors correctly") KJ_EXPECT(error_thrown); } +KJ_TEST("Dropping the client promise cancels an executing Proxy.Cancel method") +{ + TestSetup setup; + constexpr std::chrono::seconds timeout{30}; + std::promise waiting; + std::promise done; + + // Install a function that blocks until its `CancelToken` fires, so + // cancellation is the only way out. + setup.server->m_impl->m_cancel_fn = [&](CancelToken cancel) { + std::mutex mutex; + std::condition_variable cv; + const OnCancel wake{cancel, [&] { + const std::lock_guard lock{mutex}; + cv.notify_all(); + }}; + std::unique_lock lock{mutex}; + waiting.set_value(); + cv.wait(lock, [&] { return cancel.canceled(); }); + done.set_value(); + }; + ProxyClient* foo{setup.client.get()}; + foo->initThreadMap(); + + // Build the request by hand, the way a non-C++ client would. A normal + // proxy call cannot be abandoned because `clientInvoke` blocks on it. + std::optional> remote; + foo->m_context.loop->sync([&] { + auto request{foo->m_client.callCancelFnAsyncRequest()}; + request.initContext().setThread( + foo->m_context.connection->m_thread_map.makeThreadRequest().send().getResult()); + remote.emplace(request.send()); + }); + KJ_REQUIRE(waiting.get_future().wait_for(timeout) == std::future_status::ready); + + auto done_future{done.get_future()}; + KJ_EXPECT(done_future.wait_for(std::chrono::seconds{0}) == std::future_status::timeout); + + // Abandon the call without disconnecting. + foo->m_context.loop->sync([&] { remote.reset(); }); + + KJ_EXPECT(done_future.wait_for(timeout) == std::future_status::ready); + + // Connection should be unaffected. + KJ_EXPECT(foo->add(1, 2) == 3); +} + } // namespace test } // namespace mp