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 {