From 88ea9fa0a791fa20f3bc21b85093aeab3d34c09c Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Sun, 23 Aug 2026 03:56:59 -0300 Subject: [PATCH 1/6] Allow wrapped C++ methods to take parameters that are not sent over RPC Add a `$Proxy.extraParam` method annotation that declares an extra C++-only parameter in the generated method signature. The parameter has no corresponding capnp parameter and is not serialized or sent over RPC. The annotation value names the parameter in generated C++ code. Client behavior: - If a matching `CustomBuildExtraParam(TypeList, ClientInvokeContext&, T&&)` overload exists, the parameter is passed to it. - Otherwise, the parameter is discarded before the RPC message is dispatched. Server behavior: - A matching `CustomReadExtraParam(TypeList, ServerContext&)` overload MUST be implemented. No data arrives for this parameter so this overload reconstructs the parameter value on the server side. Constraints: - Only one extra parameter is allowed per method. - The extra parameter is expected to be the last parameter in the C++ method signature. The test checks the value the client passes is discarded and the one the server reconstructs arrives instead. --- include/mp/proxy-types.h | 84 ++++++++++++++++++++++++++++++++++------ include/mp/proxy.capnp | 6 +++ src/mp/gen.cpp | 60 ++++++++++++++++++++++++++-- test/mp/test/foo-types.h | 6 +++ test/mp/test/foo.capnp | 1 + test/mp/test/foo.h | 1 + test/mp/test/test.cpp | 5 +++ 7 files changed, 149 insertions(+), 14 deletions(-) diff --git a/include/mp/proxy-types.h b/include/mp/proxy-types.h index 3a5dd3d8..451123e9 100644 --- a/include/mp/proxy-types.h +++ b/include/mp/proxy-types.h @@ -534,12 +534,70 @@ ClientParam MakeClientParam(Types&&... values) return {std::forward(values)...}; } +//! Client parameter with no capnp field, generated for method parameters +//! declared with `$Proxy.extraParam`. Its value is handed to +//! MaybeBuildExtraParam instead of being built into the request. +template +struct ClientParam +{ + ClientParam(Types&&... values) : m_values{std::forward(values)...} {} + + struct BuildParams : IterateFieldsHelper + { + template + void handleField(ClientInvokeContext& invoke_context, Params&, ParamList) + { + auto const fun = [&](Values&&... values) { + (MaybeBuildExtraParam(TypeList>(), invoke_context, std::forward(values)), ...); + }; + + std::apply(fun, std::move(m_client_param->m_values)); + } + BuildParams(ClientParam* client_param) : m_client_param(client_param) {} + ClientParam* m_client_param; + }; + + struct ReadResults : IterateFieldsHelper + { + template + void handleField(ClientInvokeContext&, Results&, ParamList) + { + } + ReadResults(ClientParam*) {} + }; + + std::tuple m_values; +}; + +template +void MaybeBuildExtraParam(TypeList param, ClientInvokeContext& invoke_context, Value&& value) +{ + if constexpr (requires { CustomBuildExtraParam(param, invoke_context, std::forward(value)); }) { + CustomBuildExtraParam(param, invoke_context, std::forward(value)); + } +} + +template +LocalType MaybeReadExtraParam(TypeList param, ServerContext& server_context) +{ + static_assert(requires { CustomReadExtraParam(param, server_context); }, + "Wrapped C++ method has more parameters than its corresponding Cap'n Proto method has fields. " + "Declare extra parameters with $Proxy.extraParam in the Cap'n Proto schema and add a matching " + "`CustomReadExtraParam` overload."); + return CustomReadExtraParam(param, server_context); +} + struct ServerCall { // FIXME: maybe call call_context.releaseParams() - template - decltype(auto) invoke(ServerContext& server_context, TypeList<>, Args&&... args) const + template + decltype(auto) invoke(ServerContext& server_context, TypeList, Args&&... args) const { + // Construct the extra parameters before cancel_lock is released below. + // CustomReadExtraParam overloads build values from the request being + // executed, so they need the same protection as normal capnp fields + // from the event loop deleting request state on cancellation. + std::tuple...> extra{MaybeReadExtraParam(TypeList>(), server_context)...}; // If cancel_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 @@ -550,9 +608,13 @@ struct ServerCall if (server_context.cancel_lock) server_context.cancel_lock->m_lock.unlock(); return TryFinally( [&]() -> decltype(auto) { - return ProxyServerMethodTraits< - typename decltype(server_context.call_context.getParams())::Reads - >::invoke(server_context, std::forward(args)...); + return std::apply( + [&](RemoveCvRef&... extra_args) -> decltype(auto) { + return ProxyServerMethodTraits< + typename decltype(server_context.call_context.getParams())::Reads + >::invoke(server_context, std::forward(args)..., extra_args...); + }, + extra); }, [&] { if (server_context.cancel_lock) server_context.cancel_lock->m_lock.lock(); @@ -587,10 +649,10 @@ struct ServerRet : Parent { ServerRet(Parent parent) : Parent(parent) {} - template - void invoke(ServerContext& server_context, TypeList<>, Args&&... args) const + template + void invoke(ServerContext& server_context, ArgTypes arg_types, Args&&... args) const { - auto&& result = Parent::invoke(server_context, TypeList<>(), std::forward(args)...); + auto&& result = Parent::invoke(server_context, arg_types, std::forward(args)...); auto&& results = server_context.call_context.getResults(); InvokeContext& invoke_context = server_context; BuildField(TypeList(), invoke_context, Make(results), @@ -603,11 +665,11 @@ struct ServerExcept : Parent { ServerExcept(Parent parent) : Parent(parent) {} - template - void invoke(ServerContext& server_context, TypeList<>, Args&&... args) const + template + void invoke(ServerContext& server_context, ArgTypes arg_types, Args&&... args) const { try { - return Parent::invoke(server_context, TypeList<>(), std::forward(args)...); + return Parent::invoke(server_context, arg_types, std::forward(args)...); } catch (const Exception& exception) { auto&& results = server_context.call_context.getResults(); BuildField(TypeList(), server_context, Make(results), exception); diff --git a/include/mp/proxy.capnp b/include/mp/proxy.capnp index e0a66fd9..9c7e1090 100644 --- a/include/mp/proxy.capnp +++ b/include/mp/proxy.capnp @@ -40,6 +40,12 @@ annotation name(field, method): Text; annotation skip(field): Void; # Synonym for count(0). +annotation extraParam(method): Text; +# Adds an extra C++-only parameter to the generated method signature. +# +# This parameter has no corresponding capnp parameter and is not serialized or +# sent over RPC. The annotation value names the parameter in generated C++ code. + interface ThreadMap $count(0) { # Interface letting clients control which thread a method call should # execute on. Clients create and name threads and pass the thread handle as diff --git a/src/mp/gen.cpp b/src/mp/gen.cpp index 8d733a91..af86cb75 100644 --- a/src/mp/gen.cpp +++ b/src/mp/gen.cpp @@ -16,6 +16,7 @@ #include #include #include +#include #include #include #include @@ -42,6 +43,7 @@ constexpr uint64_t COUNT_ANNOTATION_ID = 0xd02682b319f69b38ull; // From prox constexpr uint64_t EXCEPTION_ANNOTATION_ID = 0x996a183200992f88ull; // From proxy.capnp constexpr uint64_t NAME_ANNOTATION_ID = 0xb594888f63f4dbb9ull; // From proxy.capnp constexpr uint64_t SKIP_ANNOTATION_ID = 0x824c08b82695d8ddull; // From proxy.capnp +constexpr uint64_t EXTRA_PARAM_ANNOTATION_ID = 0xf1731507694af05aull; // From proxy.capnp template static bool AnnotationExists(const Reader& reader, uint64_t id) @@ -136,6 +138,9 @@ struct Field bool requested = false; bool skip = false; kj::StringPtr exception; + //! Name of a $Proxy.extraParam parameter, set only on entries that have no + //! capnp field behind them. + kj::StringPtr extra_name; }; struct FieldList @@ -542,6 +547,42 @@ static void Generate(kj::StringPtr src_prefix, } fields.mergeFields(); + kj::StringPtr extra_param; + bool has_extra{GetAnnotationText(method.getProto(), EXTRA_PARAM_ANNOTATION_ID, &extra_param)}; + if (has_extra) { + if (is_construct || is_destroy) { + throw std::runtime_error(method_prefix + + ": $Proxy.extraParam is not supported on construct and destroy methods, which have no " + "corresponding C++ method"); + } + int annotations_count{0}; + for (const auto annotation : method.getProto().getAnnotations()) { + if (annotation.getId() == EXTRA_PARAM_ANNOTATION_ID) ++annotations_count; + } + if (annotations_count > 1) { + throw std::runtime_error(method_prefix + ": only one $Proxy.extraParam per method is supported"); + } + if (!extra_param.size()) { + throw std::runtime_error(method_prefix + ": $Proxy.extraParam requires a parameter name"); + } + if (fields.field_idx.contains(extra_param)) { + throw std::runtime_error(method_prefix + ": $Proxy.extraParam name '" + extra_param.cStr() + + "' collides with a capnp field name"); + } + + // The extra entry goes right after the last field that + // consumes C++ arguments, so the parameter is last in the + // C++ method signature. + auto pos = fields.fields.begin(); + for (auto it = pos; it != fields.fields.end(); ++it) { + if (!it->skip && it->args > 0) pos = std::next(it); + } + Field field; + field.extra_name = extra_param; + field.args = 1; + fields.fields.insert(pos, field); + } + if (!is_construct && !is_destroy && (&method_interface == &interface)) { methods << "template<>\n"; methods << "struct ProxyMethod<" << method_prefix << "Params>\n"; @@ -559,9 +600,10 @@ static void Generate(kj::StringPtr src_prefix, for (const auto& field : fields.fields) { if (field.skip) continue; - const auto& f = field.param_is_set ? field.param : field.result; - auto field_name = f.getProto().getName(); - add_accessor(field_name); + const bool extra{field.extra_name.size() > 0}; + const auto field_name = + extra ? field.extra_name : (field.param_is_set ? field.param : field.result).getProto().getName(); + if (!extra) add_accessor(field_name); std::ostringstream fwd_args; for (int i = 0; i < field.args; ++i) { @@ -579,6 +621,12 @@ static void Generate(kj::StringPtr src_prefix, ++argc; } + + if (extra) { + // No capnp field, no accessor. + client_invoke << ", MakeClientParam(" << fwd_args.str() << ")"; + continue; + } client_invoke << ", "; if (field.exception.size()) { @@ -613,6 +661,12 @@ static void Generate(kj::StringPtr src_prefix, client << " using M" << method_ordinal << " = ProxyClientMethodTraits<" << method_prefix << "Params>;\n"; + if (has_extra) { + client << " static_assert(M" << method_ordinal << "::Params::size == " << argc + << ", \"C++ method " << proxied_class_type << "::" << proxied_method_name + << " should have " << argc << " parameters to match capnp method " << method_prefix + << ".\");\n"; + } client << " " << static_str << "typename M" << method_ordinal << "::Result " << method_name << "(" << super_str << client_args.str() << ")"; client << ";\n"; diff --git a/test/mp/test/foo-types.h b/test/mp/test/foo-types.h index 85c36349..39e4c12e 100644 --- a/test/mp/test/foo-types.h +++ b/test/mp/test/foo-types.h @@ -112,6 +112,12 @@ inline void CustomPassMessage(InvokeContext& invoke_context, fn(mut); builder.setMessage(mut.message + " return"); } + +template +int CustomReadExtraParam(TypeList, ServerContext& server_context) +{ + return 1; +} } // namespace mp #endif // MP_TEST_FOO_TYPES_H diff --git a/test/mp/test/foo.capnp b/test/mp/test/foo.capnp index 392719ec..55872ac4 100644 --- a/test/mp/test/foo.capnp +++ b/test/mp/test/foo.capnp @@ -39,6 +39,7 @@ interface FooInterface $Proxy.wrap("mp::test::FooImplementation") { passDataPointers @22 (arg :List(Data)) -> (result :List(Data)); listBars @25 (context :Proxy.Context, n :Int32) -> (result :List(BarInterface)); callMessageAsync @26 (context :Proxy.Context) -> (result :FooMessage); + passExtra @27 (arg :Int32) -> (result :Int32) $Proxy.extraParam("extra"); } interface FooInit $Proxy.wrap("mp::test::FooInit") { diff --git a/test/mp/test/foo.h b/test/mp/test/foo.h index e8782708..8bfd0474 100644 --- a/test/mp/test/foo.h +++ b/test/mp/test/foo.h @@ -112,6 +112,7 @@ class FooImplementation FooEnum passEnum(FooEnum foo) { return foo; } double passDouble(double value) { return value; } int passFn(std::function fn) { return fn(); } + int passExtra(int arg, int extra) { return arg + extra; } std::vector passDataPointers(std::vector values) { return values; } std::vector> listBars(int n) { diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 5ecb7cc4..13ea37cc 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -267,6 +267,11 @@ KJ_TEST("Call FooInterface methods") KJ_EXPECT(foo->passFn([]{ return 10; }) == 10); + // The `CustomReadExtraParam` overload in `foo-types.h` builds the + // server-side value, hardcoded to 1. As a result this always returns + // arg + 1 regardless of the value passed for extra. + KJ_EXPECT(foo->passExtra(1, 999) == 2); + // Recursive async IPC calls KJ_EXPECT(foo->passFn([foo]{ return foo->passFn([]{ return 1; }); From 4bac5cea258a2541cfc323acb1babec1e3be5f66 Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Sun, 23 Aug 2026 05:30:06 -0300 Subject: [PATCH 2/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 would be confusing next to the CancelState class added in the following commits. Pure rename, no behavior change. --- include/mp/proxy-io.h | 4 ++-- include/mp/proxy-types.h | 8 ++++---- include/mp/type-context.h | 18 +++++++++--------- test/mp/test/test.cpp | 2 +- 4 files changed, 16 insertions(+), 16 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index 49b0611a..a859b338 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -58,11 +58,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 451123e9..70a9c156 100644 --- a/include/mp/proxy-types.h +++ b/include/mp/proxy-types.h @@ -593,19 +593,19 @@ struct ServerCall template decltype(auto) invoke(ServerContext& server_context, TypeList, Args&&... args) const { - // Construct the extra parameters before cancel_lock is released below. + // Construct the extra parameters before request_lock is released below. // CustomReadExtraParam overloads build values from the request being // executed, so they need the same protection as normal capnp fields // from the event loop deleting request state on cancellation. std::tuple...> extra{MaybeReadExtraParam(TypeList>(), server_context)...}; - // 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 std::apply( @@ -617,7 +617,7 @@ struct ServerCall extra); }, [&] { - 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 7cfc7e79..fa65c929 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 @@ -155,11 +155,11 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& // Release the cancel 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 @@ -172,7 +172,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)}; diff --git a/test/mp/test/test.cpp b/test/mp/test/test.cpp index 13ea37cc..415b43f3 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -537,7 +537,7 @@ KJ_TEST("Calling async IPC method, with server disconnect after cleanup") // Use testing_hook_async_request_done to trigger a disconnect from the // worker thread after it executes an async request but before it returns. // Without the bugfix, the m_on_cancel callback would be called at this - // point, accessing the cancel_mutex stack variable that had gone out of + // point, accessing the request_mutex stack variable that had gone out of // scope. TestSetup setup; ProxyClient* foo = setup.client.get(); From 8d34bb09f070ecb14884a4583cb3e2cb551f3263 Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Sun, 23 Aug 2026 09:20:39 -0300 Subject: [PATCH 3/6] proxy: make client IPC calls cancelable Add `ClientCancelState` and `RequestCanceler`. `ClientCancelState` is created by `clientInvoke`, tracks whether the call was canceled, and can cancel the request promise from any thread. `RequestCanceler` inherits from `kj::Canceler`, wraps the request promise, and is attached to it. Canceling rejects the wrapped promise, wakes the blocked client thread through the exception path, and makes the call throw `InterruptException`. The next commit adds the `CustomBuildExtraParam` overload that lets callers cancel the call. Nothing triggers cancellation yet. --- include/mp/proxy-io.h | 61 ++++++++++++++++++++++++++++++++++++++++ include/mp/proxy-types.h | 12 ++++++-- src/mp/gen.cpp | 8 ++++++ 3 files changed, 78 insertions(+), 3 deletions(-) diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index a859b338..ca450d59 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -39,6 +39,7 @@ struct InvokeContext struct ClientInvokeContext : InvokeContext { ThreadContext& thread_context; + std::function)> cancel_receiver; ClientInvokeContext(Connection& conn, ThreadContext& thread_context) : InvokeContext{conn}, thread_context{thread_context} { @@ -383,6 +384,66 @@ class EventLoop std::function testing_hook_misc; }; +//! Cancellation state of one IPC call, created by clientInvoke. +class ClientCancelState +{ +public: + explicit ClientCancelState(EventLoop& loop) : m_loop(loop) {} + + //! Cancel the in-flight request, waking the blocked client thread. + //! Callable from any thread. + inline void cancel(); + + //! Whether cancel was called. + bool canceled() + { + const Lock lock{m_mutex}; + return m_canceled; + } + + //! Keeps the event loop alive while the caller holds the cancel + //! function. + EventLoopRef m_loop; + Mutex m_mutex; + bool m_canceled MP_GUARDED_BY(m_mutex){false}; + //! Canceler of the request promise, owned by the RequestCanceler attached + //! to it. Null before the request is sent and after it completes. + kj::Canceler* m_canceler MP_GUARDED_BY(m_mutex){nullptr}; +}; + +//! kj::Canceler wrapping a request promise. Attached to the promise so it +//! is created and destroyed on the event loop thread, keeping +//! ClientCancelState::m_canceler valid while the request is in flight. +struct RequestCanceler : kj::Canceler +{ + explicit RequestCanceler(std::shared_ptr state) : m_state(std::move(state)) + { + const Lock lock{m_state->m_mutex}; + m_state->m_canceler = this; + } + ~RequestCanceler() + { + const Lock lock{m_state->m_mutex}; + m_state->m_canceler = nullptr; + } + std::shared_ptr m_state; +}; + +void ClientCancelState::cancel() +{ + { + const Lock lock{m_mutex}; + if (m_canceled) return; + m_canceled = true; + // Null canceler means the call already completed, so there is nothing to cancel. + if (!m_canceler) return; + } + m_loop->sync([&] { + const Lock lock{m_mutex}; + if (m_canceler) m_canceler->cancel("canceled by client"); + }); +} + //! Single element task queue used to handle recursive capnp calls. (If the //! server makes a callback into the client in the middle of a request, while the client //! thread is blocked waiting for server response, this is what allows the diff --git a/include/mp/proxy-types.h b/include/mp/proxy-types.h index 70a9c156..a27f8be3 100644 --- a/include/mp/proxy-types.h +++ b/include/mp/proxy-types.h @@ -794,7 +794,9 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel std::exception_ptr exception; std::string kj_exception; bool done = false; + bool canceled = false; const char* disconnected = nullptr; + const auto cancel_state = std::make_shared(*proxy_client.m_context.loop); proxy_client.m_context.loop->sync([&]() { if (!proxy_client.m_context.connection) { const Lock lock(thread_context.waiter->m_mutex); @@ -815,7 +817,8 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel MP_LOGPLAIN(*proxy_client.m_context.loop, Log::Trace) << "send data: " << LogEscape(request.toString(), proxy_client.m_context.loop->m_log_opts.max_chars); - proxy_client.m_context.loop->m_task_set->add(request.send().then( + auto request_canceler = kj::heap(cancel_state); + proxy_client.m_context.loop->m_task_set->add(request_canceler->wrap(request.send()).then( [&](::capnp::Response&& response) { MP_LOGPLAIN(*proxy_client.m_context.loop, Log::Debug) << "{" << thread_context.thread_name << "} IPC client recv " @@ -833,7 +836,9 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel thread_context.waiter->m_cv.notify_all(); }, [&](const ::kj::Exception& e) { - if (e.getType() == ::kj::Exception::Type::DISCONNECTED) { + if (cancel_state->canceled()) { + canceled = true; + } else if (e.getType() == ::kj::Exception::Type::DISCONNECTED) { disconnected = "IPC client method call interrupted by disconnect."; } else { kj_exception = kj::str("kj::Exception: ", e).cStr(); @@ -843,12 +848,13 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel const Lock lock(thread_context.waiter->m_mutex); done = true; thread_context.waiter->m_cv.notify_all(); - })); + }).attach(kj::mv(request_canceler))); }); Lock lock(thread_context.waiter->m_mutex); thread_context.waiter->wait(lock, [&done]() { return done; }); if (exception) std::rethrow_exception(exception); + if (canceled) throw InterruptException{"canceled"}; if (!kj_exception.empty()) MP_LOGPLAIN(*proxy_client.m_context.loop, Log::Raise) << kj_exception; if (disconnected) MP_LOGPLAIN(*proxy_client.m_context.loop, Log::Raise) << disconnected; } diff --git a/src/mp/gen.cpp b/src/mp/gen.cpp index af86cb75..bc5e56e5 100644 --- a/src/mp/gen.cpp +++ b/src/mp/gen.cpp @@ -320,6 +320,13 @@ static void Generate(kj::StringPtr src_prefix, std::ofstream cpp_client(output_path + ".proxy-client.c++"); cpp_client << "// Generated by " PROXY_BIN " from " << src_file << "\n\n"; + cpp_client << "// GCC reports false positives in Cap'n Proto's kj::Maybe and\n"; + cpp_client << "// kj::_::ExceptionOr internals when clientInvoke wraps request\n"; + cpp_client << "// promises in a kj::Canceler. The pragma must precede the\n"; + cpp_client << "// includes to cover warnings reported at header locations.\n"; + cpp_client << "#if defined(__GNUC__) && !defined(__clang__)\n"; + cpp_client << "#pragma GCC diagnostic ignored \"-Wmaybe-uninitialized\"\n"; + cpp_client << "#endif\n\n"; cpp_client << "// IWYU pragma: no_include \n"; cpp_client << "// IWYU pragma: no_include \n"; cpp_client << "// IWYU pragma: begin_keep\n"; @@ -332,6 +339,7 @@ static void Generate(kj::StringPtr src_prefix, cpp_client << "#include \n"; cpp_client << "#include \n"; cpp_client << "#include \n"; + cpp_client << "#include \n"; cpp_client << "#include \n"; cpp_client << "#include \n"; cpp_client << "#include \n"; From b377c34c579355f23f8795fe6ca2fcd5553ff7fc Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Mon, 24 Aug 2026 14:01:20 -0300 Subject: [PATCH 4/6] proxy: support cancellation extra parameters Add type-cancel.h, which defines the cancellation argument types and their extra-parameter overloads. - On the client side, any `std::function)>` declared with `$Proxy.extraParam` receives a function that cancels the request. - On the server side, it registers a callback to run when cancellation is detected. --- CMakeLists.txt | 1 + include/mp/proxy-io.h | 8 ++++++ include/mp/proxy-types.h | 9 ++++++- include/mp/type-cancel.h | 57 +++++++++++++++++++++++++++++++++++++++ include/mp/type-context.h | 2 ++ include/mp/util.h | 10 +++---- 6 files changed, 80 insertions(+), 7 deletions(-) create mode 100644 include/mp/type-cancel.h diff --git a/CMakeLists.txt b/CMakeLists.txt index c2bfcb41..0c51de81 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -157,6 +157,7 @@ set(MP_PUBLIC_HEADERS include/mp/proxy.h include/mp/type-char.h include/mp/type-chrono.h + include/mp/type-cancel.h include/mp/type-context.h include/mp/type-data.h include/mp/type-decay.h diff --git a/include/mp/proxy-io.h b/include/mp/proxy-io.h index ca450d59..4f8dbb71 100644 --- a/include/mp/proxy-io.h +++ b/include/mp/proxy-io.h @@ -61,12 +61,20 @@ struct ServerInvokeContext : InvokeContext //! (`call_context.getResults()`). Lock* request_lock{nullptr}; //! For IPC methods that execute asynchronously, not on the event-loop + //! thread: mutex request_lock refers to, set together with it. Null for + //! methods executing on the event-loop thread. + Mutex* request_mutex{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}; + //! For IPC methods that execute asynchronously, not on the event-loop + //! thread: callback registered by a wrapped method. Runs when request + //! cancellation is detected. + std::function cancel_fn; 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 a27f8be3..0d5686bf 100644 --- a/include/mp/proxy-types.h +++ b/include/mp/proxy-types.h @@ -612,12 +612,15 @@ struct ServerCall [&](RemoveCvRef&... extra_args) -> decltype(auto) { return ProxyServerMethodTraits< typename decltype(server_context.call_context.getParams())::Reads - >::invoke(server_context, std::forward(args)..., extra_args...); + >::invoke(server_context, std::forward(args)..., std::move(extra_args)...); }, extra); }, [&] { if (server_context.request_lock) server_context.request_lock->m_lock.lock(); + // The method returned, so destroy the callback it registered + // through its cancellation argument, if any. + server_context.cancel_fn = nullptr; // 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 @@ -851,6 +854,10 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel }).attach(kj::mv(request_canceler))); }); + if (invoke_context && invoke_context->cancel_receiver) { + invoke_context->cancel_receiver([cancel_state] { cancel_state->cancel(); }); + } + Lock lock(thread_context.waiter->m_mutex); thread_context.waiter->wait(lock, [&done]() { return done; }); if (exception) std::rethrow_exception(exception); diff --git a/include/mp/type-cancel.h b/include/mp/type-cancel.h new file mode 100644 index 00000000..b6184634 --- /dev/null +++ b/include/mp/type-cancel.h @@ -0,0 +1,57 @@ +// 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 +#include + +#include +#include + +namespace mp { +//! Called when a request is canceled. +using CancelFn = std::function; + +//! Called to register a CancelFn that is called when a request is canceled. +//! On the client, calling the received CancelFn cancels the in-flight request. +//! On the server, the wrapped method receives its own CancelArg registering a +//! callback that runs if the request is canceled. +using CancelArg = std::function; + +//! Store the caller's CancelArg so clientInvoke can pass it a function +//! canceling the call once the request is sent. +inline void CustomBuildExtraParam(TypeList, ClientInvokeContext& invoke_context, CancelArg&& value) +{ + if (value) invoke_context.cancel_receiver = std::move(value); +} + +//! Builds the CancelArg passed to a wrapped method. It registers one +//! callback per request, run under request_mutex on the event loop, or +//! immediately on the worker thread if the request was already canceled. +template +CancelArg CustomReadExtraParam(TypeList, ServerContext& server_context) +{ + // request_mutex is set by mp.Context's PassField overload. + // If null, assume we're on the event loop thread, where cancellation + // signals are dispatched. + if (!server_context.request_mutex) { + return [](CancelFn) {}; + } + return [&server_context](CancelFn fn) { + { + const Lock lock{*server_context.request_mutex}; + if (!server_context.request_canceled) { + server_context.cancel_fn = std::move(fn); + return; + } + } + fn(); + }; +} +} // namespace mp + +#endif // MP_PROXY_TYPE_CANCEL_H diff --git a/include/mp/type-context.h b/include/mp/type-context.h index fa65c929..63b3fdb8 100644 --- a/include/mp/type-context.h +++ b/include/mp/type-context.h @@ -100,6 +100,7 @@ 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.request_mutex = &request_mutex; loop.sync([&] { // Detect request being canceled before it executes. if (cancel_monitor.m_canceled) { @@ -123,6 +124,7 @@ auto PassField(Priority<1>, TypeList<>, ServerContext& server_context, const Fn& // as well before accessing the structs. Lock request_lock{request_mutex}; server_context.request_canceled = true; + if (server_context.cancel_fn) server_context.cancel_fn(); }; // Update requests_threads map if not canceled. We know // the request is not canceled currently because diff --git a/include/mp/util.h b/include/mp/util.h index 30742014..29fa5b8c 100644 --- a/include/mp/util.h +++ b/include/mp/util.h @@ -359,12 +359,10 @@ struct InterruptException final : std::exception { 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: From ffd098b3ffe1a6093b6d00114427fce831fc9292 Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Mon, 24 Aug 2026 15:49:39 -0300 Subject: [PATCH 5/6] build: require Cap'n Proto 1.0 The `$Cxx.allowCancellation` annotation used by the cancellation tests does not exist in older versions. Remove the configure-time checks that only covered them, and move the olddeps CI config to 1.0.0. --- CMakeLists.txt | 58 +---------------------------------------- ci/configs/olddeps.bash | 2 +- doc/install.md | 2 +- 3 files changed, 3 insertions(+), 59 deletions(-) diff --git a/CMakeLists.txt b/CMakeLists.txt index 0c51de81..f2199cb1 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -13,7 +13,7 @@ endif() include("cmake/compat_find.cmake") find_package(Threads REQUIRED) -find_package(CapnProto 0.9 NO_MODULE) +find_package(CapnProto 1.0 NO_MODULE) if(NOT CapnProto_FOUND) message(FATAL_ERROR "Cap'n Proto is required but was not found.\n" @@ -23,62 +23,6 @@ find_package(CapnProto 0.9 NO_MODULE) ) endif() -# Cap'n Proto compatibility checks -set(CAPNPROTO_ISSUES "") -set(CAPNPROTO_CVE_AFFECTED FALSE) -set(CAPNPROTO_CLANG_INCOMPATIBLE FALSE) - -# Check for list-of-pointers memory access bug from Nov 2022 -# https://nvd.nist.gov/vuln/detail/CVE-2022-46149 -# https://github.com/advisories/GHSA-qqff-4vw4-f6hx -# https://github.com/capnproto/capnproto/security/advisories/GHSA-qqff-4vw4-f6hx -# https://github.com/capnproto/capnproto/blob/master/security-advisories/2022-11-30-0-pointer-list-bounds.md -# https://capnproto.org/news/2022-11-30-CVE-2022-46149-security-advisory.html -# https://dwrensha.github.io/capnproto-rust/2022/11/30/out_of_bounds_memory_access_bug.html -if(CapnProto_VERSION STREQUAL "0.9.0" - OR CapnProto_VERSION STREQUAL "0.9.1" - OR CapnProto_VERSION STREQUAL "0.10.0" - OR CapnProto_VERSION STREQUAL "0.10.1" - OR CapnProto_VERSION STREQUAL "0.10.2") - set(CAPNPROTO_CVE_AFFECTED TRUE) - string(APPEND CAPNPROTO_ISSUES "- CVE-2022-46149 security vulnerability (details: https://github.com/advisories/GHSA-qqff-4vw4-f6hx)\n") -endif() - -# Check for Cap'n Proto / Clang / C++20 incompatibility -# Cap'n Proto 0.9.x and 0.10.x are incompatible with Clang 16+ when using C++20 -# due to P2468R2 implementation. This was fixed in Cap'n Proto 1.0+. -# See: https://github.com/bitcoin-core/libmultiprocess/issues/199 -if((CapnProto_VERSION VERSION_GREATER_EQUAL "0.9.0") AND - (CapnProto_VERSION VERSION_LESS "1.0.0") AND - (CMAKE_CXX_COMPILER_ID STREQUAL "Clang") AND - (CMAKE_CXX_COMPILER_VERSION VERSION_GREATER_EQUAL "16") AND - (CMAKE_CXX_STANDARD EQUAL 20)) - set(CAPNPROTO_CLANG_INCOMPATIBLE TRUE) - string(APPEND CAPNPROTO_ISSUES "- Incompatible with Clang ${CMAKE_CXX_COMPILER_VERSION} when using C++20\n") -endif() - -if(CAPNPROTO_CVE_AFFECTED OR CAPNPROTO_CLANG_INCOMPATIBLE) - set(RESOLUTION_OPTIONS "") - - # Fixes both issues - string(APPEND RESOLUTION_OPTIONS " - Upgrade to Cap'n Proto version 1.0 or newer (recommended)\n") - - if(CAPNPROTO_CVE_AFFECTED AND NOT CAPNPROTO_CLANG_INCOMPATIBLE) - string(APPEND RESOLUTION_OPTIONS " - Upgrade to a patched minor version (0.9.2, 0.10.3, or later)\n") - elseif(CAPNPROTO_CLANG_INCOMPATIBLE AND NOT CAPNPROTO_CVE_AFFECTED) - string(APPEND RESOLUTION_OPTIONS " - Use GCC instead of Clang\n") - endif() - - string(APPEND RESOLUTION_OPTIONS " - For Bitcoin Core compilation build with -DENABLE_IPC=OFF to disable multiprocess support\n") - - message(FATAL_ERROR - "The version of Cap'n Proto detected: ${CapnProto_VERSION} has known compatibility issues:\n" - "${CAPNPROTO_ISSUES}" - "To resolve, choose one of the following:\n" - "${RESOLUTION_OPTIONS}" - ) -endif() - set(MPGEN_EXECUTABLE "" CACHE FILEPATH "If specified, should be full path to an external mpgen binary to use rather than the one built internally.") option(MP_ENABLE_CLANG_TIDY "Run clang-tidy with the compiler." OFF) diff --git a/ci/configs/olddeps.bash b/ci/configs/olddeps.bash index 151a300c..179fb837 100644 --- a/ci/configs/olddeps.bash +++ b/ci/configs/olddeps.bash @@ -4,5 +4,5 @@ CI_DIR=build-olddeps # requires an older GCC. NIXPKGS_CHANNEL=nixos-25.05 export CXXFLAGS="-Werror -Wall -Wextra -Wpedantic -Wno-unused-parameter -Wno-error=array-bounds" -NIX_ARGS=(--argstr capnprotoVersion "0.9.2" --argstr cmakeVersion "3.12.4" --argstr gccVersion "11") +NIX_ARGS=(--argstr capnprotoVersion "1.0.0" --argstr cmakeVersion "3.12.4" --argstr gccVersion "11") BUILD_ARGS=(-k) diff --git a/doc/install.md b/doc/install.md index 015cfd19..699dd469 100644 --- a/doc/install.md +++ b/doc/install.md @@ -1,6 +1,6 @@ # libmultiprocess Installation -Installation currently requires Cap'n Proto 0.9 or higher: +Installation currently requires Cap'n Proto 1.0 or higher: ```sh apt install libcapnp-dev capnproto From 0d7a2ffebf1611e4513fc4b620541d35a51ff8ed Mon Sep 17 00:00:00 2001 From: xyzconstant <263061129+xyzconstant@users.noreply.github.com> Date: Sun, 23 Aug 2026 10:10:54 -0300 Subject: [PATCH 6/6] test: Add testing for request cancellation - One test cancels an in-flight `ProxyClient` call from another thread. - Another drops the response promise mid-execution, imitating non-libmultiprocess clients. --- test/mp/test/foo-types.h | 1 + test/mp/test/foo.capnp | 1 + test/mp/test/foo.h | 10 +++++ test/mp/test/test.cpp | 95 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 107 insertions(+) diff --git a/test/mp/test/foo-types.h b/test/mp/test/foo-types.h index 39e4c12e..16df0321 100644 --- a/test/mp/test/foo-types.h +++ b/test/mp/test/foo-types.h @@ -13,6 +13,7 @@ #include #include #include +#include #include #include #include diff --git a/test/mp/test/foo.capnp b/test/mp/test/foo.capnp index 55872ac4..37afe610 100644 --- a/test/mp/test/foo.capnp +++ b/test/mp/test/foo.capnp @@ -40,6 +40,7 @@ interface FooInterface $Proxy.wrap("mp::test::FooImplementation") { listBars @25 (context :Proxy.Context, n :Int32) -> (result :List(BarInterface)); callMessageAsync @26 (context :Proxy.Context) -> (result :FooMessage); passExtra @27 (arg :Int32) -> (result :Int32) $Proxy.extraParam("extra"); + callCancelFnAsync @28 (context :Proxy.Context) -> () $Proxy.extraParam("cancel") $Cxx.allowCancellation; } interface FooInit $Proxy.wrap("mp::test::FooInit") { diff --git a/test/mp/test/foo.h b/test/mp/test/foo.h index 8bfd0474..a22fa345 100644 --- a/test/mp/test/foo.h +++ b/test/mp/test/foo.h @@ -88,6 +88,10 @@ class SimpleBar : public Bar int value() override { return m_value; } int m_value; }; + +using CancelFn = std::function; +using CancelArg = std::function; + class FooImplementation { public: @@ -126,8 +130,14 @@ class FooImplementation void callFnAsync() { assert(m_fn); m_fn(); } int callIntFnAsync(int arg) { assert(m_int_fn); return m_int_fn(arg); } FooMessage callMessageAsync() { assert(m_fn); m_fn(); return {}; } + void callCancelFnAsync(CancelArg cancel) + { + assert(m_cancel_fn); + m_cancel_fn(std::move(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 415b43f3..58716deb 100644 --- a/test/mp/test/test.cpp +++ b/test/mp/test/test.cpp @@ -26,6 +26,7 @@ #include #include #include +#include #include #include #include @@ -766,5 +767,99 @@ KJ_TEST("Call async IPC method without thread or pool errors correctly") KJ_EXPECT(error_thrown); } +KJ_TEST("Cancel an in-flight IPC call") +{ + TestSetup setup; + ProxyClient* foo = setup.client.get(); + foo->initThreadMap(); + std::promise waiting; + std::promise done; + + // Install a function that blocks until its `CancelArg` fires, so + // cancellation is the only way out. + setup.server->m_impl->m_cancel_fn = [&](CancelArg cancel) { + std::mutex mutex; + std::condition_variable cv; + bool canceled = false; + cancel([&] { + const std::lock_guard lock{mutex}; + canceled = true; + cv.notify_all(); + }); + std::unique_lock lock{mutex}; + waiting.set_value(); + cv.wait(lock, [&] { return canceled; }); + done.set_value(); + }; + + std::promise cancel_fn; + std::thread canceler([&] { + CancelFn fire{cancel_fn.get_future().get()}; + waiting.get_future().wait(); + fire(); + }); + bool interrupted = false; + try { + foo->callCancelFnAsync([&](CancelFn fn) { cancel_fn.set_value(std::move(fn)); }); + } catch (const InterruptException&) { + interrupted = true; + } + canceler.join(); + KJ_EXPECT(interrupted); + KJ_EXPECT(done.get_future().wait_for(std::chrono::minutes{5}) == std::future_status::ready); + + // Connection should be unaffected. + KJ_EXPECT(foo->add(1, 2) == 3); +} + +KJ_TEST("Dropping the client promise cancels an executing method") +{ + TestSetup setup; + constexpr std::chrono::seconds timeout{30}; + std::promise waiting; + std::promise done; + + // Install a function that blocks until its `CancelArg` fires, so + // cancellation is the only way out. + setup.server->m_impl->m_cancel_fn = [&](CancelArg cancel) { + std::mutex mutex; + std::condition_variable cv; + bool canceled = false; + cancel([&] { + const std::lock_guard lock{mutex}; + canceled = true; + cv.notify_all(); + }); + std::unique_lock lock{mutex}; + waiting.set_value(); + cv.wait(lock, [&] { return 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