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;
 
    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(ThreadContext& thread_context, std::thread&& thread);
    ~ProxyServer();
    kj::Promise<void> getName(GetNameContext context) override;
    ThreadContext& m_thread_context;
    std::thread m_thread;
};
 
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;
 
    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;
};
 
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 cancelled 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 cases 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;
 
    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 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();
    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