Bitcoin Core 31.99.0
P2P Digital Currency
proxy-io.h
Go to the documentation of this file.
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
28namespace mp {
29struct ThreadContext;
30struct Listener;
31
33{
35};
36
38{
42 {
43 }
44};
45
46template <typename ProxyServer, typename CallContext_>
48{
49 using CallContext = CallContext_;
50
53 int req;
59 Lock* cancel_lock{nullptr};
66 bool request_canceled{false};
67
70 {
71 }
72};
73
74template <typename Interface, typename Params, typename Results>
75using ServerContext = ServerInvokeContext<ProxyServer<Interface>, ::capnp::CallContext<Params, Results>>;
76
77template <>
78struct ProxyClient<Thread> : public ProxyClientBase<Thread, ::capnp::Void>
79{
81 // https://stackoverflow.com/questions/22357887/comparing-two-mapiterators-why-does-it-need-the-copy-constructor-of-stdpair
82 ProxyClient(const ProxyClient&) = delete;
84
94 std::optional<CleanupIt> m_disconnect_cb;
95};
96
97template <>
98struct ProxyServer<Thread> final : public Thread::Server
99{
100public:
101 ProxyServer(Connection& connection, ThreadContext& thread_context, std::thread&& thread);
102 ~ProxyServer();
103 kj::Promise<void> getName(GetNameContext context) override;
104
109 template<typename T, typename Fn>
110 kj::Promise<T> post(Fn&& fn);
111
117 kj::Promise<void> m_thread_ready{kj::READY_NOW};
118};
119
121class LoggingErrorHandler : public kj::TaskSet::ErrorHandler
122{
123public:
125 void taskFailed(kj::Exception&& exception) override;
127};
128
130enum class Log {
131 Trace = 0,
132 Debug,
133 Info,
134 Warning,
135 Error,
136 Raise,
137};
138
139kj::StringPtr KJ_STRINGIFY(Log flags);
140
142
144 std::string message;
145
148};
149
150using LogFn = std::function<void(LogMessage)>;
151
156
159 size_t max_chars{200};
160
164};
165
167{
168public:
169 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 ~Logger() noexcept(false)
177 {
178 if (enabled()) m_options.log_fn({std::move(m_buffer).str(), m_log_level});
179 }
180
181 template <typename T>
182 friend Logger& operator<<(Logger& logger, T&& value)
183 {
184 if (logger.enabled()) logger.m_buffer << std::forward<T>(value);
185 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 explicit operator bool() const
195 {
196 return enabled();
197 }
198
199private:
200 bool enabled() const
201 {
203 }
204
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
214std::string LongThreadName(const char* exe_name);
215
242{
243public:
245 EventLoop(const char* exe_name, LogFn log_fn, void* context = nullptr)
246 : EventLoop(exe_name, LogOptions{std::move(log_fn)}, context){}
247
249 EventLoop(const char* exe_name, LogOptions log_opts, void* context = nullptr);
250
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
262 void loop();
263
266 void post(kj::Function<void()> fn);
267
271 template <typename Callable>
272 void sync(Callable&& callable)
273 {
274 post(std::forward<Callable>(callable));
275 }
276
279 void addAsyncCleanup(std::function<void()> fn);
280
292 void startAsyncThread() MP_REQUIRES(m_mutex);
293
295 bool done() const MP_REQUIRES(m_mutex);
296
299 const char* m_exe_name;
300
302 std::thread::id m_thread_id = std::this_thread::get_id();
303
306 std::thread m_async_thread;
307
309 kj::Function<void()>* m_post_fn MP_GUARDED_BY(m_mutex) = nullptr;
310
312 std::optional<CleanupList> m_async_fns MP_GUARDED_BY(m_mutex);
313
315 int m_wait_fd = -1;
316
318 int m_post_fd = -1;
319
326 int m_num_refs MP_GUARDED_BY(m_mutex) = 0;
327
330 Mutex m_mutex;
331 std::condition_variable m_cv;
332
334 kj::AsyncIoContext m_io_context;
335
337 LoggingErrorHandler m_error_handler{*this};
338
340 std::unique_ptr<kj::TaskSet> m_task_set;
341
343 std::list<Connection> m_incoming_connections;
344
347
350
352 std::function<void()> testing_hook_makethread;
353
357 std::function<void()> testing_hook_makethread_created;
358
362 std::function<void()> testing_hook_async_request_start;
363
365 std::function<void()> testing_hook_async_request_done;
366
368 std::function<void()> testing_hook_connected;
369
371 std::function<void()> testing_hook_disconnected;
372};
373
385struct Waiter
386{
387 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 void wait(Lock& lock, Predicate pred) MP_REQUIRES(m_mutex)
401 {
402 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 while (m_fn) {
409 auto fn = std::move(*m_fn);
410 m_fn.reset();
411 Unlock(lock, fn);
412 }
413 const bool done = pred();
414 return done;
415 });
416 }
417
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
436{
437public:
438 Connection(EventLoop& loop, kj::Own<kj::AsyncIoStream>&& stream_)
439 : m_loop(loop), m_stream(kj::mv(stream_)),
440 m_network(*m_stream, ::capnp::rpc::twoparty::Side::CLIENT, ::capnp::ReaderOptions()),
441 m_rpc_system(::capnp::makeRpcClient(m_network)) {}
443 kj::Own<kj::AsyncIoStream>&& stream_,
444 const std::function<::capnp::Capability::Client(Connection&)>& make_client)
445 : m_loop(loop), m_stream(kj::mv(stream_)),
446 m_network(*m_stream, ::capnp::rpc::twoparty::Side::SERVER, ::capnp::ReaderOptions()),
447 m_rpc_system(::capnp::makeRpcServer(m_network, make_client(*this))) {}
448
454 ~Connection();
455
459 CleanupIt addSyncCleanup(std::function<void()> fn);
460 void removeSyncCleanup(CleanupIt it);
461
463 template <typename F>
464 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 m_on_disconnect.add(m_network.onDisconnect().then(
472 [f = std::forward<F>(f), this]() mutable { m_loop->m_task_set->add(kj::evalLater(kj::mv(f))); }));
473 }
474
476 kj::Own<kj::AsyncIoStream> m_stream;
477 LoggingErrorHandler m_error_handler{*m_loop};
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
491 ::capnp::CapabilityServerSet<Thread> m_threads;
492
494 struct PoolSlot {
495 Thread::Client client;
496 size_t depth{0};
497 };
498 std::vector<PoolSlot> m_thread_pool;
499
503 kj::Canceler m_canceler;
504
509};
510
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 ServerVatId() { vat_id.setSide(::capnp::rpc::twoparty::Side::SERVER); }
525};
526
527template <typename Interface, typename Impl>
529 Connection* connection,
530 bool destroy_connection)
531 : m_client(std::move(client)), m_context(connection)
532
533{
534 MP_LOG(*m_context.loop, Log::Debug) << "Creating " << CxxTypeName(*this) << " " << this;
535 // Handler for the connection getting destroyed before this client object.
536 auto disconnect_cb = m_context.connection->addSyncCleanup([this]() {
537 // Release client capability by move-assigning to temporary.
538 {
539 typename Interface::Client(std::move(m_client));
540 }
541 Lock lock{m_context.loop->m_mutex};
542 m_context.connection = nullptr;
543 });
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 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 if (kj::runCatchingExceptions([&]{ Sub::destroy(*this); }) != nullptr) {
566 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 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.
578
579 // Release client capability by move-assigning to temporary.
580 {
581 typename Interface::Client(std::move(m_client));
582 }
583 if (destroy_connection) {
584 delete m_context.connection;
585 m_context.connection = nullptr;
586 }
587 });
588 }
589 });
590 Sub::construct(*this);
591}
592
593template <typename Interface, typename Impl>
595{
596 MP_LOG(*m_context.loop, Log::Debug) << "Cleaning up " << CxxTypeName(*this) << " " << this;
597 CleanupRun(m_context.cleanup_fns);
598 MP_LOG(*m_context.loop, Log::Debug) << "Destroying " << CxxTypeName(*this) << " " << this;
599}
600
601template <typename Interface, typename Impl>
602ProxyServerBase<Interface, Impl>::ProxyServerBase(std::shared_ptr<Impl> impl, Connection& connection)
603 : m_impl(std::move(impl)), m_context(&connection)
604{
605 MP_LOG(*m_context.loop, Log::Debug) << "Creating " << CxxTypeName(*this) << " " << this;
606 assert(m_impl);
607}
608
621template <typename Interface, typename Impl>
623{
624 MP_LOG(*m_context.loop, Log::Debug) << "Cleaning up " << CxxTypeName(*this) << " " << this;
625 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 m_context.loop->addAsyncCleanup([impl=std::move(m_impl), fns=std::move(m_context.cleanup_fns)]() mutable {
646 impl.reset();
647 CleanupRun(fns);
648 });
649 }
650 assert(m_context.cleanup_fns.empty());
651 MP_LOG(*m_context.loop, Log::Debug) << "Destroying " << CxxTypeName(*this) << " " << this;
652}
653
671template <typename Interface, typename Impl>
673{
674 m_impl.reset();
675 CleanupRun(m_context.cleanup_fns);
676}
677
684using ConnThreads = std::map<Connection*, std::optional<ProxyClient<Thread>>>;
685using 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.
690std::tuple<ConnThread, bool> SetThread(GuardedRef<ConnThreads> threads, Connection* connection, const std::function<Thread::Client()>& make_thread);
691
706{
708 std::string thread_name;
709
725 std::unique_ptr<Waiter> waiter = nullptr;
726
744 ConnThreads callback_threads MP_GUARDED_BY(waiter->m_mutex);
745
755 ConnThreads request_threads MP_GUARDED_BY(waiter->m_mutex);
756
760 bool loop_thread = false;
761};
762
763template<typename T, typename Fn>
764kj::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
830template <typename InitInterface>
831std::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
854template <typename InitInterface, typename InitImpl, typename OnDisconnect>
855void _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.";
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();
871 });
872}
873
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
889template <typename InitInterface, typename InitImpl>
890void _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
910template <typename InitInterface, typename InitImpl>
911void 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
922template <typename InitInterface, typename InitImpl>
923void 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
933extern 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
int ret
if(!SetupNetworking())
int flags
Definition: bitcoin-tx.cpp:530
Helper class that detects when a promise is canceled.
Definition: util.h:321
Object holding network & rpc state associated with either an incoming server connection,...
Definition: proxy-io.h:436
CleanupIt addSyncCleanup(std::function< void()> fn)
Register synchronous cleanup function to run on event loop thread (with access to capnp thread local ...
Definition: proxy.cpp:156
EventLoopRef m_loop
Definition: proxy-io.h:475
::capnp::TwoPartyVatNetwork m_network
Definition: proxy-io.h:482
kj::Own< kj::AsyncIoStream > m_stream
Definition: proxy-io.h:476
Connection(EventLoop &loop, kj::Own< kj::AsyncIoStream > &&stream_, const std::function<::capnp::Capability::Client(Connection &)> &make_client)
Definition: proxy-io.h:442
Connection(EventLoop &loop, kj::Own< kj::AsyncIoStream > &&stream_)
Definition: proxy-io.h:438
void onDisconnect(F &&f)
Add disconnect handler.
Definition: proxy-io.h:464
std::vector< PoolSlot > m_thread_pool
Definition: proxy-io.h:498
kj::Canceler m_canceler
Canceler for canceling promises that we want to discard when the connection is destroyed.
Definition: proxy-io.h:503
::capnp::CapabilityServerSet< Thread > m_threads
Collection of server-side IPC worker threads (ProxyServer<Thread> objects previously returned by Thre...
Definition: proxy-io.h:491
CleanupList m_sync_cleanup_fns
Cleanup functions to run if connection is broken unexpectedly.
Definition: proxy-io.h:508
std::optional<::capnp::RpcSystem<::capnp::rpc::twoparty::VatId > > m_rpc_system
Definition: proxy-io.h:483
void removeSyncCleanup(CleanupIt it)
Definition: proxy.cpp:170
Event loop implementation.
Definition: proxy-io.h:242
kj::AsyncIoContext m_io_context
Capnp IO context.
Definition: proxy-io.h:334
void sync(Callable &&callable)
Wrapper around EventLoop::post that takes advantage of the fact that callable will not go out of scop...
Definition: proxy-io.h:272
EventLoop(const char *exe_name, LogFn log_fn, void *context=nullptr)
Construct event loop object with default logging options.
Definition: proxy-io.h:245
std::function< void()> testing_hook_connected
Hook called on the event loop thread when a client has connected.
Definition: proxy-io.h:368
std::list< Connection > m_incoming_connections
List of connections.
Definition: proxy-io.h:343
Mutex m_mutex
Mutex and condition variable used to post tasks to event loop and async thread.
Definition: proxy-io.h:330
std::function< void()> testing_hook_disconnected
Hook called on the event loop thread when a client has disconnected.
Definition: proxy-io.h:371
LogOptions m_log_opts
Logging options.
Definition: proxy-io.h:346
std::unique_ptr< kj::TaskSet > m_task_set
Capnp list of pending promises.
Definition: proxy-io.h:340
void * m_context
External context pointer.
Definition: proxy-io.h:349
std::function< void()> testing_hook_makethread
Hook called when ProxyServer<ThreadMap>::makeThread() is called.
Definition: proxy-io.h:352
std::function< void()> testing_hook_makethread_created
Hook called on the worker thread inside makeThread(), after the thread context is set up and thread_c...
Definition: proxy-io.h:357
EventLoop(const char *exe_name, std::function< void(bool, std::string)> old_callback, void *context=nullptr)
Backwards-compatible constructor for previous (deprecated) logging callback signature.
Definition: proxy-io.h:252
std::function< void()> testing_hook_async_request_done
Hook called on the worker thread just before returning results.
Definition: proxy-io.h:365
std::function< void()> testing_hook_async_request_start
Hook called on the worker thread when it starts to execute an async request.
Definition: proxy-io.h:362
Event loop smart pointer automatically managing m_num_refs.
Definition: proxy.h:51
Definition: util.h:177
friend Logger & operator<<(Logger &logger, T &&value)
Definition: proxy-io.h:182
Logger(const Logger &)=delete
bool enabled() const
Definition: proxy-io.h:200
~Logger() noexcept(false)
Definition: proxy-io.h:176
Logger & operator=(Logger &&)=delete
Log m_log_level
Definition: proxy-io.h:206
Logger(Logger &&)=delete
Logger(const LogOptions &options, Log log_level)
Definition: proxy-io.h:169
const LogOptions & m_options
Definition: proxy-io.h:205
friend Logger & operator<<(Logger &&logger, T &&value)
Definition: proxy-io.h:189
std::ostringstream m_buffer
Definition: proxy-io.h:207
Logger & operator=(const Logger &)=delete
Handler for kj::TaskSet failed task events.
Definition: proxy-io.h:122
EventLoop & m_loop
Definition: proxy-io.h:126
LoggingErrorHandler(EventLoop &loop)
Definition: proxy-io.h:124
void taskFailed(kj::Exception &&exception) override
Definition: proxy.cpp:46
Base class for generated ProxyClient classes that implement a C++ interface and forward calls to a ca...
Definition: proxy.h:81
Interface::Client m_client
Definition: proxy.h:132
ProxyContext m_context
Definition: proxy.h:133
~ProxyClientBase() noexcept
Definition: proxy-io.h:594
ProxyClientBase(typename Interface::Client client, Connection *connection, bool destroy_connection)
Construct libmultiprocess client object wrapping Cap'n Proto client object with a reference to the as...
Definition: proxy-io.h:528
std::optional< mp::EventLoop > m_loop
EventLoop object which manages I/O events for all connections.
Definition: protocol.cpp:144
Context m_context
Definition: protocol.cpp:141
#define MP_GUARDED_BY(x)
Definition: util.h:166
#define MP_REQUIRES(x)
Definition: util.h:162
std::unique_ptr< ProxyClient< messages::FooInterface > > client
UnixListener listener
std::thread thread
Thread variable should be after other struct members so the thread does not start until the other mem...
Definition: basic.cpp:8
Functions to serialize / deserialize common bitcoin types.
Definition: common-types.h:57
void Unlock(Lock &lock, Callback &&callback)
Definition: util.h:213
void ListenConnections(EventLoop &loop, int fd, InitImpl &init, std::optional< size_t > max_connections=std::nullopt)
Given listening socket file descriptor and an init object, handle incoming connections and requests b...
Definition: proxy-io.h:923
std::string CxxTypeName(const T &)
Definition: util.h:296
kj::StringPtr KJ_STRINGIFY(Log flags)
Definition: proxy.cpp:472
std::list< std::function< void()> > CleanupList
Definition: proxy.h:36
void ServeStream(EventLoop &loop, int fd, InitImpl &init)
Given stream file descriptor and an init object, handle requests on the stream by calling methods on ...
Definition: proxy-io.h:911
void _Serve(EventLoop &loop, kj::Own< kj::AsyncIoStream > &&stream, InitImpl &init, OnDisconnect &&on_disconnect)
Given stream and init objects, construct a new ProxyServer object that handles requests from the stre...
Definition: proxy-io.h:855
std::function< void(LogMessage)> LogFn
Definition: proxy-io.h:150
std::tuple< ConnThread, bool > SetThread(GuardedRef< ConnThreads > threads, Connection *connection, const std::function< Thread::Client()> &make_thread)
Definition: proxy.cpp:329
thread_local ThreadContext g_thread_context
Definition: proxy.cpp:44
std::unique_ptr< ProxyClient< InitInterface > > ConnectStream(EventLoop &loop, int fd)
Given stream file descriptor, make a new ProxyClient object to send requests over the stream.
Definition: proxy-io.h:831
Log
Log flags. Update stringify function if changed!
Definition: proxy-io.h:130
std::string LongThreadName(const char *exe_name)
Definition: proxy.cpp:467
ConnThreads::iterator ConnThread
Definition: proxy-io.h:685
typename CleanupList::iterator CleanupIt
Definition: proxy.h:37
void CleanupRun(CleanupList &fns)
Definition: proxy.h:39
void _Listen(const std::shared_ptr< Listener > &listener, EventLoop &loop, InitImpl &init)
Definition: proxy-io.h:890
std::map< Connection *, std::optional< ProxyClient< Thread > > > ConnThreads
Map from Connection to local or remote thread handle which will be used over that connection.
Definition: proxy-io.h:684
#define MP_LOG(loop,...)
Definition: proxy-io.h:212
ThreadContext & thread_context
Definition: proxy-io.h:39
ClientInvokeContext(Connection &conn, ThreadContext &thread_context)
Definition: proxy-io.h:40
A thread created by makePool with associated pending work queue. Vector is filled once by makePool() ...
Definition: proxy-io.h:494
Thread::Client client
Definition: proxy-io.h:495
Connection & connection
Definition: proxy-io.h:34
std::optional< size_t > m_max_connections
Definition: proxy-io.h:885
kj::Own< kj::ConnectionReceiver > m_receiver
Definition: proxy-io.h:884
Listener(kj::Own< kj::ConnectionReceiver > &&receiver, std::optional< size_t > max_connections)
Definition: proxy-io.h:876
bool atCapacity() const
Definition: proxy-io.h:879
Log level
The severity level of this message.
Definition: proxy-io.h:147
std::string message
Message to be logged.
Definition: proxy-io.h:144
LogFn log_fn
External logging callback.
Definition: proxy-io.h:155
size_t max_chars
Maximum number of characters to use when representing request and response structs as strings.
Definition: proxy-io.h:159
Log log_level
Messages with a severity level less than log_level will not be reported.
Definition: proxy-io.h:163
ProxyClient(const ProxyClient &)=delete
std::optional< CleanupIt > m_disconnect_cb
Reference to callback function that is run if there is a sudden disconnect and the Connection object ...
Definition: proxy-io.h:94
Mapping from capnp interface type to proxy client implementation (specializations are generated by pr...
Definition: proxy.h:25
EventLoopRef loop
Definition: proxy.h:71
Connection * connection
Definition: proxy.h:70
CleanupList cleanup_fns
Definition: proxy.h:72
ThreadContext & m_thread_context
Definition: proxy-io.h:113
Base class for generated ProxyServer classes that implement capnp server methods and forward calls to...
Definition: proxy.h:148
ProxyContext m_context
Definition: proxy.h:172
virtual ~ProxyServerBase()
ProxyServer destructor, called from the EventLoop thread by Cap'n Proto garbage collection code after...
Definition: proxy-io.h:622
std::shared_ptr< Impl > m_impl
Implementation pointer that may or may not be owned and deleted when this capnp server goes out of sc...
Definition: proxy.h:171
Mapping from capnp interface type to proxy server implementation (specializations are generated by pr...
Definition: proxy.h:28
CallContext_ CallContext
Definition: proxy-io.h:49
ServerInvokeContext(ProxyServer &proxy_server, CallContext &call_context, int req)
Definition: proxy-io.h:68
CallContext & call_context
Definition: proxy-io.h:52
ProxyServer & proxy_server
Definition: proxy-io.h:51
bool request_canceled
For IPC methods that execute asynchronously, not on the event-loop thread, this is set to true if the...
Definition: proxy-io.h:66
Lock * cancel_lock
For IPC methods that execute asynchronously, not on the event-loop thread: lock preventing the event-...
Definition: proxy-io.h:59
Vat id for server side of connection.
Definition: proxy-io.h:520
The thread_local ThreadContext g_thread_context struct provides information about individual threads ...
Definition: proxy-io.h:706
ConnThreads callback_threads MP_GUARDED_BY(waiter->m_mutex)
When client is making a request to a server, this is the callbackThread argument it passes in the req...
ConnThreads request_threads MP_GUARDED_BY(waiter->m_mutex)
When client is making a request to a server, this is the thread argument it passes in the request,...
std::string thread_name
Identifying string for debug.
Definition: proxy-io.h:708
Single element task queue used to handle recursive capnp calls.
Definition: proxy-io.h:386
void wait(Lock &lock, Predicate pred) MP_REQUIRES(m_mutex)
Definition: proxy-io.h:400
Mutex m_mutex
Mutex mainly used internally by waiter class, but also used externally to guard access to related sta...
Definition: proxy-io.h:425
std::optional< kj::Function< void()> > m_fn MP_GUARDED_BY(m_mutex)
std::condition_variable m_cv MP_GUARDED_BY(m_mutex)
bool post(Fn &&fn)
Definition: proxy-io.h:390
Waiter()=default
assert(!tx.IsCoinBase())