5#ifndef MP_PROXY_TYPE_CONTEXT_H
6#define MP_PROXY_TYPE_CONTEXT_H
12template <
typename Output>
17 typename std::enable_if<std::is_same<
decltype(output.get()), Context::Builder>::value>::type* enable =
nullptr)
28 GuardedRef{thread_context.waiter->m_mutex, thread_context.callback_threads}, &connection,
29 [&] {
return connection.m_threads.add(kj::heap<
ProxyServer<Thread>>(thread_context, std::thread{})); })};
34 auto make_request_thread{[&]{
41 auto request = connection.m_thread_map.makeThreadRequest();
42 request.setName(thread_context.thread_name);
43 return request.send().getResult();
46 GuardedRef{thread_context.waiter->m_mutex, thread_context.request_threads},
47 &connection, make_request_thread)};
49 auto context = output.init();
50 context.setThread(request_thread->second->m_client);
51 context.setCallbackThread(callback_thread->second->m_client);
56template <
typename Accessor,
typename ServerContext,
typename Fn,
typename... Args>
58 typename std::enable_if<
59 std::is_same<
decltype(Accessor::get(server_context.call_context.getParams())), Context::Reader>::value,
60 kj::Promise<typename ServerContext::CallContext>>::type
62 const auto& params = server_context.call_context.getParams();
63 Context::Reader context_arg = Accessor::get(params);
64 auto future = kj::newPromiseAndFulfiller<typename ServerContext::CallContext>();
65 auto& server = server_context.proxy_server;
66 int req = server_context.req;
67 auto invoke = [fulfiller = kj::mv(future.fulfiller),
68 call_context = kj::mv(server_context.call_context), &server, req, fn,
args...]()
mutable {
69 const auto& params = call_context.getParams();
70 Context::Reader context_arg = Accessor::get(params);
91 auto& request_threads = thread_context.request_threads;
94 server.m_context.loop->sync([&] {
95 std::tie(request_thread, inserted) =
SetThread(
96 GuardedRef{thread_context.waiter->m_mutex, request_threads}, server.m_context.connection,
97 [&] {
return context_arg.getCallbackThread(); });
105 const bool erase_thread{inserted};
106 KJ_DEFER(
if (erase_thread) {
112 server.m_context.loop->sync([&] {
119 ConnThreads::node_type removed;
121 Lock lock(thread_context.waiter->m_mutex);
122 removed = request_threads.extract(server.m_context.connection);
126 fn.invoke(server_context,
args...);
128 KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() {
129 server.m_context.loop->sync([&] {
130 auto fulfiller_dispose = kj::mv(fulfiller);
131 fulfiller_dispose->fulfill(kj::mv(call_context));
135 server.m_context.loop->sync([&]() {
136 auto fulfiller_dispose = kj::mv(fulfiller);
137 fulfiller_dispose->reject(kj::mv(*exception));
145 auto thread_client = context_arg.getThread();
146 return server.m_context.connection->m_threads.getLocalServer(thread_client)
147 .then([&server, invoke = kj::mv(invoke), req](
const kj::Maybe<Thread::Server&>& perhaps)
mutable {
151 KJ_IF_MAYBE (thread_server, perhaps) {
153 server.m_context.loop->log()
155 thread.m_thread_context.waiter->post(std::move(invoke));
157 server.m_context.loop->log()
158 <<
"IPC server error request #" << req <<
", missing thread to execute request";
159 throw std::runtime_error(
"invalid thread handle");
163 .then([invoke_wait = kj::mv(future.promise)]()
mutable {
return kj::mv(invoke_wait); });
Functions to serialize / deserialize common bitcoin types.
auto PassField(Priority< 1 >, TypeList< LocalType & >, ServerContext &server_context, Fn &&fn, Args &&... args) -> Require< typename decltype(Accessor::get(server_context.call_context.getParams()))::Calls >
PassField override for callable interface reference arguments.
ServerInvokeContext< ProxyServer< Interface >, ::capnp::CallContext< Params, Results > > ServerContext
std::tuple< ConnThread, bool > SetThread(GuardedRef< ConnThreads > threads, Connection *connection, const std::function< Thread::Client()> &make_thread)
thread_local ThreadContext g_thread_context
ConnThreads::iterator ConnThread
void CustomBuildField(TypeList< LocalType >, Priority< 1 >, InvokeContext &invoke_context, Value &&value, Output &&output)
Overload multiprocess library's CustomBuildField hook to allow any serializable object to be stored i...
ThreadContext & thread_context
Function parameter type for prioritizing overloaded function calls that would otherwise be ambiguous.
ThreadContext & m_thread_context
std::string thread_name
Identifying string for debug.
Generic utility functions used by capnp code.
consteval auto _(util::TranslatedLiteral str)