Inside Bitcoin

1// Copyright (c) 2021-present The Bitcoin Core developers
2// Distributed under the MIT software license, see the accompanying
3// file COPYING or http://www.opensource.org/licenses/mit-license.php.
4
5#include <interfaces/init.h>
6#include <ipc/capnp/context.h>
7#include <ipc/capnp/init.capnp.h>
8#include <ipc/capnp/init.capnp.proxy.h>
9#include <ipc/capnp/protocol.h>
10#include <ipc/exception.h>
11#include <ipc/protocol.h>
12#include <kj/async.h>
13#include <mp/proxy-io.h>
14#include <mp/proxy-types.h>
15#include <mp/util.h>
16#include <util/log.h>
17#include <util/threadnames.h>
18
19#include <cassert>
20#include <cerrno>
21#include <future>
22#include <memory>
23#include <mutex>
24#include <optional>
25#include <string>
26#include <sys/socket.h>
27#include <system_error>
28#include <thread>
29
30namespace ipc {
31namespace capnp {
32namespace {
33
34mp::Log GetRequestedIPCLogLevel()
35{
36    if (util::log::ShouldTraceLog(BCLog::IPC)) return mp::Log::Trace;
37    if (util::log::ShouldDebugLog(BCLog::IPC)) return mp::Log::Debug;
38
39    // Info, Warning, and Error are logged unconditionally
40    return mp::Log::Info;
41}
42
43void IpcLogFn(mp::LogMessage message)
44{
45    switch (message.level) {
46    case mp::Log::Trace:
47        LogTrace(BCLog::IPC, "%s", message.message);
48        return;
49    case mp::Log::Debug:
50        LogDebug(BCLog::IPC, "%s", message.message);
51        return;
52    case mp::Log::Info:
53        LogInfo("ipc: %s", message.message);
54        return;
55    case mp::Log::Warning:
56        LogWarning("ipc: %s", message.message);
57        return;
58    case mp::Log::Error:
59        LogError("ipc: %s", message.message);
60        return;
61    case mp::Log::Raise:
62        LogError("ipc: %s", message.message);
63        throw Exception(message.message);
64    } // no default case, so the compiler can warn about missing cases
65
66    // Be conservative and assume that if MP ever adds a new log level, it
67    // should only be shown at our most verbose level.
68    LogTrace(BCLog::IPC, "%s", message.message);
69}
70
71class CapnpProtocol : public Protocol
72{
73public:
74    CapnpProtocol(const char* exe_name) : m_exe_name{exe_name} {}
75    ~CapnpProtocol() noexcept(true)
76    {
77        m_loop_ref.reset();
78        if (m_loop_thread.joinable()) m_loop_thread.join();
79        assert(!m_loop);
80    };
81    std::unique_ptr<interfaces::Init> connect(mp::Stream stream) override
82    {
83        startLoop();
84        return mp::ConnectStream<messages::Init>(*m_loop, std::move(stream));
85    }
86    void listen(mp::SocketId listen_fd, interfaces::Init& init) override
87    {
88        startLoop();
89        if (::listen(listen_fd, /*backlog=*/5) != 0) {
90            throw std::system_error(errno, std::system_category());
91        }
92        mp::ListenConnections<messages::Init>(*m_loop, listen_fd, init);
93    }
94    void serve(interfaces::Init& init, const std::function<mp::Stream()>& make_stream) override
95    {
96        assert(!m_loop);
97        mp::CurrentThread().thread_name = mp::ThreadName(m_exe_name);
98        mp::LogOptions opts = {
99            .log_fn = IpcLogFn,
100            .log_level = GetRequestedIPCLogLevel()
101        };
102        m_loop.emplace(m_exe_name, std::move(opts), &m_context);
103        mp::ServeStream<messages::Init>(*m_loop, make_stream(), init);
104        m_parent_connection = &m_loop->m_incoming_connections.back();
105        m_loop->loop();
106        m_loop.reset();
107    }
108    void disconnectIncoming() override
109    {
110        if (!m_loop) return;
111        // Delete incoming connections, except the connection to a parent
112        // process (if there is one), since a parent process should be able to
113        // monitor and control this process, even during shutdown.
114        m_loop->sync([&] {
115            m_loop->m_incoming_connections.remove_if([this](mp::Connection& c) { return &c != m_parent_connection; });
116        });
117    }
118    mp::Stream makeStream(mp::SocketId socket) override
119    {
120        startLoop();
121        return mp::MakeStream(*m_loop, socket);
122    }
123    void addCleanup(std::type_index type, void* iface, std::function<void()> cleanup) override
124    {
125        mp::ProxyTypeRegister::types().at(type)(iface).cleanup_fns.emplace_back(std::move(cleanup));
126    }
127    Context& context() override { return m_context; }
128    void startLoop()
129    {
130        if (m_loop) return;
131        std::promise<void> promise;
132        m_loop_thread = std::thread([&] {
133            util::ThreadRename("capnp-loop");
134            mp::LogOptions opts = {
135                .log_fn = IpcLogFn,
136                .log_level = GetRequestedIPCLogLevel()
137            };
138            m_loop.emplace(m_exe_name, std::move(opts), &m_context);
139            m_loop_ref.emplace(*m_loop);
140            promise.set_value();
141            m_loop->loop();
142            m_loop.reset();
143        });
144        promise.get_future().wait();
145    }
146    const char* m_exe_name;
147    Context m_context;
148    //! EventLoop object which manages I/O events for all connections.
149    std::optional<mp::EventLoop> m_loop;
150    //! Reference to the same EventLoop. Increments the loop’s refcount on
151    //! creation, decrements on destruction. The loop thread exits when the
152    //! refcount reaches 0. Other IPC objects also hold their own EventLoopRef.
153    std::optional<mp::EventLoopRef> m_loop_ref;
154    //! Connection to parent, if this is a child process spawned by a parent process.
155    mp::Connection* m_parent_connection{nullptr};
156    std::thread m_loop_thread;
157};
158} // namespace
159
160std::unique_ptr<Protocol> MakeCapnpProtocol(const char* exe_name) { return std::make_unique<CapnpProtocol>(exe_name); }
161} // namespace capnp
162} // namespace ipc
163