Inside Bitcoin

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.
385struct 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.
435class Connection
436{
437public:
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