Inside Bitcoin

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
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
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.
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
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.
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.";
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
874struct 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
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
908//! Given stream file descriptor and an init object, handle requests on the
909//! stream by calling methods on the Init object.
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
920//! Given listening socket file descriptor and an init object, handle incoming
921//! connections and requests by calling methods on the Init object.
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
941