proxy-io.h

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.
 
#ifndef MP_PROXY_IO_H
#define MP_PROXY_IO_H
 
#include <mp/proxy.h>
#include <mp/util.h>
 
#include <mp/proxy.capnp.h>
 
#include <capnp/rpc-twoparty.h>
 
#include <assert.h>
#include <condition_variable>
#include <functional>
#include <kj/function.h>
#include <map>
#include <memory>
#include <optional>
#include <sstream>
#include <string>
#include <thread>
 
namespace mp {
struct ThreadContext;
 
struct InvokeContext
{
    Connection& connection;
};
 
struct ClientInvokeContext : InvokeContext
{
    ThreadContext& thread_context;
    ClientInvokeContext(Connection& conn, ThreadContext& thread_context)
        : InvokeContext{conn}, thread_context{thread_context}
    {
    }
};
 
template <typename ProxyServer, typename CallContext_>
struct ServerInvokeContext : InvokeContext
{
    using CallContext = CallContext_;
 
    ProxyServer& proxy_server;
    CallContext& call_context;
    int req;
    Lock* cancel_lock{nullptr};
    bool request_canceled{false};
 
    ServerInvokeContext(ProxyServer& proxy_server, CallContext& call_context, int req)
        : InvokeContext{*proxy_server.m_context.connection}, proxy_server{proxy_server}, call_context{call_context}, req{req}
    {
    }
};
 
template <typename Interface, typename Params, typename Results>
using ServerContext = ServerInvokeContext<ProxyServer<Interface>, ::capnp::CallContext<Params, Results>>;
 
template <>
struct ProxyClient<Thread> : public ProxyClientBase<Thread, ::capnp::Void>
{
    using ProxyClientBase::ProxyClientBase;
    // https://stackoverflow.com/questions/22357887/comparing-two-mapiterators-why-does-it-need-the-copy-constructor-of-stdpair
    ProxyClient(const ProxyClient&) = delete;
    ~ProxyClient();

    std::optional<CleanupIt> m_disconnect_cb;
};
 
template <>
struct ProxyServer<Thread> final : public Thread::Server
{
public:
    ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread);
    ~ProxyServer();
    kj::Promise<void> getName(GetNameContext context) override;

    template<typename T, typename Fn>
    kj::Promise<T> post(Fn&& fn);
 
    EventLoopRef m_loop;
    ThreadContext& m_thread_context;
    std::thread m_thread;
    kj::Promise<void> m_thread_ready{kj::READY_NOW};
};

class LoggingErrorHandler : public kj::TaskSet::ErrorHandler
{
public:
    LoggingErrorHandler(EventLoop& loop) : m_loop(loop) {}
    void taskFailed(kj::Exception&& exception) override;
    EventLoop& m_loop;
};

enum class Log {
    Trace = 0,
    Debug,
    Info,
    Warning,
    Error,
    Raise,
};
 
kj::StringPtr KJ_STRINGIFY(Log flags);
 
struct LogMessage {

    std::string message;

    Log level;
};
 
using LogFn = std::function<void(LogMessage)>;
 
struct LogOptions {

    LogFn log_fn;

    //! Maximum number of characters to use when representing
    //! request and response structs as strings.
    size_t max_chars{200};

    Log log_level{Log::Trace};
};
 
class Logger
{
public:
    Logger(const LogOptions& options, Log log_level) : m_options(options), m_log_level(log_level) {}
 
    Logger(Logger&&) = delete;
    Logger& operator=(Logger&&) = delete;
    Logger(const Logger&) = delete;
    Logger& operator=(const Logger&) = delete;
 
    ~Logger() noexcept(false)
    {
        if (enabled()) m_options.log_fn({std::move(m_buffer).str(), m_log_level});
    }
 
    template <typename T>
    friend Logger& operator<<(Logger& logger, T&& value)
    {
        if (logger.enabled()) logger.m_buffer << std::forward<T>(value);
        return logger;
    }
 
    template <typename T>
    friend Logger& operator<<(Logger&& logger, T&& value)
    {
        return logger << std::forward<T>(value);
    }
 
    explicit operator bool() const
    {
        return enabled();
    }
 
private:
    bool enabled() const
    {
        return m_options.log_fn && m_log_level >= m_options.log_level;
    }
 
    const LogOptions& m_options;
    Log m_log_level;
    std::ostringstream m_buffer;
};
 
#define MP_LOGPLAIN(loop, ...) if (mp::Logger logger{(loop).m_log_opts, __VA_ARGS__}; logger) logger
 
#define MP_LOG(loop, ...) MP_LOGPLAIN(loop, __VA_ARGS__) << "{" << LongThreadName((loop).m_exe_name) << "} "
 
std::string LongThreadName(const char* exe_name);

class EventLoop
{
public:
    EventLoop(const char* exe_name, LogFn log_fn, void* context = nullptr)
        : EventLoop(exe_name, LogOptions{std::move(log_fn)}, context){}

    EventLoop(const char* exe_name, LogOptions log_opts, void* context = nullptr);

    EventLoop(const char* exe_name, std::function<void(bool, std::string)> old_callback, void* context = nullptr)
        : EventLoop(exe_name,
                LogFn{[old_callback = std::move(old_callback)](LogMessage log_data) {old_callback(log_data.level == Log::Raise, std::move(log_data.message));}},
                context){}
 
    ~EventLoop();

    void loop();

    void post(kj::Function<void()> fn);

    template <typename Callable>
    void sync(Callable&& callable)
    {
        post(std::forward<Callable>(callable));
    }

    void addAsyncCleanup(std::function<void()> fn);

    void startAsyncThread() MP_REQUIRES(m_mutex);

    bool done() const MP_REQUIRES(m_mutex);

    const char* m_exe_name;

    std::thread::id m_thread_id = std::this_thread::get_id();

    std::thread m_async_thread;

    kj::Function<void()>* m_post_fn MP_GUARDED_BY(m_mutex) = nullptr;

    std::optional<CleanupList> m_async_fns MP_GUARDED_BY(m_mutex);

    int m_wait_fd = -1;

    int m_post_fd = -1;

    int m_num_clients MP_GUARDED_BY(m_mutex) = 0;

    Mutex m_mutex;
    std::condition_variable m_cv;

    kj::AsyncIoContext m_io_context;

    LoggingErrorHandler m_error_handler{*this};

    std::unique_ptr<kj::TaskSet> m_task_set;

    std::list<Connection> m_incoming_connections;

    LogOptions m_log_opts;

    void* m_context;

    std::function<void()> testing_hook_makethread;

    std::function<void()> testing_hook_makethread_created;

    std::function<void()> testing_hook_async_request_start;

    std::function<void()> testing_hook_async_request_done;
};

struct Waiter
{
    Waiter() = default;
 
    template <typename Fn>
    bool post(Fn&& fn)
    {
        const Lock lock(m_mutex);
        if (m_fn) return false;
        m_fn = std::forward<Fn>(fn);
        m_cv.notify_all();
        return true;
    }
 
    template <class Predicate>
    void wait(Lock& lock, Predicate pred)
    {
        m_cv.wait(lock.m_lock, [&]() MP_REQUIRES(m_mutex) {
            // Important for this to be "while (m_fn)", not "if (m_fn)" to avoid
            // a lost-wakeup bug. A new m_fn and m_cv notification might be sent
            // after the fn() call and before the lock.lock() call in this loop
            // in the case where a capnp response is sent and a brand new
            // request is immediately received.
            while (m_fn) {
                auto fn = std::move(*m_fn);
                m_fn.reset();
                Unlock(lock, fn);
            }
            const bool done = pred();
            return done;
        });
    }

    Mutex m_mutex;
    std::condition_variable m_cv;
    std::optional<kj::Function<void()>> m_fn MP_GUARDED_BY(m_mutex);
};

class Connection
{
public:
    Connection(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream_)
        : m_loop(loop), m_stream(kj::mv(stream_)),
          m_network(*m_stream, ::capnp::rpc::twoparty::Side::CLIENT, ::capnp::ReaderOptions()),
          m_rpc_system(::capnp::makeRpcClient(m_network)) {}
    Connection(EventLoop& loop,
        kj::Own<kj::AsyncIoStream>&& stream_,
        const std::function<::capnp::Capability::Client(Connection&)>& make_client)
        : m_loop(loop), m_stream(kj::mv(stream_)),
          m_network(*m_stream, ::capnp::rpc::twoparty::Side::SERVER, ::capnp::ReaderOptions()),
          m_rpc_system(::capnp::makeRpcServer(m_network, make_client(*this))) {}

    ~Connection();

    CleanupIt addSyncCleanup(std::function<void()> fn);
    void removeSyncCleanup(CleanupIt it);

    template <typename F>
    void onDisconnect(F&& f)
    {
        // Add disconnect handler to local TaskSet to ensure it is canceled and
        // will never run after connection object is destroyed. But when disconnect
        // handler fires, do not call the function f right away, instead add it
        // to the EventLoop TaskSet to avoid "Promise callback destroyed itself"
        // error in the typical case where f deletes this Connection object.
        m_on_disconnect.add(m_network.onDisconnect().then(
            [f = std::forward<F>(f), this]() mutable { m_loop->m_task_set->add(kj::evalLater(kj::mv(f))); }));
    }
 
    EventLoopRef m_loop;
    kj::Own<kj::AsyncIoStream> m_stream;
    LoggingErrorHandler m_error_handler{*m_loop};
    kj::TaskSet m_on_disconnect{m_error_handler};
    ::capnp::TwoPartyVatNetwork m_network;
    std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system;
 
    // ThreadMap interface client, used to create a remote server thread when an
    // client IPC call is being made for the first time from a new thread.
    ThreadMap::Client m_thread_map{nullptr};

    ::capnp::CapabilityServerSet<Thread> m_threads;

    kj::Canceler m_canceler;

    CleanupList m_sync_cleanup_fns;
};

struct ServerVatId
{
    ::capnp::word scratch[4]{};
    ::capnp::MallocMessageBuilder message{scratch};
    ::capnp::rpc::twoparty::VatId::Builder vat_id{message.getRoot<::capnp::rpc::twoparty::VatId>()};
    ServerVatId() { vat_id.setSide(::capnp::rpc::twoparty::Side::SERVER); }
};
 
template <typename Interface, typename Impl>
ProxyClientBase<Interface, Impl>::ProxyClientBase(typename Interface::Client client,
    Connection* connection,
    bool destroy_connection)
    : m_client(std::move(client)), m_context(connection)
 
{
    // Handler for the connection getting destroyed before this client object.
    auto disconnect_cb = m_context.connection->addSyncCleanup([this]() {
        // Release client capability by move-assigning to temporary.
        {
            typename Interface::Client(std::move(m_client));
        }
        Lock lock{m_context.loop->m_mutex};
        m_context.connection = nullptr;
    });
 
    // Two shutdown sequences are supported:
    //
    // - A normal sequence where client proxy objects are deleted by external
    //   code that no longer needs them
    //
    // - A garbage collection sequence where the connection or event loop shuts
    //   down while external code is still holding client references.
    //
    // The first case is handled here when m_context.connection is not null. The
    // second case is handled by the disconnect_cb function, which sets
    // m_context.connection to null so nothing happens here.
    m_context.cleanup_fns.emplace_front([this, destroy_connection, disconnect_cb]{
    {
        // If the capnp interface defines a destroy method, call it to destroy
        // the remote object, waiting for it to be deleted server side. If the
        // capnp interface does not define a destroy method, this will just call
        // an empty stub defined in the ProxyClientBase class and do nothing.
        Sub::destroy(*this);
 
        // FIXME: Could just invoke removed addCleanup fn here instead of duplicating code
        m_context.loop->sync([&]() {
            // Remove disconnect callback on cleanup so it doesn't run and try
            // to access this object after it's destroyed. This call needs to
            // run inside loop->sync() on the event loop thread because
            // otherwise, if there were an ill-timed disconnect, the
            // onDisconnect handler could fire and delete the Connection object
            // before the removeSyncCleanup call.
            if (m_context.connection) m_context.connection->removeSyncCleanup(disconnect_cb);
 
            // Release client capability by move-assigning to temporary.
            {
                typename Interface::Client(std::move(m_client));
            }
            if (destroy_connection) {
                delete m_context.connection;
                m_context.connection = nullptr;
            }
        });
    }
    });
    Sub::construct(*this);
}
 
template <typename Interface, typename Impl>
ProxyClientBase<Interface, Impl>::~ProxyClientBase() noexcept
{
    CleanupRun(m_context.cleanup_fns);
}
 
template <typename Interface, typename Impl>
ProxyServerBase<Interface, Impl>::ProxyServerBase(std::shared_ptr<Impl> impl, Connection& connection)
    : m_impl(std::move(impl)), m_context(&connection)
{
    assert(m_impl);
}

template <typename Interface, typename Impl>
ProxyServerBase<Interface, Impl>::~ProxyServerBase()
{
    if (m_impl) {
        // If impl is non-null at this point, it means no client is waiting for
        // the m_impl server object to be destroyed synchronously. This can
        // happen either if the interface did not define a "destroy" method (see
        // invokeDestroy method below), or if a destroy method was defined, but
        // the connection was broken before it could be called.
        //
        // In either case, be conservative and run the cleanup on an
        // asynchronous thread, to avoid destructors or cleanup functions
        // blocking or deadlocking the current EventLoop thread, since they
        // could be making IPC calls.
        //
        // Technically this is a little too conservative since if the interface
        // defines a "destroy" method, but the destroy method does not accept a
        // Context parameter specifying a worker thread, the cleanup method
        // would run on the EventLoop thread normally (when connection is
        // unbroken), but will not run on the EventLoop thread now (when
        // connection is broken). Probably some refactoring of the destructor
        // and invokeDestroy function is possible to make this cleaner and more
        // consistent.
        m_context.loop->addAsyncCleanup([impl=std::move(m_impl), fns=std::move(m_context.cleanup_fns)]() mutable {
            impl.reset();
            CleanupRun(fns);
        });
    }
    assert(m_context.cleanup_fns.empty());
}

template <typename Interface, typename Impl>
void ProxyServerBase<Interface, Impl>::invokeDestroy()
{
    m_impl.reset();
    CleanupRun(m_context.cleanup_fns);
}

using ConnThreads = std::map<Connection*, std::optional<ProxyClient<Thread>>>;
using ConnThread = ConnThreads::iterator;
 
// Retrieve ProxyClient<Thread> object associated with this connection from a
// map, or create a new one and insert it into the map. Return map iterator and
// inserted bool.
std::tuple<ConnThread, bool> SetThread(GuardedRef<ConnThreads> threads, Connection* connection, const std::function<Thread::Client()>& make_thread);

struct ThreadContext
{
    std::string thread_name;

    std::unique_ptr<Waiter> waiter = nullptr;

    ConnThreads callback_threads MP_GUARDED_BY(waiter->m_mutex);

    ConnThreads request_threads MP_GUARDED_BY(waiter->m_mutex);

    bool loop_thread = false;
};
 
template<typename T, typename Fn>
kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
{
    auto ready = kj::newPromiseAndFulfiller<void>(); // Signaled when waiter is ready to post again.
    auto cancel_monitor_ptr = kj::heap<CancelMonitor>();
    CancelMonitor& cancel_monitor = *cancel_monitor_ptr;
    // Keep a reference to the ProxyServer<Thread> instance by assigning it to
    // the self variable. ProxyServer instances are reference-counted and if the
    // client drops its reference, this variable keeps the instance alive until
    // the thread finishes executing. The self variable needs to be destroyed on
    // the event loop thread so it is freed in a sync() call below.
    auto self = thisCap();
    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 {
        auto result = kj::newPromiseAndFulfiller<T>(); // Signaled when fn() is called, with its return value.
        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 {
            // Fulfill ready.promise now, as soon as the Waiter starts executing
            // this lambda, so the next ProxyServer<Thread>::post() call can
            // immediately call waiter->post(). It is important to do this
            // before calling fn() because fn() can make an IPC call back to the
            // client, which can make another IPC call to this server thread.
            // (This typically happens when IPC methods take std::function
            // parameters.) When this happens the second call to the server
            // thread should not be blocked waiting for the first call.
            m_loop->sync([ready_fulfiller = kj::mv(ready_fulfiller)]() mutable {
                ready_fulfiller->fulfill();
                ready_fulfiller = nullptr;
            });
            std::optional<T> result_value;
            kj::Maybe<kj::Exception> exception{kj::runCatchingExceptions([&]{ result_value.emplace(fn(*cancel_monitor_ptr)); })};
            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 {
                // Destroy CancelMonitor here before fulfilling or rejecting the
                // promise so it doesn't get triggered when the promise is
                // destroyed.
                cancel_monitor_ptr = nullptr;
                // Send results to the fulfiller. Technically it would be ok to
                // skip this if promise was canceled, but it's simpler to just
                // do it unconditionally.
                KJ_IF_MAYBE(e, exception) {
                    assert(!result_value);
                    result_fulfiller->reject(kj::mv(*e));
                } else {
                    assert(result_value);
                    result_fulfiller->fulfill(kj::mv(*result_value));
                    result_value.reset();
                }
                result_fulfiller = nullptr;
                // Use evalLater to destroy the ProxyServer<Thread> self
                // reference, if it is the last reference, because the
                // ProxyServer<Thread> destructor needs to join the thread,
                // which can't happen until this sync() block has exited.
                m_loop->m_task_set->add(kj::evalLater([self = kj::mv(self)] {}));
            });
        });
        // Assert that calling Waiter::post did not fail. It could only return
        // false if a new function was posted before the previous one finished
        // executing, but new functions are only posted when m_thread_ready is
        // signaled, so this should never happen.
        assert(posted);
        return kj::mv(result.promise);
    }).attach(kj::heap<CancelProbe>(cancel_monitor));
    m_thread_ready = kj::mv(ready.promise);
    return ret;
}

template <typename InitInterface>
std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, int fd)
{
    typename InitInterface::Client init_client(nullptr);
    std::unique_ptr<Connection> connection;
    loop.sync([&] {
        auto stream =
            loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP);
        connection = std::make_unique<Connection>(loop, kj::mv(stream));
        init_client = connection->m_rpc_system->bootstrap(ServerVatId().vat_id).castAs<InitInterface>();
        Connection* connection_ptr = connection.get();
        connection->onDisconnect([&loop, connection_ptr] {
            MP_LOG(loop, Log::Warning) << "IPC client: unexpected network disconnect.";
            delete connection_ptr;
        });
    });
    return std::make_unique<ProxyClient<InitInterface>>(
        kj::mv(init_client), connection.release(), /* destroy_connection= */ true);
}

template <typename InitInterface, typename InitImpl>
void _Serve(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream, InitImpl& init)
{
    loop.m_incoming_connections.emplace_front(loop, kj::mv(stream), [&](Connection& connection) {
        // Disable deleter so proxy server object doesn't attempt to delete the
        // init implementation when the proxy client is destroyed or
        // disconnected.
        return kj::heap<ProxyServer<InitInterface>>(std::shared_ptr<InitImpl>(&init, [](InitImpl*){}), connection);
    });
    auto it = loop.m_incoming_connections.begin();
    MP_LOG(loop, Log::Info) << "IPC server: socket connected.";
    it->onDisconnect([&loop, it] {
        MP_LOG(loop, Log::Info) << "IPC server: socket disconnected.";
        loop.m_incoming_connections.erase(it);
    });
}

template <typename InitInterface, typename InitImpl>
void _Listen(EventLoop& loop, kj::Own<kj::ConnectionReceiver>&& listener, InitImpl& init)
{
    auto* ptr = listener.get();
    loop.m_task_set->add(ptr->accept().then(
        [&loop, &init, listener = kj::mv(listener)](kj::Own<kj::AsyncIoStream>&& stream) mutable {
            _Serve<InitInterface>(loop, kj::mv(stream), init);
            _Listen<InitInterface>(loop, kj::mv(listener), init);
        }));
}

template <typename InitInterface, typename InitImpl>
void ServeStream(EventLoop& loop, int fd, InitImpl& init)
{
    _Serve<InitInterface>(
        loop, loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP), init);
}

template <typename InitInterface, typename InitImpl>
void ListenConnections(EventLoop& loop, int fd, InitImpl& init)
{
    loop.sync([&]() {
        _Listen<InitInterface>(loop,
            loop.m_io_context.lowLevelProvider->wrapListenSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
            init);
    });
}
 
extern thread_local ThreadContext g_thread_context; // NOLINT(bitcoin-nontrivial-threadlocal)
// Silence nonstandard bitcoin tidy error "Variable with non-trivial destructor
// cannot be thread_local" which should not be a problem on modern platforms, and
// could lead to a small memory leak at worst on older ones.
 
} // namespace mp
 
#endif // MP_PROXY_IO_H