11#include <mp/proxy.capnp.h>
13#include <capnp/rpc-twoparty.h>
17#include <condition_variable>
20#include <kj/function.h>
46template <
typename ProxyServer,
typename CallContext_>
74template <
typename Interface,
typename Params,
typename Results>
103 kj::Promise<void> getName(GetNameContext context)
override;
109 template<
typename T,
typename Fn>
110 kj::Promise<T> post(Fn&& fn);
117 kj::Promise<void> m_thread_ready{kj::READY_NOW};
125 void taskFailed(kj::Exception&& exception)
override;
181 template <
typename T>
188 template <
typename T>
191 return logger << std::forward<T>(value);
194 explicit operator bool()
const
210#define MP_LOGPLAIN(loop, ...) if (mp::Logger logger{(loop).m_log_opts, __VA_ARGS__}; logger) logger
212#define MP_LOG(loop, ...) MP_LOGPLAIN(loop, __VA_ARGS__) << "{" << LongThreadName((loop).m_exe_name) << "} "
252 EventLoop(
const char* exe_name, std::function<
void(
bool, std::string)> old_callback,
void* context =
nullptr)
254 LogFn{[old_callback =
std::move(old_callback)](
LogMessage log_data) {old_callback(log_data.level ==
Log::Raise, std::move(log_data.message));}},
266 void post(kj::Function<
void()> fn);
271 template <
typename Callable>
274 post(std::forward<Callable>(callable));
279 void addAsyncCleanup(std::function<
void()> fn);
299 const
char* m_exe_name;
331 std::condition_variable m_cv;
334 kj::AsyncIoContext m_io_context;
389 template <
typename Fn>
392 const Lock lock(m_mutex);
393 if (m_fn)
return false;
394 m_fn = std::forward<Fn>(fn);
399 template <
class Predicate>
402 m_cv.wait(lock.m_lock, [&]()
MP_REQUIRES(m_mutex) {
409 auto fn = std::move(*m_fn);
413 const bool done = pred();
439 :
m_loop(loop), m_stream(kj::mv(stream_)),
440 m_network(*m_stream, ::capnp::rpc::twoparty::Side::CLIENT, ::capnp::ReaderOptions()),
441 m_rpc_system(::capnp::makeRpcClient(m_network)) {}
443 kj::Own<kj::AsyncIoStream>&& stream_,
444 const std::function<::capnp::Capability::Client(
Connection&)>& make_client)
445 :
m_loop(loop), m_stream(kj::mv(stream_)),
446 m_network(*m_stream, ::capnp::rpc::twoparty::Side::SERVER, ::capnp::ReaderOptions()),
447 m_rpc_system(::capnp::makeRpcServer(m_network, make_client(*this))) {}
459 CleanupIt addSyncCleanup(std::function<
void()> fn);
463 template <
typename F>
471 m_on_disconnect.add(m_network.onDisconnect().then(
472 [f = std::forward<F>(f),
this]()
mutable { m_loop->m_task_set->add(kj::evalLater(kj::mv(f))); }));
481 kj::TaskSet m_on_disconnect{m_error_handler};
483 std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>>
m_rpc_system;
487 ThreadMap::Client m_thread_map{
nullptr};
521 ::capnp::word scratch[4]{};
522 ::capnp::MallocMessageBuilder message{scratch};
523 ::capnp::rpc::twoparty::VatId::Builder vat_id{message.getRoot<::capnp::rpc::twoparty::VatId>()};
524 ServerVatId() { vat_id.setSide(::capnp::rpc::twoparty::Side::SERVER); }
527template <
typename Interface,
typename Impl>
530 bool destroy_connection)
539 typename Interface::Client(std::move(
m_client));
565 if (kj::runCatchingExceptions([&]{ Sub::destroy(*
this); }) !=
nullptr) {
581 typename Interface::Client(std::move(m_client));
583 if (destroy_connection) {
590 Sub::construct(*
this);
593template <
typename Interface,
typename Impl>
601template <
typename Interface,
typename Impl>
621template <
typename Interface,
typename Impl>
645 m_context.loop->addAsyncCleanup([impl=std::move(m_impl), fns=std::move(
m_context.cleanup_fns)]()
mutable {
671template <
typename Interface,
typename Impl>
684using ConnThreads = std::map<Connection*, std::optional<ProxyClient<Thread>>>;
725 std::unique_ptr<Waiter> waiter =
nullptr;
760 bool loop_thread =
false;
763template<
typename T,
typename Fn>
766 auto ready = kj::newPromiseAndFulfiller<void>();
767 auto cancel_monitor_ptr = kj::heap<CancelMonitor>();
774 auto self = thisCap();
775 auto ret = m_thread_ready.then([
this, self = std::move(self), fn = std::forward<Fn>(fn), ready_fulfiller = kj::mv(ready.fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr)]()
mutable {
776 auto result = kj::newPromiseAndFulfiller<T>();
777 bool posted = m_thread_context.waiter->post([this, self = std::move(self), fn = std::forward<Fn>(fn), ready_fulfiller = kj::mv(ready_fulfiller), result_fulfiller = kj::mv(result.fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr)]() mutable {
786 m_loop->sync([ready_fulfiller = kj::mv(ready_fulfiller)]() mutable {
787 ready_fulfiller->fulfill();
788 ready_fulfiller = nullptr;
790 std::optional<T> result_value;
791 kj::Maybe<kj::Exception> exception{kj::runCatchingExceptions([&]{ result_value.emplace(fn(*cancel_monitor_ptr)); })};
792 m_loop->sync([this, &result_value, &exception, self = kj::mv(self), result_fulfiller = kj::mv(result_fulfiller), cancel_monitor_ptr = kj::mv(cancel_monitor_ptr)]() mutable {
796 cancel_monitor_ptr = nullptr;
800 KJ_IF_MAYBE(e, exception) {
801 assert(!result_value);
802 result_fulfiller->reject(kj::mv(*e));
804 assert(result_value);
805 result_fulfiller->fulfill(kj::mv(*result_value));
806 result_value.reset();
808 result_fulfiller = nullptr;
813 m_loop->m_task_set->add(kj::evalLater([self = kj::mv(self)] {}));
821 return kj::mv(result.promise);
822 }).attach(kj::heap<CancelProbe>(cancel_monitor));
823 m_thread_ready = kj::mv(ready.promise);
830template <
typename InitInterface>
833 typename InitInterface::Client init_client(
nullptr);
834 std::unique_ptr<Connection> connection;
837 loop.
m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP);
838 connection = std::make_unique<Connection>(loop, kj::mv(stream));
839 init_client = connection->m_rpc_system->bootstrap(
ServerVatId().vat_id).castAs<InitInterface>();
840 Connection* connection_ptr = connection.get();
843 delete connection_ptr;
846 return std::make_unique<ProxyClient<InitInterface>>(
847 kj::mv(init_client), connection.release(),
true);
854template <
typename InitInterface,
typename InitImpl,
typename OnDisconnect>
861 return kj::heap<ProxyServer<InitInterface>>(std::shared_ptr<InitImpl>(&
init, [](
InitImpl*){}), connection);
864 MP_LOG(loop, Log::Info) <<
"IPC server: socket connected.";
866 it->onDisconnect([&loop, it, on_disconnect = std::forward<OnDisconnect>(on_disconnect)]()
mutable {
867 MP_LOG(loop, Log::Info) <<
"IPC server: socket disconnected.";
876 explicit Listener(kj::Own<kj::ConnectionReceiver>&& receiver, std::optional<size_t> max_connections)
877 : m_receiver(kj::mv(receiver)), m_max_connections(max_connections) {}
881 return m_max_connections && m_active_connections >= *m_max_connections;
886 size_t m_active_connections{0};
889template <
typename InitInterface,
typename InitImpl>
894 auto* receiver =
listener->m_receiver.get();
896 [&loop, &
init,
listener](kj::Own<kj::AsyncIoStream>&& stream) {
897 ++listener->m_active_connections;
898 _Serve<InitInterface>(loop, kj::mv(stream), init, [&loop, &init, listener] {
899 const bool resume_accept{listener->atCapacity()};
900 assert(listener->m_active_connections > 0);
901 --listener->m_active_connections;
902 if (resume_accept) _Listen<InitInterface>(listener, loop, init);
910template <
typename InitInterface,
typename InitImpl>
913 _Serve<InitInterface>(
915 loop.
m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
922template <
typename InitInterface,
typename InitImpl>
926 auto listener{std::make_shared<Listener>(
927 loop.
m_io_context.lowLevelProvider->wrapListenSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
Helper class that detects when a promise is canceled.
Object holding network & rpc state associated with either an incoming server connection,...
CleanupIt addSyncCleanup(std::function< void()> fn)
Register synchronous cleanup function to run on event loop thread (with access to capnp thread local ...
::capnp::TwoPartyVatNetwork m_network
kj::Own< kj::AsyncIoStream > m_stream
Connection(EventLoop &loop, kj::Own< kj::AsyncIoStream > &&stream_, const std::function<::capnp::Capability::Client(Connection &)> &make_client)
Connection(EventLoop &loop, kj::Own< kj::AsyncIoStream > &&stream_)
void onDisconnect(F &&f)
Add disconnect handler.
std::vector< PoolSlot > m_thread_pool
kj::Canceler m_canceler
Canceler for canceling promises that we want to discard when the connection is destroyed.
::capnp::CapabilityServerSet< Thread > m_threads
Collection of server-side IPC worker threads (ProxyServer<Thread> objects previously returned by Thre...
CleanupList m_sync_cleanup_fns
Cleanup functions to run if connection is broken unexpectedly.
std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId > > m_rpc_system
void removeSyncCleanup(CleanupIt it)
Event loop implementation.
kj::AsyncIoContext m_io_context
Capnp IO context.
void sync(Callable &&callable)
Wrapper around EventLoop::post that takes advantage of the fact that callable will not go out of scop...
EventLoop(const char *exe_name, LogFn log_fn, void *context=nullptr)
Construct event loop object with default logging options.
std::function< void()> testing_hook_connected
Hook called on the event loop thread when a client has connected.
std::list< Connection > m_incoming_connections
List of connections.
Mutex m_mutex
Mutex and condition variable used to post tasks to event loop and async thread.
std::function< void()> testing_hook_disconnected
Hook called on the event loop thread when a client has disconnected.
LogOptions m_log_opts
Logging options.
std::unique_ptr< kj::TaskSet > m_task_set
Capnp list of pending promises.
void * m_context
External context pointer.
std::function< void()> testing_hook_makethread
Hook called when ProxyServer<ThreadMap>::makeThread() is called.
std::function< void()> testing_hook_makethread_created
Hook called on the worker thread inside makeThread(), after the thread context is set up and thread_c...
EventLoop(const char *exe_name, std::function< void(bool, std::string)> old_callback, void *context=nullptr)
Backwards-compatible constructor for previous (deprecated) logging callback signature.
std::function< void()> testing_hook_async_request_done
Hook called on the worker thread just before returning results.
std::function< void()> testing_hook_async_request_start
Hook called on the worker thread when it starts to execute an async request.
Event loop smart pointer automatically managing m_num_refs.
friend Logger & operator<<(Logger &logger, T &&value)
Logger(const Logger &)=delete
~Logger() noexcept(false)
Logger & operator=(Logger &&)=delete
Logger(const LogOptions &options, Log log_level)
const LogOptions & m_options
friend Logger & operator<<(Logger &&logger, T &&value)
std::ostringstream m_buffer
Logger & operator=(const Logger &)=delete
Handler for kj::TaskSet failed task events.
LoggingErrorHandler(EventLoop &loop)
void taskFailed(kj::Exception &&exception) override
Base class for generated ProxyClient classes that implement a C++ interface and forward calls to a ca...
Interface::Client m_client
~ProxyClientBase() noexcept
ProxyClientBase(typename Interface::Client client, Connection *connection, bool destroy_connection)
Construct libmultiprocess client object wrapping Cap'n Proto client object with a reference to the as...
std::optional< mp::EventLoop > m_loop
EventLoop object which manages I/O events for all connections.
std::unique_ptr< ProxyClient< messages::FooInterface > > client
std::thread thread
Thread variable should be after other struct members so the thread does not start until the other mem...
Functions to serialize / deserialize common bitcoin types.
void Unlock(Lock &lock, Callback &&callback)
void ListenConnections(EventLoop &loop, int fd, InitImpl &init, std::optional< size_t > max_connections=std::nullopt)
Given listening socket file descriptor and an init object, handle incoming connections and requests b...
std::string CxxTypeName(const T &)
kj::StringPtr KJ_STRINGIFY(Log flags)
std::list< std::function< void()> > CleanupList
void ServeStream(EventLoop &loop, int fd, InitImpl &init)
Given stream file descriptor and an init object, handle requests on the stream by calling methods on ...
void _Serve(EventLoop &loop, kj::Own< kj::AsyncIoStream > &&stream, InitImpl &init, OnDisconnect &&on_disconnect)
Given stream and init objects, construct a new ProxyServer object that handles requests from the stre...
std::function< void(LogMessage)> LogFn
std::tuple< ConnThread, bool > SetThread(GuardedRef< ConnThreads > threads, Connection *connection, const std::function< Thread::Client()> &make_thread)
thread_local ThreadContext g_thread_context
std::unique_ptr< ProxyClient< InitInterface > > ConnectStream(EventLoop &loop, int fd)
Given stream file descriptor, make a new ProxyClient object to send requests over the stream.
Log
Log flags. Update stringify function if changed!
std::string LongThreadName(const char *exe_name)
ConnThreads::iterator ConnThread
typename CleanupList::iterator CleanupIt
void CleanupRun(CleanupList &fns)
void _Listen(const std::shared_ptr< Listener > &listener, EventLoop &loop, InitImpl &init)
std::map< Connection *, std::optional< ProxyClient< Thread > > > ConnThreads
Map from Connection to local or remote thread handle which will be used over that connection.
ThreadContext & thread_context
ClientInvokeContext(Connection &conn, ThreadContext &thread_context)
A thread created by makePool with associated pending work queue. Vector is filled once by makePool() ...
std::optional< size_t > m_max_connections
kj::Own< kj::ConnectionReceiver > m_receiver
Listener(kj::Own< kj::ConnectionReceiver > &&receiver, std::optional< size_t > max_connections)
Log level
The severity level of this message.
std::string message
Message to be logged.
LogFn log_fn
External logging callback.
size_t max_chars
Maximum number of characters to use when representing request and response structs as strings.
Log log_level
Messages with a severity level less than log_level will not be reported.
ProxyClient(const ProxyClient &)=delete
std::optional< CleanupIt > m_disconnect_cb
Reference to callback function that is run if there is a sudden disconnect and the Connection object ...
Mapping from capnp interface type to proxy client implementation (specializations are generated by pr...
ThreadContext & m_thread_context
Base class for generated ProxyServer classes that implement capnp server methods and forward calls to...
virtual ~ProxyServerBase()
ProxyServer destructor, called from the EventLoop thread by Cap'n Proto garbage collection code after...
std::shared_ptr< Impl > m_impl
Implementation pointer that may or may not be owned and deleted when this capnp server goes out of sc...
Mapping from capnp interface type to proxy server implementation (specializations are generated by pr...
ServerInvokeContext(ProxyServer &proxy_server, CallContext &call_context, int req)
CallContext & call_context
ProxyServer & proxy_server
bool request_canceled
For IPC methods that execute asynchronously, not on the event-loop thread, this is set to true if the...
Lock * cancel_lock
For IPC methods that execute asynchronously, not on the event-loop thread: lock preventing the event-...
Vat id for server side of connection.
The thread_local ThreadContext g_thread_context struct provides information about individual threads ...
ConnThreads callback_threads MP_GUARDED_BY(waiter->m_mutex)
When client is making a request to a server, this is the callbackThread argument it passes in the req...
ConnThreads request_threads MP_GUARDED_BY(waiter->m_mutex)
When client is making a request to a server, this is the thread argument it passes in the request,...
std::string thread_name
Identifying string for debug.
Single element task queue used to handle recursive capnp calls.
void wait(Lock &lock, Predicate pred) MP_REQUIRES(m_mutex)
Mutex m_mutex
Mutex mainly used internally by waiter class, but also used externally to guard access to related sta...
std::optional< kj::Function< void()> > m_fn MP_GUARDED_BY(m_mutex)
std::condition_variable m_cv MP_GUARDED_BY(m_mutex)