From 64c302ad7f39d1ce05529202523e32490265deb8 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sun, 12 Jul 2026 02:56:29 +0300 Subject: [PATCH] Make RecursiveOperationManager::sendResponse fully asynchronous MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #81 fixed the permanent worker-pool deadlock in sendResponse (the per-session ListeningRpcCommunicator was destroyed on a pool thread; the destructor's scope join parked the thread waiting for the communicator's own queued send task, which needed a pool thread). One bounded park remained: RecursiveOperationSessionRpcRemote::sendResponse wrapped the generated client notify in blockingWait(), parking the RPC handler's shared worker-pool thread for up to sessionRpcTimeout (10 s) under pool saturation. This change removes that park. Design: - RecursiveOperationSessionRpcRemote::sendResponse is now a folly::coro::Task coroutine that co_awaits the generated client notify (which must be co_awaited in-frame per the repo's lazy-task rule: the request locals live in the awaiting frame) and swallows failures, keeping the TS fire-and-forget semantics. - RecursiveOperationManager::sendResponse no longer does any of the work on the calling thread: the whole send runs as a detached coroutine on a new manager-owned GuardedAsyncScope (sendResponseScope), started on SharedExecutors::worker(). The task constructs the per-session ListeningRpcCommunicator and the RpcRemote, co_awaits the notify (suspends instead of parking), then disposes of the communicator. - Disposal keeps the deadlock rule from PR #81: after the notify completes, the communicator's own client scope may still hold the send task's unfinished tail, and the coroutine may even have been resumed from inside that task, so a non-empty-scope join here could self-deadlock. The task destroys the communicator in place only when pendingAsyncTaskCount() == 0 (an empty-scope join needs no pool thread, and a zero count proves the coroutine is not running inside one of the communicator's tasks; after destroy() detaches the transport listeners the count cannot grow, so the observation is stable). Otherwise it retires the communicator into the existing retiredSessionCommunicators backstop, pruned as before. - The communicator is constructed INSIDE the detached task, so a sendResponse racing stop() is dropped by the scope's close() gate without ever creating the communicator: the dropped lambda destroys only plain values at the add() call site and cannot block. A dropped response after stop is correct for a fire-and-forget send. - stop() closes sendResponseScope (on the owner's thread, while the session transport is still alive per DhtNode::stop() ordering) BEFORE clearing the retired list, because draining tasks may still retire their communicators into it. The scope is declared after the list so destructor ordering matches. Test updates: - RecursiveOperationManagerTest: the response send is now detached, so the transport.sendCount assertions moved after manager->stop(), which drains the send scope and makes the observation deterministic. - RecursiveOperationSessionTest: the RpcRemote sendResponse coroutine is blockingWait()ed on the test thread (not a pool worker). Verification (macOS arm64 Debug; pool size forced via a temporary STREAMR_TEST_WORKER_THREADS override in SharedExecutors::worker(), removed before commit): - DhtNodeExternalApiTest.ExternalStoreDataHappyPath: 15/15 pass with a 4-thread pool, 10/10 with 2 threads, 5/5 with 1 thread (~0.1-0.4 s per run). Before this change the 1-thread runs failed after ~10 s (the bounded blockingWait park swallowed the only worker until the notify timed out); no hangs anywhere (90 s watchdog per run). - Full suites green: streamr-dht-test-unit (181), streamr-dht-test- integration (42/43; the one failure was the known load-sensitive Layer1ScaleTest.SingleLayer1Dht flake, which exercises only joinDht — recursive operations never run in it — and passed when rerun in isolation), streamr-dht-test-end-to-end 4x (3x + once on the final clean build), streamr-proto-rpc-test-unit (24), streamr-proto-rpc-test-integration (4). Co-Authored-By: Claude Fable 5 --- .../RecursiveOperationManager.cppm | 176 ++++++++++++++---- .../RecursiveOperationSessionRpcRemote.cppm | 25 ++- .../unit/RecursiveOperationManagerTest.cpp | 8 +- .../unit/RecursiveOperationSessionTest.cpp | 6 +- 4 files changed, 166 insertions(+), 49 deletions(-) diff --git a/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm b/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm index 700b9b0f..4fefb423 100644 --- a/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm +++ b/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationManager.cppm @@ -33,6 +33,8 @@ import streamr.dht.protos; import streamr.utils.CoroutineHelper; import streamr.utils.EnableSharedFromThis; +import streamr.utils.GuardedAsyncScope; +import streamr.utils.SharedExecutors; import streamr.utils.waitForEvent; import streamr.logger.SLogger; import streamr.protorpc.RpcCommunicator; @@ -116,20 +118,25 @@ private: ongoingSessions; bool stopped = false; std::unique_ptr rpcLocal; - // sendResponse()'s per-session communicators, kept alive until their - // fire-and-forget send task settles. sendResponse runs on shared - // worker-pool threads (RPC server handlers); destroying the - // communicator there joins its async scope, and that join parks the - // pool thread waiting for the communicator's own queued send task, - // which needs a pool thread — with every worker parked in such a join - // the queued sends can never run and the pool deadlocks permanently - // (linux-arm64 CI: DhtNodeExternalApiTest.ExternalStoreDataHappyPath - // hung its 1200 s timeout; reproduced deterministically with a - // 1-thread pool). Retired communicators are destroyed only once their - // scopes are empty (an empty-scope join needs no pool thread) or in - // stop(), which runs on the owner's thread, never on the pool. + // sendResponse()'s per-session communicators whose internal scopes + // were still non-empty when their detached send task finished with + // them. Destroying a communicator joins its async scope; doing that on + // a shared worker-pool thread while the scope still holds a task parks + // the pool thread waiting for work that itself needs a pool thread — + // with every worker parked in such a join the pool deadlocks + // permanently (linux-arm64 CI: the former in-handler destruction hung + // DhtNodeExternalApiTest.ExternalStoreDataHappyPath for its 1200 s + // timeout; reproduced deterministically with a 1-thread pool). Retired + // communicators are destroyed only once their scopes are empty (an + // empty-scope join needs no pool thread) or in stop(), which runs on + // the owner's thread, never on the pool. std::vector> retiredSessionCommunicators; + // Runs sendResponse()'s detached fire-and-forget send tasks. Declared + // AFTER retiredSessionCommunicators so that its destructor (which + // joins the tasks) runs BEFORE the list dies: a draining task may + // still retire its communicator into the list. + streamr::utils::GuardedAsyncScope sendResponseScope; // Called under this->mutex. Frees every retired communicator whose // send task has settled; the rest stay retired until the next call or @@ -236,30 +243,107 @@ private: dataEntries, noCloserNodesFound); } else { - auto remoteCommunicator = - std::make_shared( - ServiceID{serviceId}, - this->options.sessionTransport, - RpcCommunicatorOptions{ - .rpcRequestTimeout = sessionRpcTimeout}); - RecursiveOperationSessionRpcRemote rpcRemote( - this->options.localPeerDescriptor, - targetPeerDescriptor, - RecursiveOperationSessionRpcClient(*remoteCommunicator), - sessionRpcTimeout); - rpcRemote.sendResponse( - routingPath, - closestConnectedNodes, - dataEntries, - noCloserNodesFound); - remoteCommunicator->destroy(); - // Do NOT destroy the communicator here: this runs on a worker - // pool thread and the destructor's scope join would park it - // (see retiredSessionCommunicators). - std::scoped_lock lock(this->mutex); - this->pruneRetiredSessionCommunicators(); - this->retiredSessionCommunicators.push_back( - std::move(remoteCommunicator)); + // Fire-and-forget response send. This method runs on shared + // worker-pool threads (RPC server handlers via doRouteRequest), + // so nothing here may park the thread: neither a blockingWait + // on the notify (a park bounded by sessionRpcTimeout, but a + // real thread lost for its duration under pool saturation) nor + // the per-session communicator's destructor (whose scope join + // once parked every worker permanently, see + // retiredSessionCommunicators). The whole send therefore runs + // as a detached task on sendResponseScope: it co_awaits the + // notify (suspends instead of parking) and then disposes of + // the communicator without ever joining a non-empty scope from + // a pool thread. + // + // Everything, including the communicator, is constructed + // INSIDE the task: if the scope has already been closed by + // stop(), the dropped lambda destroys only plain values + // (vectors and protobuf messages) here at the add() call site + // — dropping a response after stop is correct for a + // fire-and-forget send, and the drop cannot block. + std::weak_ptr weakSelf = + this->sharedFromThis(); + this->sendResponseScope.add( + streamr::utils::co_withExecutor( + &streamr::utils::SharedExecutors::worker(), + folly::coro::co_invoke( + [weakSelf, + serviceId, + targetPeerDescriptor, + routingPath, + closestConnectedNodes, + dataEntries, + noCloserNodesFound]() mutable + -> folly::coro::Task { + // The manager outlives every task here: its + // owner keeps a reference across stop(), whose + // close() drains this scope. The lock can only + // fail if the manager is torn down without + // stop() (the scope member's destructor join), + // in which case nothing has been created yet + // and dropping the response is correct. + auto self = weakSelf.lock(); + if (!self) { + co_return; + } + std::shared_ptr + remoteCommunicator; + try { + remoteCommunicator = + std::make_shared( + ServiceID{serviceId}, + self->options.sessionTransport, + RpcCommunicatorOptions{ + .rpcRequestTimeout = + sessionRpcTimeout}); + RecursiveOperationSessionRpcRemote rpcRemote( + self->options.localPeerDescriptor, + targetPeerDescriptor, + RecursiveOperationSessionRpcClient( + *remoteCommunicator), + sessionRpcTimeout); + co_await rpcRemote.sendResponse( + std::move(routingPath), + std::move(closestConnectedNodes), + std::move(dataEntries), + noCloserNodesFound); + } catch (...) { + // The scope requires tasks that never + // complete with an exception; a lost + // response is fine (fire-and-forget). + SLogger::debug( + "sendResponse task failed to send " + "RecursiveOperationResponse"); + } + if (remoteCommunicator) { + remoteCommunicator->destroy(); + // Dispose of the communicator. The notify + // has completed, but the send task it + // queued on the communicator's own scope + // may still have a tiny unfinished tail — + // and this coroutine may even have been + // resumed from inside that very task, so + // joining a non-empty scope here could + // self-deadlock. A zero count proves both + // that the join is a no-op and that we are + // not running inside one of its tasks, so + // the communicator may die on this thread; + // destroy() above detached the transport + // listeners, so the count can no longer + // grow and the zero observation is stable. + // Otherwise retire it; the pruning in + // later sendResponse calls and stop() is + // the backstop. + if (remoteCommunicator + ->pendingAsyncTaskCount() != 0) { + std::scoped_lock lock(self->mutex); + self->pruneRetiredSessionCommunicators(); + self->retiredSessionCommunicators.push_back( + std::move(remoteCommunicator)); + } + } + }))); } } @@ -424,7 +508,6 @@ public: } void stop() { - std::vector> retiredToDestroy; { std::scoped_lock lock(this->mutex); this->stopped = true; @@ -432,11 +515,26 @@ public: session->stop(); } this->ongoingSessions.clear(); + } + // Drain the detached response sends BEFORE destroying the retired + // communicators: a draining task may still retire its communicator + // into the list (it takes this->mutex to do so, which is why the + // close runs outside the lock). close() blocks — bounded by + // sessionRpcTimeout, the cap on every send task — and runs on the + // owner's thread (DhtNode::stop() calls this while the session + // transport is still alive), never on a pool worker, so the join + // may safely wait for tasks running on the pool. sendResponse + // calls racing this close are dropped by the scope's gate, which + // is correct for a fire-and-forget response after stop. + this->sendResponseScope.close(); + std::vector> retiredToDestroy; + { + std::scoped_lock lock(this->mutex); retiredToDestroy.swap(this->retiredSessionCommunicators); } // Destroyed outside the mutex on the stopper's thread (never a - // pool worker), where the scope joins may safely wait for the - // in-flight send tasks to settle on the pool. + // pool worker), where the destructors' scope joins may safely wait + // for the send-task tails to settle on the pool. retiredToDestroy.clear(); } }; diff --git a/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationSessionRpcRemote.cppm b/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationSessionRpcRemote.cppm index 63d19d55..663759e6 100644 --- a/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationSessionRpcRemote.cppm +++ b/packages/streamr-dht/modules/dht/recursive-operation/RecursiveOperationSessionRpcRemote.cppm @@ -51,10 +51,23 @@ public: std::move(client), timeout) {} - void sendResponse( - const std::vector& routingPath, - const std::vector& closestConnectedNodes, - const std::vector& dataEntries, + // Fire-and-forget (TS parity): builds the report, co_awaits the + // generated client notify (bounded by the RpcRemote timeout) and + // swallows send failures. Returning a Task instead of blocking keeps + // the caller's thread free — the former blockingWait here parked a + // shared worker-pool thread for up to the RPC timeout under pool + // saturation (see RecursiveOperationManager::sendResponse). + // + // NOTE the repo-wide lazy-task rule: the generated client call below + // returns a LAZY task whose request locals (report/options) live in + // THIS frame, so it must be co_awaited here — plain-returning it would + // dangle the locals (SEGV in Any::PackFrom). `this` and the + // communicator behind the client must stay alive until the returned + // task completes. + folly::coro::Task sendResponse( + std::vector routingPath, + std::vector closestConnectedNodes, + std::vector dataEntries, bool noCloserNodesFound) { RecursiveOperationResponse report; for (const auto& peer : routingPath) { @@ -69,8 +82,8 @@ public: report.set_noclosernodesfound(noCloserNodesFound); auto options = this->formDhtRpcOptions(); try { - streamr::utils::blockingWait(this->getClient().sendResponse( - std::move(report), std::move(options), this->getTimeout())); + co_await this->getClient().sendResponse( + std::move(report), std::move(options), this->getTimeout()); } catch (const std::exception& /*err*/) { SLogger::trace("Failed to send RecursiveOperationResponse"); } diff --git a/packages/streamr-dht/test/unit/RecursiveOperationManagerTest.cpp b/packages/streamr-dht/test/unit/RecursiveOperationManagerTest.cpp index 29bba2d6..08eecd7e 100644 --- a/packages/streamr-dht/test/unit/RecursiveOperationManagerTest.cpp +++ b/packages/streamr-dht/test/unit/RecursiveOperationManagerTest.cpp @@ -281,8 +281,10 @@ TEST_F(RecursiveOperationManagerTest, NoTargets) { "routeRequest", this->createRoutedMessage()); ASSERT_TRUE(ack.has_error()); EXPECT_EQ(ack.error(), RouteMessageError::NO_TARGETS); - EXPECT_EQ(this->transport.sendCount, 1U); + // The response send is a detached task; stop() drains it, so the + // sendCount observation after stop() is deterministic. manager->stop(); + EXPECT_EQ(this->transport.sendCount, 1U); } TEST_F(RecursiveOperationManagerTest, Error) { @@ -293,6 +295,8 @@ TEST_F(RecursiveOperationManagerTest, Error) { "routeRequest", this->createRoutedMessage()); ASSERT_TRUE(ack.has_error()); EXPECT_EQ(ack.error(), RouteMessageError::DUPLICATE); - EXPECT_EQ(this->transport.sendCount, 0U); + // stop() drains the detached response-send scope first, so a zero + // sendCount after it proves no response was (or will be) sent. manager->stop(); + EXPECT_EQ(this->transport.sendCount, 0U); } diff --git a/packages/streamr-dht/test/unit/RecursiveOperationSessionTest.cpp b/packages/streamr-dht/test/unit/RecursiveOperationSessionTest.cpp index abadc209..d1f9a9a2 100644 --- a/packages/streamr-dht/test/unit/RecursiveOperationSessionTest.cpp +++ b/packages/streamr-dht/test/unit/RecursiveOperationSessionTest.cpp @@ -67,11 +67,13 @@ class RecursiveOperationSessionTest : public ::testing::Test { mockPeerDescriptor, this->localPeerDescriptor, RecursiveOperationSessionRpcClient(communicator)); - remote.sendResponse( + // sendResponse is now a coroutine (fire-and-forget notify); + // blocking here is fine, the test thread is not a pool worker. + blockingWait(remote.sendResponse( {createMockPeerDescriptor(), createMockPeerDescriptor()}, {createMockPeerDescriptor(), createMockPeerDescriptor()}, {}, - true); + true)); } void TearDown() override {