Branch data Line data Source code
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 : :
30 : : namespace ipc {
31 : : namespace capnp {
32 : : namespace {
33 : :
34 : 2 : mp::Log GetRequestedIPCLogLevel()
35 : : {
36 [ - + ]: 2 : if (util::log::ShouldTraceLog(BCLog::IPC)) return mp::Log::Trace;
37 [ # # ]: 0 : 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 : :
43 : 430 : void IpcLogFn(mp::LogMessage message)
44 : : {
45 [ + + + - : 430 : switch (message.level) {
- - - ]
46 : 96 : case mp::Log::Trace:
47 [ + - ]: 96 : LogTrace(BCLog::IPC, "%s", message.message);
48 : : return;
49 : 318 : case mp::Log::Debug:
50 [ + - ]: 318 : LogDebug(BCLog::IPC, "%s", message.message);
51 : : return;
52 : 16 : case mp::Log::Info:
53 : 16 : LogInfo("ipc: %s", message.message);
54 : 16 : return;
55 : 0 : case mp::Log::Warning:
56 : 0 : LogWarning("ipc: %s", message.message);
57 : 0 : return;
58 : 0 : case mp::Log::Error:
59 : 0 : LogError("ipc: %s", message.message);
60 : 0 : return;
61 : 0 : case mp::Log::Raise:
62 : 0 : LogError("ipc: %s", message.message);
63 [ # # ]: 0 : 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 [ - - ]: 430 : LogTrace(BCLog::IPC, "%s", message.message);
69 : : }
70 : :
71 : : class CapnpProtocol : public Protocol
72 : : {
73 : : public:
74 : 2 : CapnpProtocol(const char* exe_name) : m_exe_name{exe_name} {}
75 : 4 : ~CapnpProtocol() noexcept(true)
76 : 2 : {
77 : 2 : m_loop_ref.reset();
78 [ + + ]: 2 : if (m_loop_thread.joinable()) m_loop_thread.join();
79 [ - + ]: 2 : assert(!m_loop);
80 : 4 : };
81 : 6 : std::unique_ptr<interfaces::Init> connect(mp::Stream stream) override
82 : : {
83 : 6 : startLoop();
84 [ - + ]: 6 : return mp::ConnectStream<messages::Init>(*m_loop, std::move(stream));
85 : : }
86 : 2 : void listen(mp::SocketId listen_fd, interfaces::Init& init) override
87 : : {
88 : 2 : startLoop();
89 [ - + ]: 2 : if (::listen(listen_fd, /*backlog=*/5) != 0) {
90 [ # # ]: 0 : throw std::system_error(errno, std::system_category());
91 : : }
92 : 2 : mp::ListenConnections<messages::Init>(*m_loop, listen_fd, init);
93 : 2 : }
94 : 1 : void serve(interfaces::Init& init, const std::function<mp::Stream()>& make_stream) override
95 : : {
96 [ - + ]: 1 : assert(!m_loop);
97 : 2 : mp::CurrentThread().thread_name = mp::ThreadName(m_exe_name);
98 [ + - ]: 1 : mp::LogOptions opts = {
99 : : .log_fn = IpcLogFn,
100 : 1 : .log_level = GetRequestedIPCLogLevel()
101 [ + - ]: 1 : };
102 [ + - ]: 1 : m_loop.emplace(m_exe_name, std::move(opts), &m_context);
103 [ + - + - ]: 1 : mp::ServeStream<messages::Init>(*m_loop, make_stream(), init);
104 [ + - ]: 1 : m_parent_connection = &m_loop->m_incoming_connections.back();
105 [ + - ]: 1 : m_loop->loop();
106 : 1 : m_loop.reset();
107 : 1 : }
108 : 0 : void disconnectIncoming() override
109 : : {
110 [ # # ]: 0 : 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 : 0 : m_loop->sync([&] {
115 [ # # ]: 0 : m_loop->m_incoming_connections.remove_if([this](mp::Connection& c) { return &c != m_parent_connection; });
116 : : });
117 : : }
118 : 7 : mp::Stream makeStream(mp::SocketId socket) override
119 : : {
120 : 7 : startLoop();
121 : 7 : return mp::MakeStream(*m_loop, socket);
122 : : }
123 : 0 : void addCleanup(std::type_index type, void* iface, std::function<void()> cleanup) override
124 : : {
125 : 0 : mp::ProxyTypeRegister::types().at(type)(iface).cleanup_fns.emplace_back(std::move(cleanup));
126 : 0 : }
127 : 0 : Context& context() override { return m_context; }
128 : 15 : void startLoop()
129 : : {
130 [ + + ]: 15 : if (m_loop) return;
131 : 1 : std::promise<void> promise;
132 : 1 : m_loop_thread = std::thread([&] {
133 [ + - ]: 1 : util::ThreadRename("capnp-loop");
134 [ + - ]: 1 : mp::LogOptions opts = {
135 : : .log_fn = IpcLogFn,
136 : 1 : .log_level = GetRequestedIPCLogLevel()
137 [ + - ]: 1 : };
138 [ + - ]: 1 : m_loop.emplace(m_exe_name, std::move(opts), &m_context);
139 [ + - ]: 1 : m_loop_ref.emplace(*m_loop);
140 [ + - ]: 1 : promise.set_value();
141 [ + - ]: 1 : m_loop->loop();
142 : 1 : m_loop.reset();
143 [ + - ]: 2 : });
144 [ + - + - ]: 2 : promise.get_future().wait();
145 : 1 : }
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 : :
160 : 2 : std::unique_ptr<Protocol> MakeCapnpProtocol(const char* exe_name) { return std::make_unique<CapnpProtocol>(exe_name); }
161 : : } // namespace capnp
162 : : } // namespace ipc
|
