1 // Copyright (c) The Bitcoin Core developers
2 // Distributed under the MIT software license, see the accompanying
3 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
4 5 #ifndef MP_PROXY_IO_H
6 #define MP_PROXY_IO_H
7 8 #include <mp/proxy.h>
9 #include <mp/util.h>
10 11 #include <mp/proxy.capnp.h>
12 13 #include <capnp/rpc-twoparty.h>
14 15 #include <assert.h>
16 #include <algorithm>
17 #include <condition_variable>
18 #include <cstdlib>
19 #include <functional>
20 #include <kj/function.h>
21 #include <map>
22 #include <memory>
23 #include <optional>
24 #include <sstream>
25 #include <string>
26 #include <thread>
27 28 namespace mp {
29 struct ThreadContext;
30 struct Listener;
31 32 struct InvokeContext
33 {
34 Connection& connection;
35 };
36 37 struct ClientInvokeContext : InvokeContext
38 {
39 ThreadContext& thread_context;
40 ClientInvokeContext(Connection& conn, ThreadContext& thread_context)
41 : InvokeContext{conn}, thread_context{thread_context}
42 {
43 }
44 };
45 46 template <typename ProxyServer, typename CallContext_>
47 struct ServerInvokeContext : InvokeContext
48 {
49 using CallContext = CallContext_;
50 51 ProxyServer& proxy_server;
52 CallContext& call_context;
53 int req;
54 //! For IPC methods that execute asynchronously, not on the event-loop
55 //! thread: lock preventing the event-loop thread from freeing the params or
56 //! results structs if the request is canceled while the worker thread is
57 //! reading params (`call_context.getParams()`) or writing results
58 //! (`call_context.getResults()`).
59 Lock* cancel_lock{nullptr};
60 //! For IPC methods that execute asynchronously, not on the event-loop
61 //! thread, this is set to true if the IPC call was canceled by the client
62 //! or canceled by a disconnection. If the call runs on the event-loop
63 //! thread, it can't be canceled. This should be accessed with cancel_lock
64 //! held if it is not null, since in the asynchronous case it is accessed
65 //! from multiple threads.
66 bool request_canceled{false};
67 68 ServerInvokeContext(ProxyServer& proxy_server, CallContext& call_context, int req)
69 : InvokeContext{*proxy_server.m_context.connection}, proxy_server{proxy_server}, call_context{call_context}, req{req}
70 {
71 }
72 };
73 74 template <typename Interface, typename Params, typename Results>
75 using ServerContext = ServerInvokeContext<ProxyServer<Interface>, ::capnp::CallContext<Params, Results>>;
76 77 template <>
78 struct ProxyClient<Thread> : public ProxyClientBase<Thread, ::capnp::Void>
79 {
80 using ProxyClientBase::ProxyClientBase;
81 // https://stackoverflow.com/questions/22357887/comparing-two-mapiterators-why-does-it-need-the-copy-constructor-of-stdpair
82 ProxyClient(const ProxyClient&) = delete;
83 ~ProxyClient();
84 85 //! Reference to callback function that is run if there is a sudden
86 //! disconnect and the Connection object is destroyed before this
87 //! ProxyClient<Thread> object. The callback will destroy this object and
88 //! remove its entry from the thread's request_threads or callback_threads
89 //! map. It will also reset m_disconnect_cb so the destructor does not
90 //! access it. In the normal case where there is no sudden disconnect, the
91 //! destructor will unregister m_disconnect_cb so the callback is never run.
92 //! Since this variable is accessed from multiple threads, accesses should
93 //! be guarded with the associated Waiter::m_mutex.
94 std::optional<CleanupIt> m_disconnect_cb;
95 };
96 97 template <>
98 struct ProxyServer<Thread> final : public Thread::Server
99 {
100 public:
101 ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread);
102 ~ProxyServer();
103 kj::Promise<void> getName(GetNameContext context) override;
104 105 //! Run a callback function fn returning T on this thread. The function will
106 //! be queued and executed as soon as the thread is idle, and when fn
107 //! returns, the promise returned by this method will be fulfilled with the
108 //! value fn returned.
109 template<typename T, typename Fn>
110 kj::Promise<T> post(Fn&& fn);
111 112 EventLoopRef m_loop;
113 ThreadContext& m_thread_context;
114 std::thread m_thread;
115 //! Promise signaled when m_thread_context.waiter is ready and there is no
116 //! post() callback function waiting to execute.
117 kj::Promise<void> m_thread_ready{kj::READY_NOW};
118 };
119 120 //! Handler for kj::TaskSet failed task events.
121 class LoggingErrorHandler : public kj::TaskSet::ErrorHandler
122 {
123 public:
124 LoggingErrorHandler(EventLoop& loop) : m_loop(loop) {}
125 void taskFailed(kj::Exception&& exception) override;
126 EventLoop& m_loop;
127 };
128 129 //! Log flags. Update stringify function if changed!
130 enum class Log {
131 Trace = 0,
132 Debug,
133 Info,
134 Warning,
135 Error,
136 Raise,
137 };
138 139 kj::StringPtr KJ_STRINGIFY(Log flags);
140 141 struct LogMessage {
142 143 //! Message to be logged
144 std::string message;
145 146 //! The severity level of this message
147 Log level;
148 };
149 150 using LogFn = std::function<void(LogMessage)>;
151 152 struct LogOptions {
153 154 //! External logging callback.
155 LogFn log_fn;
156 157 //! Maximum number of characters to use when representing
158 //! request and response structs as strings.
159 size_t max_chars{200};
160 161 //! Messages with a severity level less than log_level will not be
162 //! reported.
163 Log log_level{Log::Trace};
164 };
165 166 class Logger
167 {
168 public:
169 Logger(const LogOptions& options, Log log_level) : m_options(options), m_log_level(log_level) {}
170 171 Logger(Logger&&) = delete;
172 Logger& operator=(Logger&&) = delete;
173 Logger(const Logger&) = delete;
174 Logger& operator=(const Logger&) = delete;
175 176 ~Logger() noexcept(false)
177 {
178 if (enabled()) m_options.log_fn({std::move(m_buffer).str(), m_log_level});
179 }
180 181 template <typename T>
182 friend Logger& operator<<(Logger& logger, T&& value)
183 {
184 if (logger.enabled()) logger.m_buffer << std::forward<T>(value);
185 return logger;
186 }
187 188 template <typename T>
189 friend Logger& operator<<(Logger&& logger, T&& value)
190 {
191 return logger << std::forward<T>(value);
192 }
193 194 explicit operator bool() const
195 {
196 return enabled();
197 }
198 199 private:
200 bool enabled() const
201 {
202 return m_options.log_fn && m_log_level >= m_options.log_level;
203 }
204 205 const LogOptions& m_options;
206 Log m_log_level;
207 std::ostringstream m_buffer;
208 };
209 210 #define MP_LOGPLAIN(loop, ...) if (mp::Logger logger{(loop).m_log_opts, __VA_ARGS__}; logger) logger
211 212 #define MP_LOG(loop, ...) MP_LOGPLAIN(loop, __VA_ARGS__) << "{" << LongThreadName((loop).m_exe_name) << "} "
213 214 std::string LongThreadName(const char* exe_name);
215 216 //! Event loop implementation.
217 //!
218 //! Cap'n Proto threading model is very simple: all I/O operations are
219 //! asynchronous and must be performed on a single thread. This includes:
220 //!
221 //! - Code starting an asynchronous operation (calling a function that returns a
222 //! promise object)
223 //! - Code notifying that an asynchronous operation is complete (code using a
224 //! fulfiller object)
225 //! - Code handling a completed operation (code chaining or waiting for a promise)
226 //!
227 //! All of this code needs to access shared state, and there is no mutex that
228 //! can be acquired to lock this state because Cap'n Proto
229 //! assumes it will only be accessed from one thread. So all this code needs to
230 //! actually run on one thread, and the EventLoop::loop() method is the entry point for
231 //! this thread. ProxyClient and ProxyServer objects that use other threads and
232 //! need to perform I/O operations post to this thread using EventLoop::post()
233 //! and EventLoop::sync() methods.
234 //!
235 //! Specifically, because ProxyClient methods can be called from arbitrary
236 //! threads, and ProxyServer methods can run on arbitrary threads, ProxyClient
237 //! methods use the EventLoop thread to send requests, and ProxyServer methods
238 //! use the thread to return results.
239 //!
240 //! Based on https://groups.google.com/d/msg/capnproto/TuQFF1eH2-M/g81sHaTAAQAJ
241 class EventLoop
242 {
243 public:
244 //! Construct event loop object with default logging options.
245 EventLoop(const char* exe_name, LogFn log_fn, void* context = nullptr)
246 : EventLoop(exe_name, LogOptions{std::move(log_fn)}, context){}
247 248 //! Construct event loop object with specified logging options.
249 EventLoop(const char* exe_name, LogOptions log_opts, void* context = nullptr);
250 251 //! Backwards-compatible constructor for previous (deprecated) logging callback signature
252 EventLoop(const char* exe_name, std::function<void(bool, std::string)> old_callback, void* context = nullptr)
253 : EventLoop(exe_name,
254 LogFn{[old_callback = std::move(old_callback)](LogMessage log_data) {old_callback(log_data.level == Log::Raise, std::move(log_data.message));}},
255 context){}
256 257 ~EventLoop();
258 259 //! Run event loop. Does not return until shutdown. This should only be
260 //! called once from the m_thread_id thread. This will block until
261 //! the m_num_refs reference count is 0.
262 void loop();
263 264 //! Run function on event loop thread. Does not return until function completes.
265 //! Must be called while the loop() function is active.
266 void post(kj::Function<void()> fn);
267 268 //! Wrapper around EventLoop::post that takes advantage of the
269 //! fact that callable will not go out of scope to avoid requirement that it
270 //! be copyable.
271 template <typename Callable>
272 void sync(Callable&& callable)
273 {
274 post(std::forward<Callable>(callable));
275 }
276 277 //! Register cleanup function to run on asynchronous worker thread without
278 //! blocking the event loop thread.
279 void addAsyncCleanup(std::function<void()> fn);
280 281 //! Start asynchronous worker thread if necessary. This is only done if
282 //! there are ProxyServerBase::m_impl objects that need to be destroyed
283 //! asynchronously, without tying up the event loop thread. This can happen
284 //! when an interface does not declare a destroy() method that would allow
285 //! the client to wait for the destructor to finish and run it on a
286 //! dedicated thread. It can also happen whenever this is a broken
287 //! connection and the client is no longer around to call the destructors
288 //! and the server objects need to be garbage collected. In both cases, it
289 //! is important that ProxyServer::m_impl destructors do not run on the
290 //! eventloop thread because they may need it to do I/O if they perform
291 //! other IPC calls.
292 void startAsyncThread() MP_REQUIRES(m_mutex);
293 294 //! Check if loop should exit.
295 bool done() const MP_REQUIRES(m_mutex);
296 297 //! Process name included in thread names so combined debug output from
298 //! multiple processes is easier to understand.
299 const char* m_exe_name;
300 301 //! ID of the event loop thread
302 std::thread::id m_thread_id = std::this_thread::get_id();
303 304 //! Handle of an async worker thread. Joined on destruction. Unset if async
305 //! method has not been called.
306 std::thread m_async_thread;
307 308 //! Callback function to run on event loop thread during post() or sync() call.
309 kj::Function<void()>* m_post_fn MP_GUARDED_BY(m_mutex) = nullptr;
310 311 //! Callback functions to run on async thread.
312 std::optional<CleanupList> m_async_fns MP_GUARDED_BY(m_mutex);
313 314 //! Pipe read handle used to wake up the event loop thread.
315 int m_wait_fd = -1;
316 317 //! Pipe write handle used to wake up the event loop thread.
318 int m_post_fd = -1;
319 320 //! Number of EventLoopRef instances referencing this event loop. This is a
321 //! sum of the number of client and server objects (Connection, ProxyClient,
322 //! ProxyServer) using the loop, plus temporary references held while
323 //! posting functions to the loop, plus any references held by external code
324 //! to keep the loop running. The loop() method exits when this count drops
325 //! to 0 (and m_async_fns is empty).
326 int m_num_refs MP_GUARDED_BY(m_mutex) = 0;
327 328 //! Mutex and condition variable used to post tasks to event loop and async
329 //! thread.
330 Mutex m_mutex;
331 std::condition_variable m_cv;
332 333 //! Capnp IO context.
334 kj::AsyncIoContext m_io_context;
335 336 //! Capnp error handler. Needs to outlive m_task_set.
337 LoggingErrorHandler m_error_handler{*this};
338 339 //! Capnp list of pending promises.
340 std::unique_ptr<kj::TaskSet> m_task_set;
341 342 //! List of connections.
343 std::list<Connection> m_incoming_connections;
344 345 //! Logging options
346 LogOptions m_log_opts;
347 348 //! External context pointer.
349 void* m_context;
350 351 //! Hook called when ProxyServer<ThreadMap>::makeThread() is called.
352 std::function<void()> testing_hook_makethread;
353 354 //! Hook called on the worker thread inside makeThread(), after the thread
355 //! context is set up and thread_context promise is fulfilled, but before it
356 //! starts waiting for requests.
357 std::function<void()> testing_hook_makethread_created;
358 359 //! Hook called on the worker thread when it starts to execute an async
360 //! request. Used by tests to control timing or inject behavior at this
361 //! point in execution.
362 std::function<void()> testing_hook_async_request_start;
363 364 //! Hook called on the worker thread just before returning results.
365 std::function<void()> testing_hook_async_request_done;
366 367 //! Hook called on the event loop thread when a client has connected.
368 std::function<void()> testing_hook_connected;
369 370 //! Hook called on the event loop thread when a client has disconnected.
371 std::function<void()> testing_hook_disconnected;
372 };
373 374 //! Single element task queue used to handle recursive capnp calls. (If the
375 //! server makes a callback into the client in the middle of a request, while the client
376 //! thread is blocked waiting for server response, this is what allows the
377 //! client to run the request in the same thread, the same way code would run in a
378 //! single process, with the callback sharing the same thread stack as the original
379 //! call.) To support this, the clientInvoke function calls Waiter::wait() to
380 //! block the client IPC thread while initial request is in progress. Then if
381 //! there is a callback, it is executed with Waiter::post().
382 //!
383 //! The Waiter class is also used server-side by `ProxyServer<Thread>::post()`
384 //! to execute IPC calls on worker threads.
385 struct Waiter
386 {
387 Waiter() = default;
388 389 template <typename Fn>
390 bool post(Fn&& fn)
391 {
392 const Lock lock(m_mutex);
393 if (m_fn) return false;
394 m_fn = std::forward<Fn>(fn);
395 m_cv.notify_all();
396 return true;
397 }
398 399 template <class Predicate>
400 void wait(Lock& lock, Predicate pred) MP_REQUIRES(m_mutex)
401 {
402 m_cv.wait(lock.m_lock, [&]() MP_REQUIRES(m_mutex) {
403 // Important for this to be "while (m_fn)", not "if (m_fn)" to avoid
404 // a lost-wakeup bug. A new m_fn and m_cv notification might be sent
405 // after the fn() call and before the lock.lock() call in this loop
406 // in the case where a capnp response is sent and a brand new
407 // request is immediately received.
408 while (m_fn) {
409 auto fn = std::move(*m_fn);
410 m_fn.reset();
411 Unlock(lock, fn);
412 }
413 const bool done = pred();
414 return done;
415 });
416 }
417 418 //! Mutex mainly used internally by waiter class, but also used externally
419 //! to guard access to related state. Specifically, since the thread_local
420 //! ThreadContext struct owns a Waiter, the Waiter::m_mutex is used to guard
421 //! access to other parts of the struct to avoid needing to deal with more
422 //! mutexes than necessary. This mutex can be held at the same time as
423 //! EventLoop::m_mutex as long as Waiter::mutex is locked first and
424 //! EventLoop::m_mutex is locked second.
425 Mutex m_mutex;
426 std::condition_variable m_cv MP_GUARDED_BY(m_mutex);
427 std::optional<kj::Function<void()>> m_fn MP_GUARDED_BY(m_mutex);
428 };
429 430 //! Object holding network & rpc state associated with either an incoming server
431 //! connection, or an outgoing client connection. It must be created and destroyed
432 //! on the event loop thread.
433 //! In addition to Cap'n Proto state, it also holds lists of callbacks to run
434 //! when the connection is closed.
435 class Connection
436 {
437 public:
438 Connection(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream_)
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)) {}
442 Connection(EventLoop& loop,
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))) {}
448 449 //! Run cleanup functions. Must be called from the event loop thread. First
450 //! calls synchronous cleanup functions while blocked (to free capnp
451 //! Capability::Client handles owned by ProxyClient objects), then schedules
452 //! asynchronous cleanup functions to run in a worker thread (to run
453 //! destructors of m_impl instances owned by ProxyServer objects).
454 ~Connection();
455 456 //! Register synchronous cleanup function to run on event loop thread (with
457 //! access to capnp thread local variables) when disconnect() is called.
458 //! any new i/o.
459 CleanupIt addSyncCleanup(std::function<void()> fn);
460 void removeSyncCleanup(CleanupIt it);
461 462 //! Add disconnect handler.
463 template <typename F>
464 void onDisconnect(F&& f)
465 {
466 // Add disconnect handler to local TaskSet to ensure it is canceled and
467 // will never run after connection object is destroyed. But when disconnect
468 // handler fires, do not call the function f right away, instead add it
469 // to the EventLoop TaskSet to avoid "Promise callback destroyed itself"
470 // error in the typical case where f deletes this Connection object.
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))); }));
473 }
474 475 EventLoopRef m_loop;
476 kj::Own<kj::AsyncIoStream> m_stream;
477 LoggingErrorHandler m_error_handler{*m_loop};
478 //! TaskSet used to cancel the m_network.onDisconnect() handler for remote
479 //! disconnections, if the connection is closed locally first by deleting
480 //! this Connection object.
481 kj::TaskSet m_on_disconnect{m_error_handler};
482 ::capnp::TwoPartyVatNetwork m_network;
483 std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system;
484 485 // ThreadMap interface client, used to create a remote server thread when an
486 // client IPC call is being made for the first time from a new thread.
487 ThreadMap::Client m_thread_map{nullptr};
488 489 //! Collection of server-side IPC worker threads (ProxyServer<Thread> objects previously returned by
490 //! ThreadMap.makeThread) used to service requests to clients.
491 ::capnp::CapabilityServerSet<Thread> m_threads;
492 493 //! A thread created by makePool with associated pending work queue. Vector is filled once by makePool() and never resized.
494 struct PoolSlot {
495 Thread::Client client;
496 size_t depth{0};
497 };
498 std::vector<PoolSlot> m_thread_pool;
499 500 //! Canceler for canceling promises that we want to discard when the
501 //! connection is destroyed. This is used to interrupt method calls that are
502 //! still executing at time of disconnection.
503 kj::Canceler m_canceler;
504 505 //! Cleanup functions to run if connection is broken unexpectedly. List
506 //! will be empty if all ProxyClient are destroyed cleanly before the
507 //! connection is destroyed.
508 CleanupList m_sync_cleanup_fns;
509 };
510 511 //! Vat id for server side of connection. Required argument to RpcSystem::bootStrap()
512 //!
513 //! "Vat" is Cap'n Proto nomenclature for a host of various objects that facilitates
514 //! bidirectional communication with other vats; it is often but not always 1-1 with
515 //! processes. Cap'n Proto doesn't reference clients or servers per se; instead everything
516 //! is just a vat.
517 //!
518 //! See also: https://github.com/capnproto/capnproto/blob/9021f0c722b36cb11e3690b0860939255ebad39c/c%2B%2B/src/capnp/rpc.capnp#L42-L56
519 struct ServerVatId
520 {
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); }
525 };
526 527 template <typename Interface, typename Impl>
528 ProxyClientBase<Interface, Impl>::ProxyClientBase(typename Interface::Client client,
529 Connection* connection,
530 bool destroy_connection)
531 : m_client(std::move(client)), m_context(connection)
532 533 {
534 MP_LOG(*m_context.loop, Log::Debug) << "Creating " << CxxTypeName(*this) << " " << this;
535 // Handler for the connection getting destroyed before this client object.
536 auto disconnect_cb = m_context.connection->addSyncCleanup([this]() {
537 // Release client capability by move-assigning to temporary.
538 {
539 typename Interface::Client(std::move(m_client));
540 }
541 Lock lock{m_context.loop->m_mutex};
542 m_context.connection = nullptr;
543 });
544 545 // Two shutdown sequences are supported:
546 //
547 // - A normal sequence where client proxy objects are deleted by external
548 // code that no longer needs them
549 //
550 // - A garbage collection sequence where the connection or event loop shuts
551 // down while external code is still holding client references.
552 //
553 // The first case is handled here when m_context.connection is not null. The
554 // second case is handled by the disconnect_cb function, which sets
555 // m_context.connection to null so nothing happens here.
556 m_context.cleanup_fns.emplace_front([this, destroy_connection, disconnect_cb]{
557 {
558 // If the capnp interface defines a destroy method, call it to destroy
559 // the remote object, waiting for it to be deleted server side. If the
560 // capnp interface does not define a destroy method, this will just call
561 // an empty stub defined in the ProxyClientBase class and do nothing.
562 // Exceptions are caught and logged rather than propagated because
563 // ~ProxyClientBase is noexcept and the peer may be gone by the time
564 // this runs.
565 if (kj::runCatchingExceptions([&]{ Sub::destroy(*this); }) != nullptr) {
566 MP_LOG(*m_context.loop, Log::Warning) << "Remote destroy call failed during cleanup. Continuing.";
567 }
568 569 // FIXME: Could just invoke removed addCleanup fn here instead of duplicating code
570 m_context.loop->sync([&]() {
571 // Remove disconnect callback on cleanup so it doesn't run and try
572 // to access this object after it's destroyed. This call needs to
573 // run inside loop->sync() on the event loop thread because
574 // otherwise, if there were an ill-timed disconnect, the
575 // onDisconnect handler could fire and delete the Connection object
576 // before the removeSyncCleanup call.
577 if (m_context.connection) m_context.connection->removeSyncCleanup(disconnect_cb);
578 579 // Release client capability by move-assigning to temporary.
580 {
581 typename Interface::Client(std::move(m_client));
582 }
583 if (destroy_connection) {
584 delete m_context.connection;
585 m_context.connection = nullptr;
586 }
587 });
588 }
589 });
590 Sub::construct(*this);
591 }
592 593 template <typename Interface, typename Impl>
594 ProxyClientBase<Interface, Impl>::~ProxyClientBase() noexcept
595 {
596 MP_LOG(*m_context.loop, Log::Debug) << "Cleaning up " << CxxTypeName(*this) << " " << this;
597 CleanupRun(m_context.cleanup_fns);
598 MP_LOG(*m_context.loop, Log::Debug) << "Destroying " << CxxTypeName(*this) << " " << this;
599 }
600 601 template <typename Interface, typename Impl>
602 ProxyServerBase<Interface, Impl>::ProxyServerBase(std::shared_ptr<Impl> impl, Connection& connection)
603 : m_impl(std::move(impl)), m_context(&connection)
604 {
605 MP_LOG(*m_context.loop, Log::Debug) << "Creating " << CxxTypeName(*this) << " " << this;
606 assert(m_impl);
607 }
608 609 //! ProxyServer destructor, called from the EventLoop thread by Cap'n Proto
610 //! garbage collection code after there are no more references to this object.
611 //! This will typically happen when the corresponding ProxyClient object on the
612 //! other side of the connection is destroyed. It can also happen earlier if the
613 //! connection is broken or destroyed. In the latter case this destructor will
614 //! typically be called inside m_rpc_system.reset() call in the ~Connection
615 //! destructor while the Connection object still exists. However, because
616 //! ProxyServer objects are refcounted, and the Connection object could be
617 //! destroyed while asynchronous IPC calls are still in-flight, it's possible
618 //! for this destructor to be called after the Connection object no longer
619 //! exists, so it is NOT valid to dereference the m_context.connection pointer
620 //! from this function.
621 template <typename Interface, typename Impl>
622 ProxyServerBase<Interface, Impl>::~ProxyServerBase()
623 {
624 MP_LOG(*m_context.loop, Log::Debug) << "Cleaning up " << CxxTypeName(*this) << " " << this;
625 if (m_impl) {
626 // If impl is non-null at this point, it means no client is waiting for
627 // the m_impl server object to be destroyed synchronously. This can
628 // happen either if the interface did not define a "destroy" method (see
629 // invokeDestroy method below), or if a destroy method was defined, but
630 // the connection was broken before it could be called.
631 //
632 // In either case, be conservative and run the cleanup on an
633 // asynchronous thread, to avoid destructors or cleanup functions
634 // blocking or deadlocking the current EventLoop thread, since they
635 // could be making IPC calls.
636 //
637 // Technically this is a little too conservative since if the interface
638 // defines a "destroy" method, but the destroy method does not accept a
639 // Context parameter specifying a worker thread, the cleanup method
640 // would run on the EventLoop thread normally (when connection is
641 // unbroken), but will not run on the EventLoop thread now (when
642 // connection is broken). Probably some refactoring of the destructor
643 // and invokeDestroy function is possible to make this cleaner and more
644 // consistent.
645 m_context.loop->addAsyncCleanup([impl=std::move(m_impl), fns=std::move(m_context.cleanup_fns)]() mutable {
646 impl.reset();
647 CleanupRun(fns);
648 });
649 }
650 assert(m_context.cleanup_fns.empty());
651 MP_LOG(*m_context.loop, Log::Debug) << "Destroying " << CxxTypeName(*this) << " " << this;
652 }
653 654 //! If the capnp interface defined a special "destroy" method, as described the
655 //! ProxyClientBase class, this method will be called and synchronously destroy
656 //! m_impl before returning to the client.
657 //!
658 //! If the capnp interface does not define a "destroy" method, this will never
659 //! be called and the ~ProxyServerBase destructor will be responsible for
660 //! deleting m_impl asynchronously, whenever the ProxyServer object gets garbage
661 //! collected by Cap'n Proto.
662 //!
663 //! This method is called in the same way other proxy server methods are called,
664 //! via the serverInvoke function. Basically serverInvoke just calls this as a
665 //! substitute for a non-existent m_impl->destroy() method. If the destroy
666 //! method has any parameters or return values they will be handled in the
667 //! normal way by PassField/ReadField/BuildField functions. Particularly if a
668 //! Context.thread parameter was passed, this method will run on the worker
669 //! thread specified by the client. Otherwise it will run on the EventLoop
670 //! thread, like other server methods without an assigned thread.
671 template <typename Interface, typename Impl>
672 void ProxyServerBase<Interface, Impl>::invokeDestroy()
673 {
674 m_impl.reset();
675 CleanupRun(m_context.cleanup_fns);
676 }
677 678 //! Map from Connection to local or remote thread handle which will be used over
679 //! that connection. This map will typically only contain one entry, but can
680 //! contain multiple if a single thread makes IPC calls over multiple
681 //! connections. A std::optional value type is used to avoid the map needing to
682 //! be locked while ProxyClient<Thread> objects are constructed, see
683 //! ThreadContext "Synchronization note" below.
684 using ConnThreads = std::map<Connection*, std::optional<ProxyClient<Thread>>>;
685 using ConnThread = ConnThreads::iterator;
686 687 // Retrieve ProxyClient<Thread> object associated with this connection from a
688 // map, or create a new one and insert it into the map. Return map iterator and
689 // inserted bool.
690 std::tuple<ConnThread, bool> SetThread(GuardedRef<ConnThreads> threads, Connection* connection, const std::function<Thread::Client()>& make_thread);
691 692 //! The thread_local ThreadContext g_thread_context struct provides information
693 //! about individual threads and a way of communicating between them. Because
694 //! it's a thread local struct, each ThreadContext instance is initialized by
695 //! the thread that owns it.
696 //!
697 //! ThreadContext is used for any client threads created externally which make
698 //! IPC calls, and for server threads created by
699 //! ProxyServer<ThreadMap>::makeThread() which execute IPC calls for clients.
700 //!
701 //! In both cases, the struct holds information like the thread name, and a
702 //! Waiter object where the EventLoop can post incoming IPC requests to execute
703 //! on the thread. The struct also holds ConnThread maps associating the thread
704 //! with local and remote ProxyClient<Thread> objects.
705 struct ThreadContext
706 {
707 //! Identifying string for debug.
708 std::string thread_name;
709 710 //! Waiter object used to allow remote clients to execute code on this
711 //! thread. For server threads created by
712 //! ProxyServer<ThreadMap>::makeThread(), this is initialized in that
713 //! function. Otherwise, for client threads created externally, this is
714 //! initialized the first time the thread tries to make an IPC call. Having
715 //! a waiter is necessary for threads making IPC calls in case a server they
716 //! are calling expects them to execute a callback during the call, before
717 //! it sends a response.
718 //!
719 //! For IPC client threads, the Waiter pointer is never cleared and the Waiter
720 //! just gets destroyed when the thread does. For server threads created by
721 //! makeThread(), this pointer is set to null in the ~ProxyServer<Thread> as
722 //! a signal for the thread to exit and destroy itself. In both cases, the
723 //! same Waiter object is used across different calls and only created and
724 //! destroyed once for the lifetime of the thread.
725 std::unique_ptr<Waiter> waiter = nullptr;
726 727 //! When client is making a request to a server, this is the
728 //! `callbackThread` argument it passes in the request, used by the server
729 //! in case it needs to make callbacks into the client that need to execute
730 //! while the client is waiting. This will be set to a local thread object.
731 //!
732 //! Synchronization note: The callback_thread and request_thread maps are
733 //! only ever accessed internally by this thread's destructor and externally
734 //! by Cap'n Proto event loop threads. Since it's possible for IPC client
735 //! threads to make calls over different connections that could have
736 //! different event loops, these maps are guarded by Waiter::m_mutex in case
737 //! different event loop threads add or remove map entries simultaneously.
738 //! However, individual ProxyClient<Thread> objects in the maps will only be
739 //! associated with one event loop and guarded by EventLoop::m_mutex. So
740 //! Waiter::m_mutex does not need to be held while accessing individual
741 //! ProxyClient<Thread> instances, and may even need to be released to
742 //! respect lock order and avoid locking Waiter::m_mutex before
743 //! EventLoop::m_mutex.
744 ConnThreads callback_threads MP_GUARDED_BY(waiter->m_mutex);
745 746 //! When client is making a request to a server, this is the `thread`
747 //! argument it passes in the request, used to control which thread on
748 //! server will be responsible for executing it. If client call is being
749 //! made from a local thread, this will be a remote thread object returned
750 //! by makeThread. If a client call is being made from a thread currently
751 //! handling a server request, this will be set to the `callbackThread`
752 //! request thread argument passed in that request.
753 //!
754 //! Synchronization note: \ref callback_threads note applies here as well.
755 ConnThreads request_threads MP_GUARDED_BY(waiter->m_mutex);
756 757 //! Whether this thread is a capnp event loop thread. Not really used except
758 //! to assert false if there's an attempt to execute a blocking operation
759 //! which could deadlock the thread.
760 bool loop_thread = false;
761 };
762 763 template<typename T, typename Fn>
764 kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
765 {
766 auto ready = kj::newPromiseAndFulfiller<void>(); // Signaled when waiter is ready to post again.
767 auto cancel_monitor_ptr = kj::heap<CancelMonitor>();
768 CancelMonitor& cancel_monitor = *cancel_monitor_ptr;
769 // Keep a reference to the ProxyServer<Thread> instance by assigning it to
770 // the self variable. ProxyServer instances are reference-counted and if the
771 // client drops its reference, this variable keeps the instance alive until
772 // the thread finishes executing. The self variable needs to be destroyed on
773 // the event loop thread so it is freed in a sync() call below.
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>(); // Signaled when fn() is called, with its return value.
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 {
778 // Fulfill ready.promise now, as soon as the Waiter starts executing
779 // this lambda, so the next ProxyServer<Thread>::post() call can
780 // immediately call waiter->post(). It is important to do this
781 // before calling fn() because fn() can make an IPC call back to the
782 // client, which can make another IPC call to this server thread.
783 // (This typically happens when IPC methods take std::function
784 // parameters.) When this happens the second call to the server
785 // thread should not be blocked waiting for the first call.
786 m_loop->sync([ready_fulfiller = kj::mv(ready_fulfiller)]() mutable {
787 ready_fulfiller->fulfill();
788 ready_fulfiller = nullptr;
789 });
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 {
793 // Destroy CancelMonitor here before fulfilling or rejecting the
794 // promise so it doesn't get triggered when the promise is
795 // destroyed.
796 cancel_monitor_ptr = nullptr;
797 // Send results to the fulfiller. Technically it would be ok to
798 // skip this if promise was canceled, but it's simpler to just
799 // do it unconditionally.
800 KJ_IF_MAYBE(e, exception) {
801 assert(!result_value);
802 result_fulfiller->reject(kj::mv(*e));
803 } else {
804 assert(result_value);
805 result_fulfiller->fulfill(kj::mv(*result_value));
806 result_value.reset();
807 }
808 result_fulfiller = nullptr;
809 // Use evalLater to destroy the ProxyServer<Thread> self
810 // reference, if it is the last reference, because the
811 // ProxyServer<Thread> destructor needs to join the thread,
812 // which can't happen until this sync() block has exited.
813 m_loop->m_task_set->add(kj::evalLater([self = kj::mv(self)] {}));
814 });
815 });
816 // Assert that calling Waiter::post did not fail. It could only return
817 // false if a new function was posted before the previous one finished
818 // executing, but new functions are only posted when m_thread_ready is
819 // signaled, so this should never happen.
820 assert(posted);
821 return kj::mv(result.promise);
822 }).attach(kj::heap<CancelProbe>(cancel_monitor));
823 m_thread_ready = kj::mv(ready.promise);
824 return ret;
825 }
826 827 //! Given stream file descriptor, make a new ProxyClient object to send requests
828 //! over the stream. Also create a new Connection object embedded in the
829 //! client that is freed when the client is closed.
830 template <typename InitInterface>
831 std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, int fd)
832 {
833 typename InitInterface::Client init_client(nullptr);
834 std::unique_ptr<Connection> connection;
835 loop.sync([&] {
836 auto stream =
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();
841 connection->onDisconnect([&loop, connection_ptr] {
842 MP_LOG(loop, Log::Warning) << "IPC client: unexpected network disconnect.";
843 delete connection_ptr;
844 });
845 });
846 return std::make_unique<ProxyClient<InitInterface>>(
847 kj::mv(init_client), connection.release(), /* destroy_connection= */ true);
848 }
849 850 //! Given stream and init objects, construct a new ProxyServer object that
851 //! handles requests from the stream by calling the init object. Embed the
852 //! ProxyServer in a Connection object that is stored and erased if
853 //! disconnected. This should be called from the event loop thread.
854 template <typename InitInterface, typename InitImpl, typename OnDisconnect>
855 void _Serve(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream, InitImpl& init, OnDisconnect&& on_disconnect)
856 {
857 loop.m_incoming_connections.emplace_front(loop, kj::mv(stream), [&](Connection& connection) {
858 // Disable deleter so proxy server object doesn't attempt to delete the
859 // init implementation when the proxy client is destroyed or
860 // disconnected.
861 return kj::heap<ProxyServer<InitInterface>>(std::shared_ptr<InitImpl>(&init, [](InitImpl*){}), connection);
862 });
863 auto it = loop.m_incoming_connections.begin();
864 MP_LOG(loop, Log::Info) << "IPC server: socket connected.";
865 if (loop.testing_hook_connected) loop.testing_hook_connected();
866 it->onDisconnect([&loop, it, on_disconnect = std::forward<OnDisconnect>(on_disconnect)]() mutable {
867 MP_LOG(loop, Log::Info) << "IPC server: socket disconnected.";
868 loop.m_incoming_connections.erase(it);
869 on_disconnect();
870 if (loop.testing_hook_disconnected) loop.testing_hook_disconnected();
871 });
872 }
873 874 struct Listener
875 {
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) {}
878 879 bool atCapacity() const
880 {
881 return m_max_connections && m_active_connections >= *m_max_connections;
882 }
883 884 kj::Own<kj::ConnectionReceiver> m_receiver;
885 std::optional<size_t> m_max_connections;
886 size_t m_active_connections{0};
887 };
888 889 template <typename InitInterface, typename InitImpl>
890 void _Listen(const std::shared_ptr<Listener>& listener, EventLoop& loop, InitImpl& init)
891 {
892 if (listener->atCapacity()) return;
893 894 auto* receiver = listener->m_receiver.get();
895 loop.m_task_set->add(receiver->accept().then(
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);
903 });
904 _Listen<InitInterface>(listener, loop, init);
905 }));
906 }
907 908 //! Given stream file descriptor and an init object, handle requests on the
909 //! stream by calling methods on the Init object.
910 template <typename InitInterface, typename InitImpl>
911 void ServeStream(EventLoop& loop, int fd, InitImpl& init)
912 {
913 _Serve<InitInterface>(
914 loop,
915 loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
916 init,
917 [] {});
918 }
919 920 //! Given listening socket file descriptor and an init object, handle incoming
921 //! connections and requests by calling methods on the Init object.
922 template <typename InitInterface, typename InitImpl>
923 void ListenConnections(EventLoop& loop, int fd, InitImpl& init, std::optional<size_t> max_connections = std::nullopt)
924 {
925 loop.sync([&]() {
926 auto listener{std::make_shared<Listener>(
927 loop.m_io_context.lowLevelProvider->wrapListenSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
928 max_connections)};
929 _Listen<InitInterface>(listener, loop, init);
930 });
931 }
932 933 extern thread_local ThreadContext g_thread_context; // NOLINT(bitcoin-nontrivial-threadlocal)
934 // Silence nonstandard bitcoin tidy error "Variable with non-trivial destructor
935 // cannot be thread_local" which should not be a problem on modern platforms, and
936 // could lead to a small memory leak at worst on older ones.
937 938 } // namespace mp
939 940 #endif // MP_PROXY_IO_H
941