proxy.cpp

Go to the documentation of this file.

// Copyright (c) The Bitcoin Core developers
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
#include <mp/proxy.h>
 
#include <mp/proxy-io.h>
#include <mp/proxy-types.h>
#include <mp/proxy.capnp.h>
#include <mp/type-threadmap.h>
#include <mp/util.h>
 
#include <atomic>
#include <capnp/capability.h>
#include <capnp/common.h> // IWYU pragma: keep
#include <capnp/rpc.h>
#include <condition_variable>
#include <functional>
#include <future>
#include <kj/async.h>
#include <kj/async-io.h>
#include <kj/async-prelude.h>
#include <kj/common.h>
#include <kj/debug.h>
#include <kj/exception.h>
#include <kj/function.h>
#include <kj/io.h>
#include <kj/memory.h>
#include <kj/string.h>
#include <cstdint>
#include <map>
#include <memory>
#include <optional>
#include <stdexcept>
#include <string>
#include <thread>
#include <tuple>
#include <utility>
#include <vector>
 
namespace mp {
 
thread_local ThreadContext g_thread_context; // NOLINT(bitcoin-nontrivial-threadlocal)
 
ThreadContext& CurrentThread()
{
    return g_thread_context;
}
 
Stream MakeStream(EventLoop&loop, SocketId socket)
{
    Stream stream;
    loop.sync([&]() {
        stream = loop.m_io_context.lowLevelProvider->wrapSocketFd(socket, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP);
    });
    return stream;
}
 
void LoggingErrorHandler::taskFailed(kj::Exception&& exception)
{
    KJ_LOG(ERROR, "Uncaught exception in daemonized task.", exception);
    MP_LOG(m_loop, Log::Error) << "Uncaught exception in daemonized task.";
}
 
EventLoopRef::EventLoopRef(EventLoop& loop, Lock* lock) : m_loop(&loop), m_lock(lock)
{
    auto loop_lock{PtrOrValue{m_lock, m_loop->m_mutex}};
    loop_lock->assert_locked(m_loop->m_mutex);
    m_loop->m_num_refs += 1;
}
 
// Due to the conditionals in this function, MP_NO_TSA is required to avoid
// error "error: mutex 'loop_lock' is not held on every path through here
// [-Wthread-safety-analysis]"
void EventLoopRef::reset(bool relock) MP_NO_TSA
{
    if (auto* loop{m_loop}) {
        m_loop = nullptr;
        auto loop_lock{PtrOrValue{m_lock, loop->m_mutex}};
        loop_lock->assert_locked(loop->m_mutex);
        assert(loop->m_num_refs > 0);
        loop->m_num_refs -= 1;
        if (loop->done()) {
            loop->m_cv.notify_all();
            // Capture loop->m_post_writer pointer before releasing the lock.
            // The pointer can't actually change before the write() call below,
            // but copying it with the lock held instead of accessing it
            // directly below prevents TSAN false positives because TSAN is
            // unaware of socketpair write() synchronization and might falsely
            // report the pointer being used in this thread and assigned in the
            // other thread without synchronization between.
            kj::OutputStream* post_writer{loop->m_post_writer.get()};
            loop_lock->unlock();
            char buffer = 0;
            // It safe to access post_writer here because the loop can't
            // exit until this write takes place. See "Intentionally do not
            // break..."  comment in EventLoop::loop
            post_writer->write(&buffer, 1);
            // By default, do not try to relock `loop_lock` after writing,
            // because the event loop could wake up and destroy itself and the
            // mutex might no longer exist.
            if (relock) loop_lock->lock();
        }
    }
}
 
ProxyContext::ProxyContext(Connection* connection) : connection(connection), loop{*connection->m_loop} {}
 
Connection::~Connection() noexcept(false)
{
    // Connection destructor is always called on the event loop thread. If this
    // is a local disconnect, it will trigger I/O, so this needs to run on the
    // event loop thread, and if there was a remote disconnect, this is called
    // by an onDisconnect callback directly from the event loop thread.
    assert(std::this_thread::get_id() == m_loop->m_thread_id);
 
    // Try to cancel any calls that may be executing.
    m_canceler.cancel("Interrupted by disconnect");
 
    // Shut down RPC system first, since this will garbage collect any
    // ProxyServer objects that were not freed before the connection was closed.
    // Typically all ProxyServer objects associated with this connection will be
    // freed before this call returns. However that will not be the case if
    // there are asynchronous IPC calls over this connection still currently
    // executing. In that case, Cap'n Proto will destroy the ProxyServer objects
    // after the calls finish.
    m_rpc_system.reset();
 
    // shutdownWrite is needed on Windows so pending data in the m_stream socket
    // will be sent instead of discarded when m_stream is destroyed. On unix,
    // this doesn't seem to be needed because data is sent more reliably.
    //
    // Sending pending data is important if the connection is a socketpair
    // because when one side of the socketpair is closed, the other side doesn't
    // seem to receive any onDisconnect event. So it is important for the other
    // side to instead receive Cap'n Proto "release" messages (see `struct
    // Release` in capnp/rpc.capnp) from local Client objects being destroyed so
    // the remote side can free resources and shut down cleanly. Without this,
    // when one side of a socket pair is closed the other side may not receive
    // these messages, preventing the remote side from freeing ProxyServer
    // resources and shutting down cleanly.
    // Use kj::runCatchingExceptions instead of try/catch because on macOS with
    // dynamic libraries, kj::Exception typeinfo differs between libcapnp and
    // the calling binary, so catch (const kj::Exception&) silently fails to
    // match. kj::runCatchingExceptions uses KJ's own interception mechanism
    // which works correctly across dynamic library boundaries.
    KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() {
        m_stream->shutdownWrite();
    })) {
        // Ignore ENOTCONN: on macOS/FreeBSD (unlike Linux), shutdown(SHUT_WR)
        // returns ENOTCONN if the peer already closed the connection. This is
        // expected when the destructor is triggered by a remote disconnect.
        // Also ignore UNIMPLEMENTED: shutdownWrite() may not be supported on
        // all stream types (e.g. streams that do not support half-close).
        if (exception->getType() != kj::Exception::Type::DISCONNECTED &&
            exception->getType() != kj::Exception::Type::UNIMPLEMENTED) {
            kj::throwRecoverableException(kj::mv(*exception));
        }
    }
 
    // ProxyClient cleanup handlers are in sync list, and ProxyServer cleanup
    // handlers are in the async list.
    //
    // The ProxyClient cleanup handlers are synchronous because they are fast
    // and don't do anything besides release capnp resources and reset state so
    // future calls to client methods immediately throw exceptions instead of
    // trying to communicate across the socket. The synchronous callbacks set
    // ProxyClient capability pointers to null, so new method calls on client
    // objects fail without triggering i/o or relying on event loop which may go
    // out of scope or trigger obscure capnp i/o errors.
    //
    // The ProxyServer cleanup handlers call user defined destructors on the server
    // object, which can run arbitrary blocking bitcoin code so they have to run
    // asynchronously in a different thread. The asynchronous cleanup functions
    // intentionally aren't started until after the synchronous cleanup
    // functions run, so client objects are fully disconnected before bitcoin
    // code in the destructors are run. This way if the bitcoin code tries to
    // make client requests the requests will just fail immediately instead of
    // sending i/o or accessing the event loop.
    //
    // The context where Connection objects are destroyed and this destructor is invoked
    // is different depending on whether this is an outgoing connection being used
    // to make an Init.makeX call() (e.g. Init.makeNode or Init.makeWalletClient) or an incoming
    // connection implementing the Init interface and handling the Init.makeX() calls.
    //
    // Either way when a connection is closed, capnp behavior is to call all
    // ProxyServer object destructors first, and then trigger an onDisconnect
    // callback.
    //
    // On incoming side of the connection, the onDisconnect callback is written
    // to delete the Connection object from the m_incoming_connections and call
    // this destructor which calls Connection::disconnect.
    //
    // On the outgoing side, the Connection object is owned by top level client
    // object client, which onDisconnect handler doesn't have ready access to,
    // so onDisconnect handler just calls Connection::disconnect directly
    // instead.
    //
    // Either way disconnect code runs in the event loop thread and called both
    // on clean and unclean shutdowns. In unclean shutdown case when the
    // connection is broken, sync and async cleanup lists will be filled with
    // callbacks. In the clean shutdown case both lists will be empty.
    Lock lock{m_loop->m_mutex};
    while (!m_sync_cleanup_fns.empty()) {
        CleanupList fn;
        fn.splice(fn.begin(), m_sync_cleanup_fns, m_sync_cleanup_fns.begin());
        Unlock(lock, fn.front());
    }
}
 
CleanupIt Connection::addSyncCleanup(std::function<void()> fn)
{
    const Lock lock(m_loop->m_mutex);
    // Add cleanup callbacks to the front of list, so sync cleanup functions run
    // in LIFO order. This is a good approach because sync cleanup functions are
    // added as client objects are created, and it is natural to clean up
    // objects in the reverse order they were created. In practice, however,
    // order should not be significant because the cleanup callbacks run
    // synchronously in a single batch when the connection is broken, and they
    // only reset the connection pointers in the client objects without actually
    // deleting the client objects.
    return m_sync_cleanup_fns.emplace(m_sync_cleanup_fns.begin(), std::move(fn));
}
 
void Connection::removeSyncCleanup(CleanupIt it)
{
    // Require cleanup functions to be removed on the event loop thread to avoid
    // needing to deal with them being removed in the middle of a disconnect.
    assert(std::this_thread::get_id() == m_loop->m_thread_id);
    const Lock lock(m_loop->m_mutex);
    m_sync_cleanup_fns.erase(it);
}
 
void EventLoop::addAsyncCleanup(std::function<void()> fn)
{
    const Lock lock(m_mutex);
    // Add async cleanup callbacks to the back of the list. Unlike the sync
    // cleanup list, this list order is more significant because it determines
    // the order server objects are destroyed when there is a sudden disconnect,
    // and it is possible objects may need to be destroyed in a certain order.
    // This function is called in ProxyServerBase destructors, and since capnp
    // destroys ProxyServer objects in LIFO order, we should preserve this
    // order, and add cleanup callbacks to the end of the list so they can be
    // run starting from the beginning of the list.
    //
    // In bitcoin core, running these callbacks in the right order is
    // particularly important for the wallet process, because it uses blocking
    // shared_ptrs and requires Chain::Notification pointers owned by the node
    // process to be destroyed before the WalletLoader objects owned by the node
    // process, otherwise shared pointer counts of the CWallet objects (which
    // inherit from Chain::Notification) will not be 1 when WalletLoader
    // destructor runs and it will wait forever for them to be released.
    m_async_fns->emplace_back(std::move(fn));
    startAsyncThread();
}
 
EventLoop::EventLoop(const char* exe_name, LogOptions log_opts, void* context)
    : m_exe_name(exe_name),
      m_io_context(kj::setupAsyncIo()),
      m_task_set(new kj::TaskSet(m_error_handler)),
      m_log_opts(std::move(log_opts)),
      m_context(context)
{
    auto pipe = m_io_context.provider->newTwoWayPipe();
    m_wait_stream = kj::mv(pipe.ends[0]);
    m_post_stream = kj::mv(pipe.ends[1]);
    KJ_IF_MAYBE(fd, m_post_stream->getFd()) {
        m_post_writer = kj::heap<kj::FdOutputStream>(*fd);
    } else {
        throw std::logic_error("Could not get file descriptor for new pipe.");
    }
}
 
EventLoop::~EventLoop()
{
    if (m_async_thread.joinable()) m_async_thread.join();
    const Lock lock(m_mutex);
    KJ_ASSERT(m_sync_fn == nullptr);
    KJ_ASSERT(!m_async_fns);
    KJ_ASSERT(!m_wait_stream);
    KJ_ASSERT(!m_post_stream);
    KJ_ASSERT(m_num_refs == 0);
 
    // Spin event loop. wait for any promises triggered by RPC shutdown.
    // auto cleanup = kj::evalLater([]{});
    // cleanup.wait(m_io_context.waitScope);
}
 
void EventLoop::loop()
{
    assert(!CurrentThread().loop_thread);
    CurrentThread().loop_thread = true;
    KJ_DEFER(CurrentThread().loop_thread = false);
 
    {
        const Lock lock(m_mutex);
        assert(!m_async_fns);
        m_async_fns.emplace();
    }
 
    kj::Own<kj::AsyncIoStream>& wait_stream{m_wait_stream};
    char buffer = 0;
    for (;;) {
        const size_t read_bytes = wait_stream->read(&buffer, 0, 1).wait(m_io_context.waitScope);
        if (read_bytes != 1) throw std::logic_error("EventLoop wait_stream closed unexpectedly");
        Lock lock(m_mutex);
        if (m_sync_fn) {
            // m_sync_fn throwing is never expected. If it does happen, the caller
            // of EventLoop::sync() will return without any indication of failure,
            // which will likely cause other bugs. Log the error and continue.
            KJ_IF_MAYBE(exception, kj::runCatchingExceptions([&]() MP_REQUIRES(m_mutex) { Unlock(lock, *m_sync_fn); })) {
                MP_LOG(*this, Log::Error) << "EventLoop: m_sync_fn threw: " << kj::str(*exception).cStr();
            }
            m_sync_fn = nullptr;
            m_cv.notify_all();
        } else if (done()) {
            // Intentionally do not break if m_sync_fn was set, even if done()
            // would return true, to ensure that the sync() m_post_writer->write()
            // call always succeeds and the loop does not exit between the time
            // that the done condition is set and the write call is made.
            break;
        }
    }
    MP_LOG(*this, Log::Info) << "EventLoop::loop done, cancelling event listeners.";
    m_task_set.reset();
    MP_LOG(*this, Log::Info) << "EventLoop::loop bye.";
    wait_stream = nullptr;
    const Lock lock(m_mutex);
    m_post_writer = nullptr;
    m_wait_stream = nullptr;
    m_post_stream = nullptr;
    m_async_fns.reset();
    m_cv.notify_all();
}
 
void EventLoop::sync(kj::FunctionParam<void()> fn)
{
    if (std::this_thread::get_id() == m_thread_id) {
        fn();
        return;
    }
    Lock lock(m_mutex);
    EventLoopRef ref(*this, &lock);
    m_cv.wait(lock.m_lock, [this]() MP_REQUIRES(m_mutex) { return m_sync_fn == nullptr; });
    m_sync_fn = &fn;
    Unlock(lock, [&] {
        char buffer = 0;
        m_post_writer->write(&buffer, 1);
    });
    m_cv.wait(lock.m_lock, [this, &fn]() MP_REQUIRES(m_mutex) { return m_sync_fn != &fn; });
}
 
void EventLoop::startAsyncThread()
{
    assert (std::this_thread::get_id() == m_thread_id);
    if (m_async_thread.joinable()) {
        // Notify to wake up the async thread if it is already running.
        m_cv.notify_all();
    } else if (!m_async_fns->empty()) {
        m_async_thread = std::thread([this] {
            SetOsThreadName("capnp-async");
            Lock lock(m_mutex);
            while (m_async_fns) {
                if (!m_async_fns->empty()) {
                    EventLoopRef ref{*this, &lock};
                    const std::function<void()> fn = std::move(m_async_fns->front());
                    m_async_fns->pop_front();
                    Unlock(lock, fn);
                    // Important to relock because of the wait() call below.
                    ref.reset(/*relock=*/true);
                    // Continue without waiting in case there are more async_fns
                    continue;
                }
                m_cv.wait(lock.m_lock);
            }
        });
    }
}
 
bool EventLoop::done() const
{
    assert(m_num_refs >= 0);
    return m_num_refs == 0 && m_async_fns->empty();
}
 
std::tuple<ConnThread, bool> SetThread(GuardedRef<ConnThreads> threads, Connection* connection, const std::function<Thread::Client()>& make_thread)
{
    assert(std::this_thread::get_id() == connection->m_loop->m_thread_id);
    ConnThread thread;
    bool inserted;
    {
        const Lock lock(threads.mutex);
        std::tie(thread, inserted) = threads.ref.try_emplace(connection);
    }
    if (inserted) {
        thread->second.emplace(make_thread(), connection, /* destroy_connection= */ false);
        thread->second->m_disconnect_cb = connection->addSyncCleanup([threads, thread] {
            // Note: it is safe to use the `thread` iterator in this cleanup
            // function, because the iterator would only be invalid if the map entry
            // was removed, and if the map entry is removed the ProxyClient<Thread>
            // destructor unregisters the cleanup.
 
            // Connection is being destroyed before thread client is, so reset
            // thread client m_disconnect_cb member so thread client destructor does not
            // try to unregister this callback after connection is destroyed.
            thread->second->m_disconnect_cb.reset();
 
            // Remove connection pointer about to be destroyed from the map
            const Lock lock(threads.mutex);
            threads.ref.erase(thread);
        });
    }
    return {thread, inserted};
}
 
ProxyClient<Thread>::~ProxyClient()
{
    // If thread is being destroyed before connection is destroyed, remove the
    // cleanup callback that was registered to handle the connection being
    // destroyed before the thread being destroyed.
    if (m_disconnect_cb) {
        // Remove disconnect callback on the event loop thread with
        // loop->sync(), so if the connection is broken there is not a race
        // between this thread trying to remove the callback and the disconnect
        // handler attempting to call it.
        m_context.loop->sync([&]() {
            if (m_disconnect_cb) {
                m_context.connection->removeSyncCleanup(*m_disconnect_cb);
            }
        });
    }
}
 
ProxyServer<Thread>::ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread)
    : m_loop{*connection.m_loop}, m_thread_context(thread_context), m_thread(std::move(thread))
{
    assert(m_thread_context.waiter.get() != nullptr);
}
 
ProxyServer<Thread>::~ProxyServer()
{
    if (!m_thread.joinable()) return;
    // Stop async thread and wait for it to exit. Need to wait because the
    // m_thread handle needs to outlive the thread to avoid "terminate called
    // without an active exception" error. An alternative to waiting would be
    // detach the thread, but this would introduce nondeterminism which could
    // make code harder to debug or extend.
    assert(m_thread_context.waiter.get());
    std::unique_ptr<Waiter> waiter;
    {
        const Lock lock(m_thread_context.waiter->m_mutex);
        waiter = std::move(m_thread_context.waiter);
        assert(!waiter->m_fn);
        // Clear client maps now to avoid deadlock in m_thread.join() call
        // below. The maps contain Thread::Client objects that need to be
        // destroyed from the event loop thread (this thread), which can't
        // happen if this thread is busy calling join.
        m_thread_context.request_threads.clear();
        m_thread_context.callback_threads.clear();
        waiter->m_cv.notify_all();
    }
    m_thread.join();
}
 
kj::Promise<void> ProxyServer<Thread>::getName(GetNameContext context)
{
    context.getResults().setResult(m_thread_context.thread_name);
    return kj::READY_NOW;
}
 
ProxyServer<ThreadMap>::ProxyServer(Connection& connection) : m_connection(connection) {}
 
kj::Promise<void> ProxyServer<ThreadMap>::makePool(MakePoolContext context)
{
    if (!m_connection.m_thread_pool.empty()) {
        throw std::runtime_error("makePool called on connection with existing pool");
    }
    EventLoop& loop{*m_connection.m_loop};
    const uint32_t count = context.getParams().getCount();
    for (uint32_t i = 0; i < count; ++i) {
        const std::string thread_name = "pool/" + std::to_string(i);
        std::promise<ThreadContext*> thread_context;
        std::thread thread([&loop, &thread_context, thread_name, i]() {
            SetOsThreadName(("capnp-pool-" + std::to_string(i)).c_str());
            CurrentThread().thread_name = ThreadName(loop.m_exe_name) + " (" + thread_name + ")";
            CurrentThread().waiter = std::make_unique<Waiter>();
            Lock lock(CurrentThread().waiter->m_mutex);
            thread_context.set_value(&CurrentThread());
            CurrentThread().waiter->wait(lock, [] { return !CurrentThread().waiter; });
        });
        auto thread_server = kj::heap<ProxyServer<Thread>>(m_connection, *thread_context.get_future().get(), std::move(thread));
        m_connection.m_thread_pool.push_back({m_connection.m_threads.add(kj::mv(thread_server))});
    }
    return kj::READY_NOW;
}
 
kj::Promise<void> ProxyServer<ThreadMap>::makeThread(MakeThreadContext context)
{
    EventLoop& loop{*m_connection.m_loop};
    if (loop.testing_hook_makethread) loop.testing_hook_makethread();
    const std::string from = context.getParams().getName();
    std::promise<ThreadContext*> thread_context;
    std::thread thread([&loop, &thread_context, from]() {
        SetOsThreadName("capnp-worker");
        CurrentThread().thread_name = ThreadName(loop.m_exe_name) + " (from " + from + ")";
        CurrentThread().waiter = std::make_unique<Waiter>();
        Lock lock(CurrentThread().waiter->m_mutex);
        thread_context.set_value(&CurrentThread());
        if (loop.testing_hook_makethread_created) loop.testing_hook_makethread_created();
        // Wait for shutdown signal from ProxyServer<Thread> destructor (signal
        // is just waiter getting set to null.)
        CurrentThread().waiter->wait(lock, [] { return !CurrentThread().waiter; });
    });
    auto thread_server = kj::heap<ProxyServer<Thread>>(m_connection, *thread_context.get_future().get(), std::move(thread));
    auto thread_client = m_connection.m_threads.add(kj::mv(thread_server));
    context.getResults().setResult(kj::mv(thread_client));
    return kj::READY_NOW;
}
 
std::atomic<int> server_reqs{0};
 
std::string LongThreadName(const char* exe_name)
{
    return CurrentThread().thread_name.empty() ? ThreadName(exe_name) : CurrentThread().thread_name;
}
 
kj::StringPtr KJ_STRINGIFY(Log v)
{
    switch (v) {
        case Log::Trace:   return "Trace";
        case Log::Debug:   return "Debug";
        case Log::Info:    return "Info";
        case Log::Warning: return "Warning";
        case Log::Error:   return "Error";
        case Log::Raise:   return "Raise";
    }
    return "<Log?>";
}
} // namespace mp