Branch data Line data Source code
1 : : // Copyright (c) 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 : : #ifndef MP_PROXY_IO_H
6 : : #define MP_PROXY_IO_H
7 : :
8 : : #include <mp/proxy.h>
9 : : #include <mp/util.h>
10 : :
11 : : #include <mp/proxy.capnp.h>
12 : :
13 : : #include <capnp/rpc-twoparty.h>
14 : :
15 : : #include <assert.h>
16 : : #include <algorithm>
17 : : #include <condition_variable>
18 : : #include <cstdlib>
19 : : #include <functional>
20 : : #include <kj/function.h>
21 : : #include <map>
22 : : #include <memory>
23 : : #include <optional>
24 : : #include <sstream>
25 : : #include <string>
26 : : #include <thread>
27 : :
28 : : namespace mp {
29 : : struct ThreadContext;
30 : : struct Listener;
31 : :
32 : : struct InvokeContext
33 : : {
34 : : Connection& connection;
35 : : };
36 : :
37 : : struct ClientInvokeContext : InvokeContext
38 : : {
39 : : ThreadContext& thread_context;
40 : 0 : ClientInvokeContext(Connection& conn, ThreadContext& thread_context)
41 [ # # # # : 0 : : InvokeContext{conn}, thread_context{thread_context}
# # # # #
# ]
42 : : {
43 : : }
44 : : };
45 : :
46 : : template <typename ProxyServer, typename CallContext_>
47 : : struct ServerInvokeContext : InvokeContext
48 : : {
49 : : using CallContext = CallContext_;
50 : :
51 : : ProxyServer& proxy_server;
52 : : CallContext& call_context;
53 : : int req;
54 : : //! For IPC methods that execute asynchronously, not on the event-loop
55 : : //! thread: lock preventing the event-loop thread from freeing the params or
56 : : //! results structs if the request is canceled while the worker thread is
57 : : //! reading params (`call_context.getParams()`) or writing results
58 : : //! (`call_context.getResults()`).
59 : : Lock* cancel_lock{nullptr};
60 : : //! For IPC methods that execute asynchronously, not on the event-loop
61 : : //! thread, this is set to true if the IPC call was canceled by the client
62 : : //! or canceled by a disconnection. If the call runs on the event-loop
63 : : //! thread, it can't be canceled. This should be accessed with cancel_lock
64 : : //! held if it is not null, since in the asynchronous case it is accessed
65 : : //! from multiple threads.
66 : : bool request_canceled{false};
67 : :
68 : 0 : ServerInvokeContext(ProxyServer& proxy_server, CallContext& call_context, int req)
69 [ # # # # : 0 : : InvokeContext{*proxy_server.m_context.connection}, proxy_server{proxy_server}, call_context{call_context}, req{req}
# # # # #
# # # ]
70 : : {
71 : : }
72 : : };
73 : :
74 : : template <typename Interface, typename Params, typename Results>
75 : : using ServerContext = ServerInvokeContext<ProxyServer<Interface>, ::capnp::CallContext<Params, Results>>;
76 : :
77 : : template <>
78 : : struct ProxyClient<Thread> : public ProxyClientBase<Thread, ::capnp::Void>
79 : : {
80 [ # # ]: 0 : using ProxyClientBase::ProxyClientBase;
81 : : // https://stackoverflow.com/questions/22357887/comparing-two-mapiterators-why-does-it-need-the-copy-constructor-of-stdpair
82 : : ProxyClient(const ProxyClient&) = delete;
83 : : ~ProxyClient();
84 : :
85 : : //! Reference to callback function that is run if there is a sudden
86 : : //! disconnect and the Connection object is destroyed before this
87 : : //! ProxyClient<Thread> object. The callback will destroy this object and
88 : : //! remove its entry from the thread's request_threads or callback_threads
89 : : //! map. It will also reset m_disconnect_cb so the destructor does not
90 : : //! access it. In the normal case where there is no sudden disconnect, the
91 : : //! destructor will unregister m_disconnect_cb so the callback is never run.
92 : : //! Since this variable is accessed from multiple threads, accesses should
93 : : //! be guarded with the associated Waiter::m_mutex.
94 : : std::optional<CleanupIt> m_disconnect_cb;
95 : : };
96 : :
97 : : template <>
98 : : struct ProxyServer<Thread> final : public Thread::Server
99 : : {
100 : : public:
101 : : ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread);
102 : : ~ProxyServer();
103 : : kj::Promise<void> getName(GetNameContext context) override;
104 : :
105 : : //! Run a callback function fn returning T on this thread. The function will
106 : : //! be queued and executed as soon as the thread is idle, and when fn
107 : : //! returns, the promise returned by this method will be fulfilled with the
108 : : //! value fn returned.
109 : : template<typename T, typename Fn>
110 : : kj::Promise<T> post(Fn&& fn);
111 : :
112 : : EventLoopRef m_loop;
113 : : ThreadContext& m_thread_context;
114 : : std::thread m_thread;
115 : : //! Promise signaled when m_thread_context.waiter is ready and there is no
116 : : //! post() callback function waiting to execute.
117 : : kj::Promise<void> m_thread_ready{kj::READY_NOW};
118 : : };
119 : :
120 : : //! Handler for kj::TaskSet failed task events.
121 : : class LoggingErrorHandler : public kj::TaskSet::ErrorHandler
122 : : {
123 : : public:
124 [ + - ]: 3 : LoggingErrorHandler(EventLoop& loop) : m_loop(loop) {}
125 : : void taskFailed(kj::Exception&& exception) override;
126 : : EventLoop& m_loop;
127 : : };
128 : :
129 : : //! Log flags. Update stringify function if changed!
130 : : enum class Log {
131 : : Trace = 0,
132 : : Debug,
133 : : Info,
134 : : Warning,
135 : : Error,
136 : : Raise,
137 : : };
138 : :
139 : : kj::StringPtr KJ_STRINGIFY(Log flags);
140 : :
141 [ + - ]: 20 : struct LogMessage {
142 : :
143 : : //! Message to be logged
144 : : std::string message;
145 : :
146 : : //! The severity level of this message
147 : : Log level;
148 : : };
149 : :
150 : : using LogFn = std::function<void(LogMessage)>;
151 : :
152 [ + - ]: 3 : struct LogOptions {
153 : :
154 : : //! External logging callback.
155 : : LogFn log_fn;
156 : :
157 : : //! Maximum number of characters to use when representing
158 : : //! request and response structs as strings.
159 : : size_t max_chars{200};
160 : :
161 : : //! Messages with a severity level less than log_level will not be
162 : : //! reported.
163 : : Log log_level{Log::Trace};
164 : : };
165 : :
166 : : class Logger
167 : : {
168 : : public:
169 [ # # # # : 6 : Logger(const LogOptions& options, Log log_level) : m_options(options), m_log_level(log_level) {}
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # ]
[ # # # #
# # # # #
# # # # #
# # # # #
# # # #
# ][ - - +
- + - ]
170 : :
171 : : Logger(Logger&&) = delete;
172 : : Logger& operator=(Logger&&) = delete;
173 : : Logger(const Logger&) = delete;
174 : : Logger& operator=(const Logger&) = delete;
175 : :
176 : 10 : ~Logger() noexcept(false)
177 : : {
178 [ + - ]: 20 : if (enabled()) m_options.log_fn({std::move(m_buffer).str(), m_log_level});
179 [ + - ]: 20 : }
180 : :
181 : : template <typename T>
182 : 60 : friend Logger& operator<<(Logger& logger, T&& value)
183 : : {
184 [ - + ]: 60 : if (logger.enabled()) logger.m_buffer << std::forward<T>(value);
185 : 60 : return logger;
186 : : }
187 : :
188 : : template <typename T>
189 : : friend Logger& operator<<(Logger&& logger, T&& value)
190 : : {
191 : : return logger << std::forward<T>(value);
192 : : }
193 : :
194 : 10 : explicit operator bool() const
195 : : {
196 [ # # # # : 14 : return enabled();
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # #
# ][ # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # #
# ][ + - +
- # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # #
# ][ - - -
- + - + -
+ - + - -
- ]
197 : : }
198 : :
199 : : private:
200 : 80 : bool enabled() const
201 : : {
202 [ - - - - : 80 : return m_options.log_fn && m_log_level >= m_options.log_level;
- - - - -
- - - + -
+ - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - ][ #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # #
# ][ - - -
- + - + -
- - - - -
- - - - -
- - - - -
- - - - -
+ - + - +
- + - + -
+ - + - +
- + - + -
- - - - #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # ][ -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - + - +
- + - + -
- - - - -
- - - - -
- - - - -
- - - - -
- - - - -
- - - + -
+ - + - +
- - - - -
- - - - ]
[ - - - -
+ - + - +
- + - + -
+ - + - +
- + - + -
+ - + - +
- - - + -
+ - + - ]
203 : : }
204 : :
205 : : const LogOptions& m_options;
206 : : Log m_log_level;
207 : : std::ostringstream m_buffer;
208 : : };
209 : :
210 : : #define MP_LOGPLAIN(loop, ...) if (mp::Logger logger{(loop).m_log_opts, __VA_ARGS__}; logger) logger
211 : :
212 : : #define MP_LOG(loop, ...) MP_LOGPLAIN(loop, __VA_ARGS__) << "{" << LongThreadName((loop).m_exe_name) << "} "
213 : :
214 : : std::string LongThreadName(const char* exe_name);
215 : :
216 : : //! Event loop implementation.
217 : : //!
218 : : //! Cap'n Proto threading model is very simple: all I/O operations are
219 : : //! asynchronous and must be performed on a single thread. This includes:
220 : : //!
221 : : //! - Code starting an asynchronous operation (calling a function that returns a
222 : : //! promise object)
223 : : //! - Code notifying that an asynchronous operation is complete (code using a
224 : : //! fulfiller object)
225 : : //! - Code handling a completed operation (code chaining or waiting for a promise)
226 : : //!
227 : : //! All of this code needs to access shared state, and there is no mutex that
228 : : //! can be acquired to lock this state because Cap'n Proto
229 : : //! assumes it will only be accessed from one thread. So all this code needs to
230 : : //! actually run on one thread, and the EventLoop::loop() method is the entry point for
231 : : //! this thread. ProxyClient and ProxyServer objects that use other threads and
232 : : //! need to perform I/O operations post to this thread using EventLoop::post()
233 : : //! and EventLoop::sync() methods.
234 : : //!
235 : : //! Specifically, because ProxyClient methods can be called from arbitrary
236 : : //! threads, and ProxyServer methods can run on arbitrary threads, ProxyClient
237 : : //! methods use the EventLoop thread to send requests, and ProxyServer methods
238 : : //! use the thread to return results.
239 : : //!
240 : : //! Based on https://groups.google.com/d/msg/capnproto/TuQFF1eH2-M/g81sHaTAAQAJ
241 : : class EventLoop
242 : : {
243 : : public:
244 : : //! Construct event loop object with default logging options.
245 : 1 : EventLoop(const char* exe_name, LogFn log_fn, void* context = nullptr)
246 [ + - ]: 2 : : EventLoop(exe_name, LogOptions{std::move(log_fn)}, context){}
247 : :
248 : : //! Construct event loop object with specified logging options.
249 : : EventLoop(const char* exe_name, LogOptions log_opts, void* context = nullptr);
250 : :
251 : : //! Backwards-compatible constructor for previous (deprecated) logging callback signature
252 : : EventLoop(const char* exe_name, std::function<void(bool, std::string)> old_callback, void* context = nullptr)
253 : : : EventLoop(exe_name,
254 : : LogFn{[old_callback = std::move(old_callback)](LogMessage log_data) {old_callback(log_data.level == Log::Raise, std::move(log_data.message));}},
255 : : context){}
256 : :
257 : : ~EventLoop();
258 : :
259 : : //! Run event loop. Does not return until shutdown. This should only be
260 : : //! called once from the m_thread_id thread. This will block until
261 : : //! the m_num_refs reference count is 0.
262 : : void loop();
263 : :
264 : : //! Run function on event loop thread. Does not return until function completes.
265 : : //! Must be called while the loop() function is active.
266 : : void post(kj::Function<void()> fn);
267 : :
268 : : //! Wrapper around EventLoop::post that takes advantage of the
269 : : //! fact that callable will not go out of scope to avoid requirement that it
270 : : //! be copyable.
271 : : template <typename Callable>
272 : 1 : void sync(Callable&& callable)
273 : : {
274 [ + - ]: 1 : post(std::forward<Callable>(callable));
275 : 1 : }
276 : :
277 : : //! Register cleanup function to run on asynchronous worker thread without
278 : : //! blocking the event loop thread.
279 : : void addAsyncCleanup(std::function<void()> fn);
280 : :
281 : : //! Start asynchronous worker thread if necessary. This is only done if
282 : : //! there are ProxyServerBase::m_impl objects that need to be destroyed
283 : : //! asynchronously, without tying up the event loop thread. This can happen
284 : : //! when an interface does not declare a destroy() method that would allow
285 : : //! the client to wait for the destructor to finish and run it on a
286 : : //! dedicated thread. It can also happen whenever this is a broken
287 : : //! connection and the client is no longer around to call the destructors
288 : : //! and the server objects need to be garbage collected. In both cases, it
289 : : //! is important that ProxyServer::m_impl destructors do not run on the
290 : : //! eventloop thread because they may need it to do I/O if they perform
291 : : //! other IPC calls.
292 : : void startAsyncThread() MP_REQUIRES(m_mutex);
293 : :
294 : : //! Check if loop should exit.
295 : : bool done() const MP_REQUIRES(m_mutex);
296 : :
297 : : //! Process name included in thread names so combined debug output from
298 : : //! multiple processes is easier to understand.
299 : : const char* m_exe_name;
300 : :
301 : : //! ID of the event loop thread
302 : : std::thread::id m_thread_id = std::this_thread::get_id();
303 : :
304 : : //! Handle of an async worker thread. Joined on destruction. Unset if async
305 : : //! method has not been called.
306 : : std::thread m_async_thread;
307 : :
308 : : //! Callback function to run on event loop thread during post() or sync() call.
309 : : kj::Function<void()>* m_post_fn MP_GUARDED_BY(m_mutex) = nullptr;
310 : :
311 : : //! Callback functions to run on async thread.
312 : : std::optional<CleanupList> m_async_fns MP_GUARDED_BY(m_mutex);
313 : :
314 : : //! Pipe read handle used to wake up the event loop thread.
315 : : int m_wait_fd = -1;
316 : :
317 : : //! Pipe write handle used to wake up the event loop thread.
318 : : int m_post_fd = -1;
319 : :
320 : : //! Number of EventLoopRef instances referencing this event loop. This is a
321 : : //! sum of the number of client and server objects (Connection, ProxyClient,
322 : : //! ProxyServer) using the loop, plus temporary references held while
323 : : //! posting functions to the loop, plus any references held by external code
324 : : //! to keep the loop running. The loop() method exits when this count drops
325 : : //! to 0 (and m_async_fns is empty).
326 : : int m_num_refs MP_GUARDED_BY(m_mutex) = 0;
327 : :
328 : : //! Mutex and condition variable used to post tasks to event loop and async
329 : : //! thread.
330 : : Mutex m_mutex;
331 : : std::condition_variable m_cv;
332 : :
333 : : //! Capnp IO context.
334 : : kj::AsyncIoContext m_io_context;
335 : :
336 : : //! Capnp error handler. Needs to outlive m_task_set.
337 : : LoggingErrorHandler m_error_handler{*this};
338 : :
339 : : //! Capnp list of pending promises.
340 : : std::unique_ptr<kj::TaskSet> m_task_set;
341 : :
342 : : //! List of connections.
343 : : std::list<Connection> m_incoming_connections;
344 : :
345 : : //! Logging options
346 : : LogOptions m_log_opts;
347 : :
348 : : //! External context pointer.
349 : : void* m_context;
350 : :
351 : : //! Hook called when ProxyServer<ThreadMap>::makeThread() is called.
352 : : std::function<void()> testing_hook_makethread;
353 : :
354 : : //! Hook called on the worker thread inside makeThread(), after the thread
355 : : //! context is set up and thread_context promise is fulfilled, but before it
356 : : //! starts waiting for requests.
357 : : std::function<void()> testing_hook_makethread_created;
358 : :
359 : : //! Hook called on the worker thread when it starts to execute an async
360 : : //! request. Used by tests to control timing or inject behavior at this
361 : : //! point in execution.
362 : : std::function<void()> testing_hook_async_request_start;
363 : :
364 : : //! Hook called on the worker thread just before returning results.
365 : : std::function<void()> testing_hook_async_request_done;
366 : :
367 : : //! Hook called on the event loop thread when a client has connected.
368 : : std::function<void()> testing_hook_connected;
369 : :
370 : : //! Hook called on the event loop thread when a client has disconnected.
371 : : std::function<void()> testing_hook_disconnected;
372 : : };
373 : :
374 : : //! Single element task queue used to handle recursive capnp calls. (If the
375 : : //! server makes a callback into the client in the middle of a request, while the client
376 : : //! thread is blocked waiting for server response, this is what allows the
377 : : //! client to run the request in the same thread, the same way code would run in a
378 : : //! single process, with the callback sharing the same thread stack as the original
379 : : //! call.) To support this, the clientInvoke function calls Waiter::wait() to
380 : : //! block the client IPC thread while initial request is in progress. Then if
381 : : //! there is a callback, it is executed with Waiter::post().
382 : : //!
383 : : //! The Waiter class is also used server-side by `ProxyServer<Thread>::post()`
384 : : //! to execute IPC calls on worker threads.
385 : 0 : struct Waiter
386 : : {
387 : 0 : Waiter() = default;
388 : :
389 : : template <typename Fn>
390 : : bool post(Fn&& fn)
391 : : {
392 : : const Lock lock(m_mutex);
393 : : if (m_fn) return false;
394 : : m_fn = std::forward<Fn>(fn);
395 : : m_cv.notify_all();
396 : : return true;
397 : : }
398 : :
399 : : template <class Predicate>
400 : 0 : void wait(Lock& lock, Predicate pred) MP_REQUIRES(m_mutex)
401 : : {
402 : 0 : m_cv.wait(lock.m_lock, [&]() MP_REQUIRES(m_mutex) {
403 : : // Important for this to be "while (m_fn)", not "if (m_fn)" to avoid
404 : : // a lost-wakeup bug. A new m_fn and m_cv notification might be sent
405 : : // after the fn() call and before the lock.lock() call in this loop
406 : : // in the case where a capnp response is sent and a brand new
407 : : // request is immediately received.
408 [ # # # # : 0 : while (m_fn) {
# # # # #
# # # ]
[ # # # # ]
409 : 0 : auto fn = std::move(*m_fn);
410 : 0 : m_fn.reset();
411 [ # # # # : 0 : Unlock(lock, fn);
# # # # #
# # # ]
[ # # # # ]
412 : : }
413 : 0 : const bool done = pred();
414 : 0 : return done;
415 : : });
416 : 0 : }
417 : :
418 : : //! Mutex mainly used internally by waiter class, but also used externally
419 : : //! to guard access to related state. Specifically, since the thread_local
420 : : //! ThreadContext struct owns a Waiter, the Waiter::m_mutex is used to guard
421 : : //! access to other parts of the struct to avoid needing to deal with more
422 : : //! mutexes than necessary. This mutex can be held at the same time as
423 : : //! EventLoop::m_mutex as long as Waiter::mutex is locked first and
424 : : //! EventLoop::m_mutex is locked second.
425 : : Mutex m_mutex;
426 : : std::condition_variable m_cv MP_GUARDED_BY(m_mutex);
427 : : std::optional<kj::Function<void()>> m_fn MP_GUARDED_BY(m_mutex);
428 : : };
429 : :
430 : : //! Object holding network & rpc state associated with either an incoming server
431 : : //! connection, or an outgoing client connection. It must be created and destroyed
432 : : //! on the event loop thread.
433 : : //! In addition to Cap'n Proto state, it also holds lists of callbacks to run
434 : : //! when the connection is closed.
435 : : class Connection
436 : : {
437 : : public:
438 : 1 : Connection(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream_)
439 : 1 : : m_loop(loop), m_stream(kj::mv(stream_)),
440 [ + - + - ]: 1 : m_network(*m_stream, ::capnp::rpc::twoparty::Side::CLIENT, ::capnp::ReaderOptions()),
441 [ + - + - : 3 : m_rpc_system(::capnp::makeRpcClient(m_network)) {}
+ - + - ]
442 : 1 : Connection(EventLoop& loop,
443 : : kj::Own<kj::AsyncIoStream>&& stream_,
444 : : const std::function<::capnp::Capability::Client(Connection&)>& make_client)
445 : 1 : : m_loop(loop), m_stream(kj::mv(stream_)),
446 [ + - + - ]: 1 : m_network(*m_stream, ::capnp::rpc::twoparty::Side::SERVER, ::capnp::ReaderOptions()),
447 [ + - + - : 3 : m_rpc_system(::capnp::makeRpcServer(m_network, make_client(*this))) {}
+ - + - +
- + - ]
448 : :
449 : : //! Run cleanup functions. Must be called from the event loop thread. First
450 : : //! calls synchronous cleanup functions while blocked (to free capnp
451 : : //! Capability::Client handles owned by ProxyClient objects), then schedules
452 : : //! asynchronous cleanup functions to run in a worker thread (to run
453 : : //! destructors of m_impl instances owned by ProxyServer objects).
454 : : ~Connection();
455 : :
456 : : //! Register synchronous cleanup function to run on event loop thread (with
457 : : //! access to capnp thread local variables) when disconnect() is called.
458 : : //! any new i/o.
459 : : CleanupIt addSyncCleanup(std::function<void()> fn);
460 : : void removeSyncCleanup(CleanupIt it);
461 : :
462 : : //! Add disconnect handler.
463 : : template <typename F>
464 : 1 : void onDisconnect(F&& f)
465 : : {
466 : : // Add disconnect handler to local TaskSet to ensure it is canceled and
467 : : // will never run after connection object is destroyed. But when disconnect
468 : : // handler fires, do not call the function f right away, instead add it
469 : : // to the EventLoop TaskSet to avoid "Promise callback destroyed itself"
470 : : // error in the typical case where f deletes this Connection object.
471 [ + - - + ]: 3 : m_on_disconnect.add(m_network.onDisconnect().then(
472 [ + - + - ]: 3 : [f = std::forward<F>(f), this]() mutable { m_loop->m_task_set->add(kj::evalLater(kj::mv(f))); }));
473 : 1 : }
474 : :
475 : : EventLoopRef m_loop;
476 : : kj::Own<kj::AsyncIoStream> m_stream;
477 : : LoggingErrorHandler m_error_handler{*m_loop};
478 : : //! TaskSet used to cancel the m_network.onDisconnect() handler for remote
479 : : //! disconnections, if the connection is closed locally first by deleting
480 : : //! this Connection object.
481 : : kj::TaskSet m_on_disconnect{m_error_handler};
482 : : ::capnp::TwoPartyVatNetwork m_network;
483 : : std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId>> m_rpc_system;
484 : :
485 : : // ThreadMap interface client, used to create a remote server thread when an
486 : : // client IPC call is being made for the first time from a new thread.
487 : : ThreadMap::Client m_thread_map{nullptr};
488 : :
489 : : //! Collection of server-side IPC worker threads (ProxyServer<Thread> objects previously returned by
490 : : //! ThreadMap.makeThread) used to service requests to clients.
491 : : ::capnp::CapabilityServerSet<Thread> m_threads;
492 : :
493 : : //! A thread created by makePool with associated pending work queue. Vector is filled once by makePool() and never resized.
494 [ # # # # ]: 0 : struct PoolSlot {
495 : : Thread::Client client;
496 : : size_t depth{0};
497 : : };
498 : : std::vector<PoolSlot> m_thread_pool;
499 : :
500 : : //! Canceler for canceling promises that we want to discard when the
501 : : //! connection is destroyed. This is used to interrupt method calls that are
502 : : //! still executing at time of disconnection.
503 : : kj::Canceler m_canceler;
504 : :
505 : : //! Cleanup functions to run if connection is broken unexpectedly. List
506 : : //! will be empty if all ProxyClient are destroyed cleanly before the
507 : : //! connection is destroyed.
508 : : CleanupList m_sync_cleanup_fns;
509 : : };
510 : :
511 : : //! Vat id for server side of connection. Required argument to RpcSystem::bootStrap()
512 : : //!
513 : : //! "Vat" is Cap'n Proto nomenclature for a host of various objects that facilitates
514 : : //! bidirectional communication with other vats; it is often but not always 1-1 with
515 : : //! processes. Cap'n Proto doesn't reference clients or servers per se; instead everything
516 : : //! is just a vat.
517 : : //!
518 : : //! See also: https://github.com/capnproto/capnproto/blob/9021f0c722b36cb11e3690b0860939255ebad39c/c%2B%2B/src/capnp/rpc.capnp#L42-L56
519 [ + - + - ]: 1 : struct ServerVatId
520 : : {
521 : : ::capnp::word scratch[4]{};
522 : : ::capnp::MallocMessageBuilder message{scratch};
523 : : ::capnp::rpc::twoparty::VatId::Builder vat_id{message.getRoot<::capnp::rpc::twoparty::VatId>()};
524 [ + - ]: 1 : ServerVatId() { vat_id.setSide(::capnp::rpc::twoparty::Side::SERVER); }
525 : : };
526 : :
527 : : template <typename Interface, typename Impl>
528 : 1 : ProxyClientBase<Interface, Impl>::ProxyClientBase(typename Interface::Client client,
529 : : Connection* connection,
530 : : bool destroy_connection)
531 [ + - ]: 1 : : m_client(std::move(client)), m_context(connection)
532 : :
533 : : {
534 [ + - + - : 3 : MP_LOG(*m_context.loop, Log::Debug) << "Creating " << CxxTypeName(*this) << " " << this;
+ - + - +
- + - + -
+ - + - +
- + - +
- ]
535 : : // Handler for the connection getting destroyed before this client object.
536 [ + - ]: 1 : auto disconnect_cb = m_context.connection->addSyncCleanup([this]() {
537 : : // Release client capability by move-assigning to temporary.
538 : : {
539 : 0 : typename Interface::Client(std::move(m_client));
540 : : }
541 : 0 : Lock lock{m_context.loop->m_mutex};
542 : 0 : m_context.connection = nullptr;
543 : 0 : });
544 : :
545 : : // Two shutdown sequences are supported:
546 : : //
547 : : // - A normal sequence where client proxy objects are deleted by external
548 : : // code that no longer needs them
549 : : //
550 : : // - A garbage collection sequence where the connection or event loop shuts
551 : : // down while external code is still holding client references.
552 : : //
553 : : // The first case is handled here when m_context.connection is not null. The
554 : : // second case is handled by the disconnect_cb function, which sets
555 : : // m_context.connection to null so nothing happens here.
556 [ + - ]: 2 : m_context.cleanup_fns.emplace_front([this, destroy_connection, disconnect_cb]{
557 : : {
558 : : // If the capnp interface defines a destroy method, call it to destroy
559 : : // the remote object, waiting for it to be deleted server side. If the
560 : : // capnp interface does not define a destroy method, this will just call
561 : : // an empty stub defined in the ProxyClientBase class and do nothing.
562 : : // Exceptions are caught and logged rather than propagated because
563 : : // ~ProxyClientBase is noexcept and the peer may be gone by the time
564 : : // this runs.
565 [ - + - + ]: 1 : if (kj::runCatchingExceptions([&]{ Sub::destroy(*this); }) != nullptr) {
566 [ # # # # : 0 : MP_LOG(*m_context.loop, Log::Warning) << "Remote destroy call failed during cleanup. Continuing.";
# # # # #
# # # ]
567 : : }
568 : :
569 : : // FIXME: Could just invoke removed addCleanup fn here instead of duplicating code
570 : 2 : m_context.loop->sync([&]() {
571 : : // Remove disconnect callback on cleanup so it doesn't run and try
572 : : // to access this object after it's destroyed. This call needs to
573 : : // run inside loop->sync() on the event loop thread because
574 : : // otherwise, if there were an ill-timed disconnect, the
575 : : // onDisconnect handler could fire and delete the Connection object
576 : : // before the removeSyncCleanup call.
577 [ + - ]: 1 : if (m_context.connection) m_context.connection->removeSyncCleanup(disconnect_cb);
578 : :
579 : : // Release client capability by move-assigning to temporary.
580 : : {
581 : 1 : typename Interface::Client(std::move(m_client));
582 : : }
583 [ + - ]: 1 : if (destroy_connection) {
584 [ + - + - ]: 1 : delete m_context.connection;
585 : 1 : m_context.connection = nullptr;
586 : : }
587 : : });
588 : : }
589 : : });
590 : 1 : Sub::construct(*this);
591 : 1 : }
592 : :
593 : : template <typename Interface, typename Impl>
594 : 1 : ProxyClientBase<Interface, Impl>::~ProxyClientBase() noexcept
595 : : {
596 [ + - ]: 2 : MP_LOG(*m_context.loop, Log::Debug) << "Cleaning up " << CxxTypeName(*this) << " " << this;
597 : 1 : CleanupRun(m_context.cleanup_fns);
598 [ + - ]: 2 : MP_LOG(*m_context.loop, Log::Debug) << "Destroying " << CxxTypeName(*this) << " " << this;
599 : 1 : }
600 : :
601 : : template <typename Interface, typename Impl>
602 : 1 : ProxyServerBase<Interface, Impl>::ProxyServerBase(std::shared_ptr<Impl> impl, Connection& connection)
603 [ + - ]: 1 : : m_impl(std::move(impl)), m_context(&connection)
604 : : {
605 [ + - + - : 3 : MP_LOG(*m_context.loop, Log::Debug) << "Creating " << CxxTypeName(*this) << " " << this;
+ - + - +
- + - + -
+ - + - +
- + - +
- ]
606 [ - + ]: 1 : assert(m_impl);
607 [ - - ]: 1 : }
608 : :
609 : : //! ProxyServer destructor, called from the EventLoop thread by Cap'n Proto
610 : : //! garbage collection code after there are no more references to this object.
611 : : //! This will typically happen when the corresponding ProxyClient object on the
612 : : //! other side of the connection is destroyed. It can also happen earlier if the
613 : : //! connection is broken or destroyed. In the latter case this destructor will
614 : : //! typically be called inside m_rpc_system.reset() call in the ~Connection
615 : : //! destructor while the Connection object still exists. However, because
616 : : //! ProxyServer objects are refcounted, and the Connection object could be
617 : : //! destroyed while asynchronous IPC calls are still in-flight, it's possible
618 : : //! for this destructor to be called after the Connection object no longer
619 : : //! exists, so it is NOT valid to dereference the m_context.connection pointer
620 : : //! from this function.
621 : : template <typename Interface, typename Impl>
622 : 1 : ProxyServerBase<Interface, Impl>::~ProxyServerBase()
623 : : {
624 [ + - ]: 2 : MP_LOG(*m_context.loop, Log::Debug) << "Cleaning up " << CxxTypeName(*this) << " " << this;
625 [ + - ]: 1 : if (m_impl) {
626 : : // If impl is non-null at this point, it means no client is waiting for
627 : : // the m_impl server object to be destroyed synchronously. This can
628 : : // happen either if the interface did not define a "destroy" method (see
629 : : // invokeDestroy method below), or if a destroy method was defined, but
630 : : // the connection was broken before it could be called.
631 : : //
632 : : // In either case, be conservative and run the cleanup on an
633 : : // asynchronous thread, to avoid destructors or cleanup functions
634 : : // blocking or deadlocking the current EventLoop thread, since they
635 : : // could be making IPC calls.
636 : : //
637 : : // Technically this is a little too conservative since if the interface
638 : : // defines a "destroy" method, but the destroy method does not accept a
639 : : // Context parameter specifying a worker thread, the cleanup method
640 : : // would run on the EventLoop thread normally (when connection is
641 : : // unbroken), but will not run on the EventLoop thread now (when
642 : : // connection is broken). Probably some refactoring of the destructor
643 : : // and invokeDestroy function is possible to make this cleaner and more
644 : : // consistent.
645 : 1 : m_context.loop->addAsyncCleanup([impl=std::move(m_impl), fns=std::move(m_context.cleanup_fns)]() mutable {
646 : 1 : impl.reset();
647 : 1 : CleanupRun(fns);
648 : : });
649 : : }
650 [ - + ]: 1 : assert(m_context.cleanup_fns.empty());
651 [ + - ]: 2 : MP_LOG(*m_context.loop, Log::Debug) << "Destroying " << CxxTypeName(*this) << " " << this;
652 [ - + ]: 2 : }
653 : :
654 : : //! If the capnp interface defined a special "destroy" method, as described the
655 : : //! ProxyClientBase class, this method will be called and synchronously destroy
656 : : //! m_impl before returning to the client.
657 : : //!
658 : : //! If the capnp interface does not define a "destroy" method, this will never
659 : : //! be called and the ~ProxyServerBase destructor will be responsible for
660 : : //! deleting m_impl asynchronously, whenever the ProxyServer object gets garbage
661 : : //! collected by Cap'n Proto.
662 : : //!
663 : : //! This method is called in the same way other proxy server methods are called,
664 : : //! via the serverInvoke function. Basically serverInvoke just calls this as a
665 : : //! substitute for a non-existent m_impl->destroy() method. If the destroy
666 : : //! method has any parameters or return values they will be handled in the
667 : : //! normal way by PassField/ReadField/BuildField functions. Particularly if a
668 : : //! Context.thread parameter was passed, this method will run on the worker
669 : : //! thread specified by the client. Otherwise it will run on the EventLoop
670 : : //! thread, like other server methods without an assigned thread.
671 : : template <typename Interface, typename Impl>
672 : : void ProxyServerBase<Interface, Impl>::invokeDestroy()
673 : : {
674 : : m_impl.reset();
675 : : CleanupRun(m_context.cleanup_fns);
676 : : }
677 : :
678 : : //! Map from Connection to local or remote thread handle which will be used over
679 : : //! that connection. This map will typically only contain one entry, but can
680 : : //! contain multiple if a single thread makes IPC calls over multiple
681 : : //! connections. A std::optional value type is used to avoid the map needing to
682 : : //! be locked while ProxyClient<Thread> objects are constructed, see
683 : : //! ThreadContext "Synchronization note" below.
684 : : using ConnThreads = std::map<Connection*, std::optional<ProxyClient<Thread>>>;
685 : : using ConnThread = ConnThreads::iterator;
686 : :
687 : : // Retrieve ProxyClient<Thread> object associated with this connection from a
688 : : // map, or create a new one and insert it into the map. Return map iterator and
689 : : // inserted bool.
690 : : std::tuple<ConnThread, bool> SetThread(GuardedRef<ConnThreads> threads, Connection* connection, const std::function<Thread::Client()>& make_thread);
691 : :
692 : : //! The thread_local ThreadContext g_thread_context struct provides information
693 : : //! about individual threads and a way of communicating between them. Because
694 : : //! it's a thread local struct, each ThreadContext instance is initialized by
695 : : //! the thread that owns it.
696 : : //!
697 : : //! ThreadContext is used for any client threads created externally which make
698 : : //! IPC calls, and for server threads created by
699 : : //! ProxyServer<ThreadMap>::makeThread() which execute IPC calls for clients.
700 : : //!
701 : : //! In both cases, the struct holds information like the thread name, and a
702 : : //! Waiter object where the EventLoop can post incoming IPC requests to execute
703 : : //! on the thread. The struct also holds ConnThread maps associating the thread
704 : : //! with local and remote ProxyClient<Thread> objects.
705 : : struct ThreadContext
706 : : {
707 : : //! Identifying string for debug.
708 : : std::string thread_name;
709 : :
710 : : //! Waiter object used to allow remote clients to execute code on this
711 : : //! thread. For server threads created by
712 : : //! ProxyServer<ThreadMap>::makeThread(), this is initialized in that
713 : : //! function. Otherwise, for client threads created externally, this is
714 : : //! initialized the first time the thread tries to make an IPC call. Having
715 : : //! a waiter is necessary for threads making IPC calls in case a server they
716 : : //! are calling expects them to execute a callback during the call, before
717 : : //! it sends a response.
718 : : //!
719 : : //! For IPC client threads, the Waiter pointer is never cleared and the Waiter
720 : : //! just gets destroyed when the thread does. For server threads created by
721 : : //! makeThread(), this pointer is set to null in the ~ProxyServer<Thread> as
722 : : //! a signal for the thread to exit and destroy itself. In both cases, the
723 : : //! same Waiter object is used across different calls and only created and
724 : : //! destroyed once for the lifetime of the thread.
725 : : std::unique_ptr<Waiter> waiter = nullptr;
726 : :
727 : : //! When client is making a request to a server, this is the
728 : : //! `callbackThread` argument it passes in the request, used by the server
729 : : //! in case it needs to make callbacks into the client that need to execute
730 : : //! while the client is waiting. This will be set to a local thread object.
731 : : //!
732 : : //! Synchronization note: The callback_thread and request_thread maps are
733 : : //! only ever accessed internally by this thread's destructor and externally
734 : : //! by Cap'n Proto event loop threads. Since it's possible for IPC client
735 : : //! threads to make calls over different connections that could have
736 : : //! different event loops, these maps are guarded by Waiter::m_mutex in case
737 : : //! different event loop threads add or remove map entries simultaneously.
738 : : //! However, individual ProxyClient<Thread> objects in the maps will only be
739 : : //! associated with one event loop and guarded by EventLoop::m_mutex. So
740 : : //! Waiter::m_mutex does not need to be held while accessing individual
741 : : //! ProxyClient<Thread> instances, and may even need to be released to
742 : : //! respect lock order and avoid locking Waiter::m_mutex before
743 : : //! EventLoop::m_mutex.
744 : : ConnThreads callback_threads MP_GUARDED_BY(waiter->m_mutex);
745 : :
746 : : //! When client is making a request to a server, this is the `thread`
747 : : //! argument it passes in the request, used to control which thread on
748 : : //! server will be responsible for executing it. If client call is being
749 : : //! made from a local thread, this will be a remote thread object returned
750 : : //! by makeThread. If a client call is being made from a thread currently
751 : : //! handling a server request, this will be set to the `callbackThread`
752 : : //! request thread argument passed in that request.
753 : : //!
754 : : //! Synchronization note: \ref callback_threads note applies here as well.
755 : : ConnThreads request_threads MP_GUARDED_BY(waiter->m_mutex);
756 : :
757 : : //! Whether this thread is a capnp event loop thread. Not really used except
758 : : //! to assert false if there's an attempt to execute a blocking operation
759 : : //! which could deadlock the thread.
760 : : bool loop_thread = false;
761 : : };
762 : :
763 : : template<typename T, typename Fn>
764 : : kj::Promise<T> ProxyServer<Thread>::post(Fn&& fn)
765 : : {
766 : : auto ready = kj::newPromiseAndFulfiller<void>(); // Signaled when waiter is ready to post again.
767 : : auto cancel_monitor_ptr = kj::heap<CancelMonitor>();
768 : : CancelMonitor& cancel_monitor = *cancel_monitor_ptr;
769 : : // Keep a reference to the ProxyServer<Thread> instance by assigning it to
770 : : // the self variable. ProxyServer instances are reference-counted and if the
771 : : // client drops its reference, this variable keeps the instance alive until
772 : : // the thread finishes executing. The self variable needs to be destroyed on
773 : : // the event loop thread so it is freed in a sync() call below.
774 : : auto self = thisCap();
775 : : 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 {
776 : : auto result = kj::newPromiseAndFulfiller<T>(); // Signaled when fn() is called, with its return value.
777 : : 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 {
778 : : // Fulfill ready.promise now, as soon as the Waiter starts executing
779 : : // this lambda, so the next ProxyServer<Thread>::post() call can
780 : : // immediately call waiter->post(). It is important to do this
781 : : // before calling fn() because fn() can make an IPC call back to the
782 : : // client, which can make another IPC call to this server thread.
783 : : // (This typically happens when IPC methods take std::function
784 : : // parameters.) When this happens the second call to the server
785 : : // thread should not be blocked waiting for the first call.
786 : : m_loop->sync([ready_fulfiller = kj::mv(ready_fulfiller)]() mutable {
787 : : ready_fulfiller->fulfill();
788 : : ready_fulfiller = nullptr;
789 : : });
790 : : std::optional<T> result_value;
791 : : kj::Maybe<kj::Exception> exception{kj::runCatchingExceptions([&]{ result_value.emplace(fn(*cancel_monitor_ptr)); })};
792 : : 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 {
793 : : // Destroy CancelMonitor here before fulfilling or rejecting the
794 : : // promise so it doesn't get triggered when the promise is
795 : : // destroyed.
796 : : cancel_monitor_ptr = nullptr;
797 : : // Send results to the fulfiller. Technically it would be ok to
798 : : // skip this if promise was canceled, but it's simpler to just
799 : : // do it unconditionally.
800 : : KJ_IF_MAYBE(e, exception) {
801 : : assert(!result_value);
802 : : result_fulfiller->reject(kj::mv(*e));
803 : : } else {
804 : : assert(result_value);
805 : : result_fulfiller->fulfill(kj::mv(*result_value));
806 : : result_value.reset();
807 : : }
808 : : result_fulfiller = nullptr;
809 : : // Use evalLater to destroy the ProxyServer<Thread> self
810 : : // reference, if it is the last reference, because the
811 : : // ProxyServer<Thread> destructor needs to join the thread,
812 : : // which can't happen until this sync() block has exited.
813 : : m_loop->m_task_set->add(kj::evalLater([self = kj::mv(self)] {}));
814 : : });
815 : : });
816 : : // Assert that calling Waiter::post did not fail. It could only return
817 : : // false if a new function was posted before the previous one finished
818 : : // executing, but new functions are only posted when m_thread_ready is
819 : : // signaled, so this should never happen.
820 : : assert(posted);
821 : : return kj::mv(result.promise);
822 : : }).attach(kj::heap<CancelProbe>(cancel_monitor));
823 : : m_thread_ready = kj::mv(ready.promise);
824 : : return ret;
825 : : }
826 : :
827 : : //! Given stream file descriptor, make a new ProxyClient object to send requests
828 : : //! over the stream. Also create a new Connection object embedded in the
829 : : //! client that is freed when the client is closed.
830 : : template <typename InitInterface>
831 : : std::unique_ptr<ProxyClient<InitInterface>> ConnectStream(EventLoop& loop, int fd)
832 : : {
833 : : typename InitInterface::Client init_client(nullptr);
834 : : std::unique_ptr<Connection> connection;
835 : : loop.sync([&] {
836 : : auto stream =
837 : : loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP);
838 : : connection = std::make_unique<Connection>(loop, kj::mv(stream));
839 : : init_client = connection->m_rpc_system->bootstrap(ServerVatId().vat_id).castAs<InitInterface>();
840 : : Connection* connection_ptr = connection.get();
841 : : connection->onDisconnect([&loop, connection_ptr] {
842 : : MP_LOG(loop, Log::Warning) << "IPC client: unexpected network disconnect.";
843 : : delete connection_ptr;
844 : : });
845 : : });
846 : : return std::make_unique<ProxyClient<InitInterface>>(
847 : : kj::mv(init_client), connection.release(), /* destroy_connection= */ true);
848 : : }
849 : :
850 : : //! Given stream and init objects, construct a new ProxyServer object that
851 : : //! handles requests from the stream by calling the init object. Embed the
852 : : //! ProxyServer in a Connection object that is stored and erased if
853 : : //! disconnected. This should be called from the event loop thread.
854 : : template <typename InitInterface, typename InitImpl, typename OnDisconnect>
855 : : void _Serve(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream, InitImpl& init, OnDisconnect&& on_disconnect)
856 : : {
857 : : loop.m_incoming_connections.emplace_front(loop, kj::mv(stream), [&](Connection& connection) {
858 : : // Disable deleter so proxy server object doesn't attempt to delete the
859 : : // init implementation when the proxy client is destroyed or
860 : : // disconnected.
861 : : return kj::heap<ProxyServer<InitInterface>>(std::shared_ptr<InitImpl>(&init, [](InitImpl*){}), connection);
862 : : });
863 : : auto it = loop.m_incoming_connections.begin();
864 : : MP_LOG(loop, Log::Info) << "IPC server: socket connected.";
865 : : if (loop.testing_hook_connected) loop.testing_hook_connected();
866 : : it->onDisconnect([&loop, it, on_disconnect = std::forward<OnDisconnect>(on_disconnect)]() mutable {
867 : : MP_LOG(loop, Log::Info) << "IPC server: socket disconnected.";
868 : : loop.m_incoming_connections.erase(it);
869 : : on_disconnect();
870 : : if (loop.testing_hook_disconnected) loop.testing_hook_disconnected();
871 : : });
872 : : }
873 : :
874 : : struct Listener
875 : : {
876 : : explicit Listener(kj::Own<kj::ConnectionReceiver>&& receiver, std::optional<size_t> max_connections)
877 : : : m_receiver(kj::mv(receiver)), m_max_connections(max_connections) {}
878 : :
879 : : bool atCapacity() const
880 : : {
881 : : return m_max_connections && m_active_connections >= *m_max_connections;
882 : : }
883 : :
884 : : kj::Own<kj::ConnectionReceiver> m_receiver;
885 : : std::optional<size_t> m_max_connections;
886 : : size_t m_active_connections{0};
887 : : };
888 : :
889 : : template <typename InitInterface, typename InitImpl>
890 : : void _Listen(const std::shared_ptr<Listener>& listener, EventLoop& loop, InitImpl& init)
891 : : {
892 : : if (listener->atCapacity()) return;
893 : :
894 : : auto* receiver = listener->m_receiver.get();
895 : : loop.m_task_set->add(receiver->accept().then(
896 : : [&loop, &init, listener](kj::Own<kj::AsyncIoStream>&& stream) {
897 : : ++listener->m_active_connections;
898 : : _Serve<InitInterface>(loop, kj::mv(stream), init, [&loop, &init, listener] {
899 : : const bool resume_accept{listener->atCapacity()};
900 : : assert(listener->m_active_connections > 0);
901 : : --listener->m_active_connections;
902 : : if (resume_accept) _Listen<InitInterface>(listener, loop, init);
903 : : });
904 : : _Listen<InitInterface>(listener, loop, init);
905 : : }));
906 : : }
907 : :
908 : : //! Given stream file descriptor and an init object, handle requests on the
909 : : //! stream by calling methods on the Init object.
910 : : template <typename InitInterface, typename InitImpl>
911 : : void ServeStream(EventLoop& loop, int fd, InitImpl& init)
912 : : {
913 : : _Serve<InitInterface>(
914 : : loop,
915 : : loop.m_io_context.lowLevelProvider->wrapSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
916 : : init,
917 : : [] {});
918 : : }
919 : :
920 : : //! Given listening socket file descriptor and an init object, handle incoming
921 : : //! connections and requests by calling methods on the Init object.
922 : : template <typename InitInterface, typename InitImpl>
923 : : void ListenConnections(EventLoop& loop, int fd, InitImpl& init, std::optional<size_t> max_connections = std::nullopt)
924 : : {
925 : : loop.sync([&]() {
926 : : auto listener{std::make_shared<Listener>(
927 : : loop.m_io_context.lowLevelProvider->wrapListenSocketFd(fd, kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP),
928 : : max_connections)};
929 : : _Listen<InitInterface>(listener, loop, init);
930 : : });
931 : : }
932 : :
933 [ # # # # : 1 : extern thread_local ThreadContext g_thread_context; // NOLINT(bitcoin-nontrivial-threadlocal)
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # # # #
# # ]
934 : : // Silence nonstandard bitcoin tidy error "Variable with non-trivial destructor
935 : : // cannot be thread_local" which should not be a problem on modern platforms, and
936 : : // could lead to a small memory leak at worst on older ones.
937 : :
938 : : } // namespace mp
939 : :
940 : : #endif // MP_PROXY_IO_H
|
