protocol.cpp

Go to the documentation of this file.

// Copyright (c) 2021-present 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 <interfaces/init.h>
#include <ipc/capnp/context.h>
#include <ipc/capnp/init.capnp.h>
#include <ipc/capnp/init.capnp.proxy.h>
#include <ipc/capnp/protocol.h>
#include <ipc/exception.h>
#include <ipc/protocol.h>
#include <kj/async.h>
#include <mp/proxy-io.h>
#include <mp/proxy-types.h>
#include <mp/util.h>
#include <util/log.h>
#include <util/threadnames.h>
 
#include <cassert>
#include <cerrno>
#include <future>
#include <list>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <sys/socket.h>
#include <system_error>
#include <thread>
 
namespace ipc {
namespace capnp {
namespace {
 
mp::Log GetRequestedIPCLogLevel()
{
    if (util::log::ShouldTraceLog(BCLog::IPC)) return mp::Log::Trace;
    if (util::log::ShouldDebugLog(BCLog::IPC)) return mp::Log::Debug;
 
    // Info, Warning, and Error are logged unconditionally
    return mp::Log::Info;
}
 
void IpcLogFn(mp::LogMessage message)
{
    switch (message.level) {
    case mp::Log::Trace:
        LogTrace(BCLog::IPC, "%s", message.message);
        return;
    case mp::Log::Debug:
        LogDebug(BCLog::IPC, "%s", message.message);
        return;
    case mp::Log::Info:
        LogInfo("ipc: %s", message.message);
        return;
    case mp::Log::Warning:
        LogWarning("ipc: %s", message.message);
        return;
    case mp::Log::Error:
        LogError("ipc: %s", message.message);
        return;
    case mp::Log::Raise:
        LogError("ipc: %s", message.message);
        throw Exception(message.message);
    } // no default case, so the compiler can warn about missing cases
 
    // Be conservative and assume that if MP ever adds a new log level, it
    // should only be shown at our most verbose level.
    LogTrace(BCLog::IPC, "%s", message.message);
}
 
class CapnpProtocol : public Protocol
{
public:
    CapnpProtocol(const char* exe_name) : m_exe_name{exe_name} {}
    ~CapnpProtocol() noexcept(true)
    {
        m_loop_ref.reset();
        if (m_loop_thread.joinable()) m_loop_thread.join();
        assert(!m_loop);
    };
    std::unique_ptr<interfaces::Init> connect(mp::Stream stream) override
    {
        startLoop();
        return mp::ConnectStream<messages::Init>(*m_loop, std::move(stream));
    }
    void listen(mp::SocketId listen_fd, interfaces::Init& init, std::optional<size_t> max_connections) override
    {
        startLoop();
        if (::listen(listen_fd, /*backlog=*/5) != 0) {
            throw std::system_error(errno, std::system_category());
        }
        mp::ListenConnections<messages::Init>(*m_loop, listen_fd, init, max_connections);
    }
    void serve(interfaces::Init& init, const std::function<mp::Stream()>& make_stream) override
    {
        assert(!m_loop);
        mp::CurrentThread().thread_name = mp::ThreadName(m_exe_name);
        mp::LogOptions opts = {
            .log_fn = IpcLogFn,
            .log_level = GetRequestedIPCLogLevel()
        };
        m_loop.emplace(m_exe_name, std::move(opts), &m_context);
        mp::ServeStream<messages::Init>(*m_loop, make_stream(), init);
        m_parent_connection = &m_loop->m_incoming_connections.back();
        m_loop->loop();
        m_loop.reset();
    }
    void disconnectIncoming() override
    {
        if (!m_loop) return;
        // Delete incoming connections, except the connection to a parent
        // process (if there is one), since a parent process should be able to
        // monitor and control this process, even during shutdown.
        m_loop->sync([&] {
            std::erase_if(m_loop->m_incoming_connections, [this](mp::Connection& c) { return &c != m_parent_connection; });
        });
    }
    mp::Stream makeStream(mp::SocketId socket) override
    {
        startLoop();
        return mp::MakeStream(*m_loop, socket);
    }
    void addCleanup(std::type_index type, void* iface, std::function<void()> cleanup) override
    {
        mp::ProxyTypeRegister::types().at(type)(iface).cleanup_fns.emplace_back(std::move(cleanup));
    }
    Context& context() override { return m_context; }
    void startLoop()
    {
        if (m_loop) return;
        std::promise<void> promise;
        m_loop_thread = std::thread([&] {
            util::ThreadRename("capnp-loop");
            mp::LogOptions opts = {
                .log_fn = IpcLogFn,
                .log_level = GetRequestedIPCLogLevel()
            };
            m_loop.emplace(m_exe_name, std::move(opts), &m_context);
            m_loop_ref.emplace(*m_loop);
            promise.set_value();
            m_loop->loop();
            m_loop.reset();
        });
        promise.get_future().wait();
    }
    const char* m_exe_name;
    Context m_context;
    std::optional<mp::EventLoop> m_loop;
    std::optional<mp::EventLoopRef> m_loop_ref;
    mp::Connection* m_parent_connection{nullptr};
    std::thread m_loop_thread;
};
} // namespace
 
std::unique_ptr<Protocol> MakeCapnpProtocol(const char* exe_name) { return std::make_unique<CapnpProtocol>(exe_name); }
} // namespace capnp
} // namespace ipc