Repository navigation
Allow request cancellation for wrapped C++ methods #342
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -18,6 +18,8 @@ | |
| #include <condition_variable> | ||
| #include <cstdlib> | ||
| #include <functional> | ||
| #include <kj/async.h> | ||
| #include <kj/exception.h> | ||
| #include <kj/function.h> | ||
| #include <kj/io.h> | ||
| #include <map> | ||
|
|
@@ -39,6 +41,17 @@ struct InvokeContext | |
| struct ClientInvokeContext : InvokeContext | ||
| { | ||
| ThreadContext& thread_context; | ||
| //! Callback that overloads can set while building a parameter to make a | ||
| //! function call cancelable. If set, clientInvoke calls it on the event loop | ||
| //! thread with the kj::Canceler object wrapping the request when the request is | ||
| //! sent, and with null when the request finishes and the canceler is destroyed. | ||
| std::function<void(kj::Canceler*)> set_canceler; | ||
| //! Callback that lets clients set a custom error handler for kj::Exception. | ||
| //! For example, a canceled request fails the same way a disconnect does, | ||
| //! and this is where the cancellation parameter can throw its own | ||
| //! exception instead. If set, clientInvoke calls it and rethrows anything | ||
| //! it throws. | ||
| std::function<void(const kj::Exception&)> handle_error; | ||
| ClientInvokeContext(Connection& conn, ThreadContext& thread_context) | ||
| : InvokeContext{conn}, thread_context{thread_context} | ||
| { | ||
|
|
@@ -58,14 +71,22 @@ 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* request_lock{nullptr}; | ||
|
Contributor
There was a problem hiding this comment. In commit 7f31996: proxy: rename This is a good change to use the naming nit: there is a comment that I think you missed; index 63b3fdb..c1ca04f 100644
--- a/include/mp/type-context.h
+++ b/include/mp/type-context.h
@@ -154,7 +154,7 @@ 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 request_lock at |
||
| //! 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<void()> 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} | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -8,6 +8,9 @@ | |
| #include <mp/proxy-io.h> | ||
|
|
||
| #include <exception> | ||
| #include <kj/async.h> | ||
| #include <kj/common.h> | ||
| #include <kj/memory.h> | ||
| #include <optional> | ||
| #include <set> | ||
| #include <typeindex> | ||
|
|
@@ -540,28 +543,84 @@ ClientParam<Accessor, Types...> MakeClientParam(Types&&... values) | |
| return {std::forward<Types>(values)...}; | ||
| } | ||
|
|
||
| //! Client parameter with no capnp field, generated for method parameters | ||
| //! declared with `$Proxy.extraParam`. Its value is handed to a | ||
| //! CustomBuildExtraParam overload instead of being built into the request. | ||
| template <typename... Types> | ||
| struct ClientParam<void, Types...> | ||
| { | ||
| ClientParam(Types&&... values) : m_values{std::forward<Types>(values)...} {} | ||
|
|
||
| struct BuildParams : IterateFieldsHelper<BuildParams, sizeof...(Types)> | ||
| { | ||
| template <typename Params, typename ParamList> | ||
| void handleField(ClientInvokeContext& invoke_context, Params&, ParamList) | ||
| { | ||
| static_assert((requires(ClientInvokeContext& context, Types&& value) { | ||
| CustomBuildExtraParam(TypeList<RemoveCvRef<Types>>(), context, std::forward<Types>(value)); | ||
| } && ...), | ||
| "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 " | ||
| "`CustomBuildExtraParam` overload."); | ||
| auto const fun = [&](Types&&... values) { | ||
| (CustomBuildExtraParam(TypeList<RemoveCvRef<Types>>(), invoke_context, std::forward<Types>(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<ReadResults, sizeof...(Types)> | ||
| { | ||
| template <typename Results, typename ParamList> | ||
| void handleField(ClientInvokeContext&, Results&, ParamList) | ||
| { | ||
| } | ||
| ReadResults(ClientParam*) {} | ||
| }; | ||
|
|
||
| std::tuple<Types&&...> m_values; | ||
| }; | ||
|
|
||
| struct ServerCall | ||
| { | ||
| // FIXME: maybe call call_context.releaseParams() | ||
| template <typename ServerContext, typename... Extra, typename... Args> | ||
| decltype(auto) invoke(ServerContext& server_context, TypeList<Extra...>, Args&&... args) const | ||
| { | ||
| // 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. | ||
| static_assert((requires { CustomReadExtraParam(TypeList<RemoveCvRef<Extra>>(), 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."); | ||
| std::tuple<RemoveCvRef<Extra>...> extra{CustomReadExtraParam(TypeList<RemoveCvRef<Extra>>(), server_context)...}; | ||
| // 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.request_lock) server_context.request_lock->m_lock.unlock(); | ||
| return TryFinally( | ||
| [&]() -> decltype(auto) { | ||
| return std::apply( | ||
| [&](RemoveCvRef<Extra>&... extra_args) -> decltype(auto) { | ||
| return ProxyServerMethodTraits< | ||
| typename decltype(server_context.call_context.getParams())::Reads | ||
| >::invoke(server_context, std::forward<Args>(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; | ||
|
Contributor
There was a problem hiding this comment. In cfdcb45: proxy: support cancellation extra parameters Could cancellation run the server callback after the local variables it references have been destroyed? The callback is cleared only after the wrapped method returns, but the method’s local variables are destroyed before that. If cancellation arrives in between, the callback could access destroyed state—for example, a local mutex or condition variable. I reproduced this with a test that pauses during method cleanup and triggers cancellation after the local state is destroyed. The callback still runs. index 619930e..5ad02a0 100644
--- a/test/mp/test/test.cpp
+++ b/test/mp/test/test.cpp
@@ -816,6 +816,85 @@ KJ_TEST("Cancel an in-flight IPC call")
KJ_EXPECT(foo->add(1, 2) == 3);
}
+KJ_TEST("Cancellation callback does not outlive server method locals")
+{
+ TestSetup setup;
+ auto* foo = setup.client.get();
+ foo->initThreadMap();
+ constexpr std::chrono::seconds timeout{30};
+ std::promise<void> locals_destroyed;
+ auto locals_destroyed_future = locals_destroyed.get_future();
+ std::promise<void> cancellation_attempted;
+ auto cancellation_attempted_future = cancellation_attempted.get_future();
+ std::atomic<bool> callback_after_destruction{false};
+
+ setup.server->m_impl->m_cancel_fn = [&](CancelArg cancel) {
+ // This guard runs after local_state is destroyed, but before the
+ // method returns to ServerCall's callback cleanup. Hold that window
+ // open so cancellation reliably arrives during normal scope exit.
+ KJ_DEFER(
+ locals_destroyed.set_value();
+ KJ_EXPECT(cancellation_attempted_future.wait_for(timeout) == std::future_status::ready);
+ );
+ );
+ KJ_EXPECT(cancellation_attempted_future.wait_for(timeout) == std::future_status::ready);
+ );
+ auto local_state = std::make_shared<int>(42);
+ cancel([state = std::weak_ptr<int>{local_state}, &callback_after_destruction] {
+ // Observe the lifetime safely, instead of dereferencing a dangling
+ // reference to a local mutex/condition variable as real code might.
+ callback_after_destruction = state.expired();
+ });
+ // Return normally; do not wait for cancellation in the method body.
+ };
+
+ std::promise<CancelFn> cancel_fn;
+ auto cancel_future = cancel_fn.get_future();
+ std::thread canceler([&] {
+ CancelFn fire = cancel_future.get();
+ KJ_EXPECT(locals_destroyed_future.wait_for(timeout) == std::future_status::ready);
+ fire();
+ cancellation_attempted.set_value();
+ });
+ 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(!callback_after_destruction.load());
+ KJ_EXPECT(foo->add(1, 2) == 3);
+}
+Should we provide a way to unregister the callback before the captured local variables go out of scope?
Contributor
Author
There was a problem hiding this comment. @ViniciusCestarii reported the same issue and suggested an RAII guard that unregisters the cancel callback before the wrapped method's locals go out of scope. Following @ryanofsky's suggestion in #342 (comment), it's a bit of a redesign. In the latest pushes, |
||
| // 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 | ||
|
|
@@ -593,10 +652,10 @@ struct ServerRet : Parent | |
| { | ||
| ServerRet(Parent parent) : Parent(parent) {} | ||
|
|
||
| template <typename ServerContext, typename ArgTypes, typename... Args> | ||
| void invoke(ServerContext& server_context, ArgTypes arg_types, Args&&... args) const | ||
| { | ||
| auto&& result = Parent::invoke(server_context, arg_types, std::forward<Args>(args)...); | ||
| auto&& results = server_context.call_context.getResults(); | ||
| InvokeContext& invoke_context = server_context; | ||
| BuildField(TypeList<decltype(result)>(), invoke_context, Make<StructField, Accessor>(results), | ||
|
|
@@ -609,11 +668,11 @@ struct ServerExcept : Parent | |
| { | ||
| ServerExcept(Parent parent) : Parent(parent) {} | ||
|
|
||
| template <typename ServerContext, typename ArgTypes, typename... Args> | ||
| void invoke(ServerContext& server_context, ArgTypes arg_types, Args&&... args) const | ||
| { | ||
| try { | ||
| return Parent::invoke(server_context, arg_types, std::forward<Args>(args)...); | ||
| } catch (const Exception& exception) { | ||
| auto&& results = server_context.call_context.getResults(); | ||
| BuildField(TypeList<Exception>(), server_context, Make<StructField, Accessor>(results), exception); | ||
|
|
@@ -759,7 +818,18 @@ 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); | ||
|
|
||
| kj::Promise<::capnp::Response<typename Request::Results>> promise{request.send()}; | ||
|
|
||
| // If `set_canceler` was set, construct a kj::Canceler object and wrap | ||
| // the request promise with it. | ||
| kj::Own<kj::Canceler> canceler; | ||
| if (invoke_context->set_canceler) { | ||
| canceler = kj::heap<kj::Canceler>(); | ||
| promise = canceler->wrap(kj::mv(promise)); | ||
| invoke_context->set_canceler(canceler.get()); | ||
| } | ||
|
|
||
| proxy_client.m_context.loop->m_task_set->add(promise.then( | ||
| [&](::capnp::Response<typename Request::Results>&& response) { | ||
| MP_LOGPLAIN(*proxy_client.m_context.loop, Log::Debug) | ||
| << "{" << thread_context.thread_name << "} IPC client recv " | ||
|
|
@@ -777,7 +847,16 @@ void clientInvoke(ProxyClient& proxy_client, const GetRequest& get_request, Fiel | |
| thread_context.waiter->m_cv.notify_all(); | ||
| }, | ||
| [&](const ::kj::Exception& e) { | ||
| if (invoke_context->handle_error) { | ||
| try { | ||
| invoke_context->handle_error(e); | ||
| } catch (...) { | ||
| exception = std::current_exception(); | ||
| } | ||
| } | ||
| if (exception) { | ||
| // Rethrown below. | ||
| } 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(); | ||
|
|
@@ -787,7 +866,12 @@ 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::defer([canceler = kj::mv(canceler), set_canceler = invoke_context->set_canceler] { | ||
| // Runs on the event loop thread when the request promise is | ||
| // destroyed. Tell the cancellation parameter the canceler is | ||
| // about to go away. | ||
| if (set_canceler) set_canceler(nullptr); | ||
| }))); | ||
| }); | ||
|
|
||
| Lock lock(thread_context.waiter->m_mutex); | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
In 527a099: build: require Cap'n Proto 1.0
nit: There is a text here that needs to be updated i think
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This makes more sense, thanks. Done in 45514af.