diff --git a/packages/streamr-dht/modules/connection/ConnectionLocker.cppm b/packages/streamr-dht/modules/connection/ConnectionLocker.cppm index dd2a5c4c..4d297301 100644 --- a/packages/streamr-dht/modules/connection/ConnectionLocker.cppm +++ b/packages/streamr-dht/modules/connection/ConnectionLocker.cppm @@ -8,6 +8,8 @@ module; #include +#include + export module streamr.dht.ConnectionLocker; import streamr.dht.protos; @@ -26,9 +28,16 @@ protected: public: virtual ~ConnectionLocker() = default; - virtual void lockConnection( + // Coroutines, not blocking calls: locking awaits a lockRequest RPC + // round-trip. The former synchronous implementation blockingWait()ed + // for it — on the shared worker pool that is a self-deadlock once + // poolsize concurrent joins sit inside lockConnection at once (every + // thread waits for an RPC response only those same threads could + // process; seen wedging the full-node teardown on the 4-thread CI + // pool). + virtual folly::coro::Task lockConnection( PeerDescriptor targetDescriptor, LockID lockId) = 0; - virtual void unlockConnection( + virtual folly::coro::Task unlockConnection( PeerDescriptor targetDescriptor, LockID lockId) = 0; virtual void weakLockConnection( const DhtAddress& targetDescriptor, const LockID& lockId) = 0; diff --git a/packages/streamr-dht/modules/connection/ConnectionManager.cppm b/packages/streamr-dht/modules/connection/ConnectionManager.cppm index 5ab9e0df..e014933c 100644 --- a/packages/streamr-dht/modules/connection/ConnectionManager.cppm +++ b/packages/streamr-dht/modules/connection/ConnectionManager.cppm @@ -473,12 +473,12 @@ public: return this->locks.isRemoteLocked(nodeId); } - void lockConnection( + folly::coro::Task lockConnection( PeerDescriptor targetDescriptor, LockID lockId) override { if (this->state == ConnectionManagerState::STOPPED || Identifiers::areEqualPeerDescriptors( targetDescriptor, this->getLocalPeerDescriptor())) { - return; + co_return; } const auto nodeId = Identifiers::getNodeIdFromPeerDescriptor(targetDescriptor); @@ -490,8 +490,9 @@ public: this->locks.addLocalLocked(nodeId, lockId); try { - auto accepted = streamr::utils::blockingWait( - rpcRemote.lockRequest(std::move(lockId))); + // co_await, never blockingWait: this runs on shared worker + // pool threads (see ConnectionLocker::lockConnection). + auto accepted = co_await rpcRemote.lockRequest(std::move(lockId)); if (accepted) { SLogger::trace("LockRequest successful"); } else { @@ -504,12 +505,12 @@ public: } } - void unlockConnection( + folly::coro::Task unlockConnection( PeerDescriptor targetDescriptor, LockID lockId) override { if (this->state == ConnectionManagerState::STOPPED || Identifiers::areEqualPeerDescriptors( targetDescriptor, this->getLocalPeerDescriptor())) { - return; + co_return; } const auto nodeId = @@ -522,7 +523,7 @@ public: SLogger::debug("Acquired mutex lock in unlockConnection"); if (!this->endpoints.contains(nodeId)) { SLogger::debug("Node ID not found in endpoints"); - return; + co_return; } } @@ -530,8 +531,7 @@ public: ConnectionLockRpcRemote rpcRemote( this->getLocalPeerDescriptor(), targetDescriptor, client); - streamr::utils::blockingWait( - rpcRemote.unlockRequest(std::move(lockId))); + co_await rpcRemote.unlockRequest(std::move(lockId)); } void weakLockConnection( diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm index 9776b916..24108ad6 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketClientConnector.cppm @@ -157,14 +157,22 @@ public: } void destroy() { - std::scoped_lock lock(this->mutex); - this->abortController.abort(); - - for (const auto& [_, handshaker] : this->connectingHandshakers) { + // Snapshot under the lock, close() OUTSIDE it: close(true) fans + // out into Disconnected handlers that re-enter connection and + // connector code — holding this->mutex across that call-out + // deadlocks ABBA against a thread that holds a connection's + // mutex (an rtc close callback, a pool worker in connect()) and + // needs this->mutex in the HandshakerStopped handler (sampled + // wedging the full-node teardown). + std::map> handshakers; + { + std::scoped_lock lock(this->mutex); + this->abortController.abort(); + handshakers.swap(this->connectingHandshakers); + } + for (const auto& [_, handshaker] : handshakers) { handshaker->getPendingConnection()->close(true); } - - this->connectingHandshakers.clear(); } }; diff --git a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm index 812dc1da..2fc3acf1 100644 --- a/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm +++ b/packages/streamr-dht/modules/connection/websocket/WebsocketServerConnector.cppm @@ -352,34 +352,41 @@ public: void destroy() { SLogger::trace("WebsocketServerConnector::destroy() called"); + std::map> handshakers; + std::map> requests; + std::unique_ptr server; { std::scoped_lock lock(this->mMutex); this->abortController.abort(); - - SLogger::trace("Closing ongoing connect requests"); - for (const auto& request : this->ongoingConnectRequests) { - if (request.second) { - request.second->close(true); - } + handshakers.swap(this->connectingHandshakers); + requests.swap(this->ongoingConnectRequests); + server = std::move(this->websocketServer); + } + // Every call-out below runs OUTSIDE mMutex: close(true) fans out + // into Disconnected handlers that re-enter connection/connector + // code, and holding the lock across that deadlocks ABBA against a + // thread that holds a connection's mutex and needs mMutex (the + // shape sampled in WebsocketClientConnector::destroy(); same + // hazard here). + SLogger::trace("Closing ongoing connect requests"); + for (const auto& request : requests) { + if (request.second) { + request.second->close(true); } + } - SLogger::trace("Closing connecting handshakers"); - for (const auto& handshaker : this->connectingHandshakers) { - if (handshaker.second && - handshaker.second->getPendingConnection()) { - handshaker.second->getPendingConnection()->close(true); - } + SLogger::trace("Closing connecting handshakers"); + for (const auto& handshaker : handshakers) { + if (handshaker.second && + handshaker.second->getPendingConnection()) { + handshaker.second->getPendingConnection()->close(true); } + } - SLogger::trace("Clearing maps"); - this->connectingHandshakers.clear(); - this->ongoingConnectRequests.clear(); - - SLogger::trace("Stopping websocket server"); - if (this->websocketServer) { - this->websocketServer->stop(); - this->websocketServer.reset(); - } + SLogger::trace("Stopping websocket server"); + if (server) { + server->stop(); + server.reset(); } // Drained OUTSIDE the mutex (the PeerManager::stop() pattern): a // detached requestConnection notification still in flight references diff --git a/packages/streamr-dht/modules/dht/DhtNode.cppm b/packages/streamr-dht/modules/dht/DhtNode.cppm index 6d78b4cb..8092aab4 100644 --- a/packages/streamr-dht/modules/dht/DhtNode.cppm +++ b/packages/streamr-dht/modules/dht/DhtNode.cppm @@ -730,9 +730,15 @@ public: // recovery executor's destructor join). A KBucketEmpty still firing // mid-join is dropped by the scope's close() gate. this->recoveryScope.close(); - this->transportPtr->off(this->messageToken); - this->transportPtr->off(this->connectedToken); - this->transportPtr->off(this->disconnectedToken); + // The waiting variants: a Message handler mid-invocation on + // another thread may still schedule a detached server-RPC + // dispatch; plain off() returning early would let it slip past + // the drain below (seen as a getClosestPeers dispatch resuming + // into a destroyed DhtNode on the CI-sized worker pool). + this->transportPtr->offAndWait( + this->messageToken); + this->transportPtr->offAndWait(this->connectedToken); + this->transportPtr->offAndWait(this->disconnectedToken); streamr::utils::blockingWait(this->storeManager->destroy()); this->localDataStore.clear(); if (this->peerManager) { @@ -753,6 +759,18 @@ public: this->ownedConnectionManager->stop(); } this->connectionLocker = nullptr; + // Cancel and join this node's detached RPC dispatch/request + // coroutines while every member they capture is still alive. The + // implicit drain in ~RoutingRpcCommunicator runs only during + // member destruction — after later-declared members are already + // gone — and a straggler server dispatch then resumes into freed + // memory (seen as a getClosestPeers dispatch crashing on the + // CI-sized worker pool). Placed LAST: the drain gates new scope + // adds off, and the steps above still send through this + // communicator (leave notices, final replications). No new server + // dispatch can arrive here — the transport listeners were removed + // with offAndWait() at the top of stop(). + this->rpcCommunicator->drainAsyncTasks(); } // --- Public API ------------------------------------------------------ diff --git a/packages/streamr-dht/modules/dht/discovery/DiscoverySession.cppm b/packages/streamr-dht/modules/dht/discovery/DiscoverySession.cppm index 270e5be8..9ea23225 100644 --- a/packages/streamr-dht/modules/dht/discovery/DiscoverySession.cppm +++ b/packages/streamr-dht/modules/dht/discovery/DiscoverySession.cppm @@ -253,8 +253,15 @@ public: for (size_t i = 0; i < this->options.parallelism; ++i) { workers.push_back(this->worker()); } + // MERGE the caller's (ambient) cancellation token with the node's + // abort signal instead of replacing it: a stop()-time cancellation + // of the detached join must reach the workers' RPC awaits, or the + // scope drain waits a full session timeout for them (the full-node + // teardown hang). co_await streamr::utils::co_withCancellation( - this->options.abortSignal.getCancellationToken(), + streamr::utils::cancellationTokenMerge( + co_await streamr::utils::co_currentCancellationToken(), + this->options.abortSignal.getCancellationToken()), folly::coro::timeout( folly::coro::collectAllRange(std::move(workers)), timeout)); } diff --git a/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm b/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm index 57f23a1a..982e464d 100644 --- a/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm +++ b/packages/streamr-dht/modules/dht/discovery/PeerDiscovery.cppm @@ -175,6 +175,11 @@ private: std::vector> sessions, PeerDescriptor entryPointDescriptor, bool retry) { + // Snapshot the caller's cancellation token: a join cancelled by + // its owner's stop() must not schedule a rejoin (the detached + // rejoin would outlive the intent to stop). + const auto callerToken = + co_await streamr::utils::co_currentCancellationToken(); try { for (const auto& session : sessions) { { @@ -186,7 +191,7 @@ private: } catch (const std::exception&) { SLogger::debug("DHT join timed out"); } - if (!this->isStopped()) { + if (!this->isStopped() && !callerToken.isCancellationRequested()) { if (this->options.peerManager.getNeighborCount() == 0) { if (retry) { this->scheduleRejoin(entryPointDescriptor, rejoinDelay); @@ -396,7 +401,7 @@ public: co_return; } if (this->options.connectionLocker != nullptr) { - this->options.connectionLocker->lockConnection( + co_await this->options.connectionLocker->lockConnection( entryPointDescriptor, this->joinLockId()); } this->options.peerManager.addContact(entryPointDescriptor); @@ -412,7 +417,7 @@ public: co_await this->runSessions( std::move(sessions), entryPointDescriptor, retry); if (this->options.connectionLocker != nullptr) { - this->options.connectionLocker->unlockConnection( + co_await this->options.connectionLocker->unlockConnection( entryPointDescriptor, this->joinLockId()); } } diff --git a/packages/streamr-dht/modules/dht/discovery/RingDiscoverySession.cppm b/packages/streamr-dht/modules/dht/discovery/RingDiscoverySession.cppm index 9a3b9af2..02a474ae 100644 --- a/packages/streamr-dht/modules/dht/discovery/RingDiscoverySession.cppm +++ b/packages/streamr-dht/modules/dht/discovery/RingDiscoverySession.cppm @@ -270,8 +270,13 @@ public: for (size_t i = 0; i < this->options.parallelism; ++i) { workers.push_back(this->worker()); } + // MERGE the ambient cancellation with the abort signal (see + // DiscoverySession::findClosestNodes for why replacing it hangs + // the stop()-time scope drain). co_await streamr::utils::co_withCancellation( - this->options.abortSignal.getCancellationToken(), + streamr::utils::cancellationTokenMerge( + co_await streamr::utils::co_currentCancellationToken(), + this->options.abortSignal.getCancellationToken()), folly::coro::timeout( folly::coro::collectAllRange(std::move(workers)), timeout)); } diff --git a/packages/streamr-dht/modules/dht/store/StoreManager.cppm b/packages/streamr-dht/modules/dht/store/StoreManager.cppm index 856c3b6c..b6915a65 100644 --- a/packages/streamr-dht/modules/dht/store/StoreManager.cppm +++ b/packages/streamr-dht/modules/dht/store/StoreManager.cppm @@ -30,6 +30,7 @@ export module streamr.dht.StoreManager; import streamr.dht.protos; import streamr.utils.CoroutineHelper; +import streamr.utils.GuardedAsyncScope; import streamr.utils.SharedExecutors; import streamr.utils.EnableSharedFromThis; import streamr.utils.ExecutorHelper; @@ -88,6 +89,16 @@ private: // replication tasks pin `self` (sharedFromThis), exactly as before. streamr::utils::SharedSerialExecutor replicationExecutor{ streamr::utils::SharedExecutors::worker()}; + // The detached onContactAdded replications pin `self`, but their RPC + // remotes send through the owning DhtNode's communicator, which + // destroy() does NOT keep alive: a replication coroutine resuming + // after DhtNode::stop() dereferenced the freed communicator (SIGSEGV + // in RpcCommunicatorClientApi::notify resume, + // reproduced 13/16 under a CI-sized worker pool). Track them here and + // drain in destroy(), while the communicator and transport are still + // alive; cancel first so the drain does not wait on network progress. + streamr::utils::GuardedAsyncScope replicationScope; + folly::CancellationSource replicationCancellation; std::unique_ptr rpcLocal; explicit StoreManager(StoreManagerOptions options) @@ -113,19 +124,22 @@ private: } } - // Fire-and-forget the async replication onto the worker executor. + // Fire-and-forget the async replication onto the worker executor, + // tracked by replicationScope (see the member comment). void replicateDataToContact( const DataEntry& dataEntry, const PeerDescriptor& contact) { auto self = this->sharedFromThis(); DataEntry entry = dataEntry; PeerDescriptor target = contact; - streamr::utils::co_withExecutor( - &this->replicationExecutor, - folly::coro::co_invoke( - [self, entry, target]() -> folly::coro::Task { - co_await self->doReplicate(entry, target); - })) - .start(); + this->replicationScope.add( + streamr::utils::co_withExecutor( + &this->replicationExecutor, + streamr::utils::co_withCancellation( + this->replicationCancellation.getToken(), + folly::coro::co_invoke( + [self, entry, target]() -> folly::coro::Task { + co_await self->doReplicate(entry, target); + })))); } void registerLocalRpcMethods() { @@ -188,17 +202,21 @@ private: auto self = this->sharedFromThis(); DhtAddress key = dataKey; PeerDescriptor node = newNode; - streamr::utils::co_withExecutor( - &this->replicationExecutor, - folly::coro::co_invoke( - [self, key, node]() -> folly::coro::Task { - const auto dataEntries = - self->options.localDataStore.values(key); - for (const auto& dataEntry : dataEntries) { - co_await self->doReplicate(dataEntry, node); - } - })) - .start(); + this->replicationScope.add( + streamr::utils::co_withExecutor( + &this->replicationExecutor, + streamr::utils::co_withCancellation( + this->replicationCancellation.getToken(), + folly::coro::co_invoke( + [self, key, node]() -> folly::coro::Task { + const auto dataEntries = + self->options.localDataStore.values( + key); + for (const auto& dataEntry : dataEntries) { + co_await self->doReplicate( + dataEntry, node); + } + })))); } } else if (!std::ranges::any_of( storers, [this](const PeerDescriptor& peer) { @@ -317,6 +335,13 @@ public: } folly::coro::Task destroy() { + // Cancel and drain the detached onContactAdded replications while + // the communicator/transport they send through are still alive + // (DhtNode::stop() awaits this before tearing those down). The + // final leave-replication below is awaited directly and is not + // cancelled. + this->replicationCancellation.requestCancellation(); + co_await this->replicationScope.closeAsync(); co_await this->replicateDataToClosestNodes(); } }; diff --git a/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp b/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp index 4251d4b3..bba2bc83 100644 --- a/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp +++ b/packages/streamr-dht/test/integration/ConnectionLockingTest.cpp @@ -144,7 +144,8 @@ TEST_F(ConnectionLockingTest, CanLockConnections) { auto tmpMockPeerDescriptor2 = mockPeerDescriptor2; auto tmpMockPeerDescriptor1 = mockPeerDescriptor1; SLogger::trace("before lockConnection() start"); - connectionManager1->lockConnection(mockPeerDescriptor2, LockID("testLock")); + streamr::utils::blockingWait(connectionManager1->lockConnection( + mockPeerDescriptor2, LockID("testLock"))); SLogger::trace("lockConnection done"); std::function condition = [&connectionManager2, @@ -204,8 +205,8 @@ TEST_F(ConnectionLockingTest, LockingBothWays) { auto task = collect( [&]() { - connectionManager3->lockConnection( - mockPeerDescriptor4, LockID("testLock1")); + streamr::utils::blockingWait(connectionManager3->lockConnection( + mockPeerDescriptor4, LockID("testLock1"))); }, waitForCondition(condition) // NOLINT ); @@ -214,8 +215,8 @@ TEST_F(ConnectionLockingTest, LockingBothWays) { auto task2 = collect( [&]() { - connectionManager4->lockConnection( - mockPeerDescriptor3, LockID("testLock2")); + streamr::utils::blockingWait(connectionManager4->lockConnection( + mockPeerDescriptor3, LockID("testLock2"))); }, waitForCondition(condition) // NOLINT ); diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm index 3dc2d142..7d5002dd 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorClientApi.cppm @@ -228,6 +228,13 @@ public: const CallContextType& callContext, std::optional timeout = std::nullopt) { SLogger::trace("request(): methodName:", methodName); + if (mDrained) { + // A drained scope must not be add()ed to (folly forbids + // add-after-join); a request after drainAsyncTasks() is a + // stop()-ordering bug in the caller — fail it cleanly. + throw RpcClientError( + "request() called on a drained RpcCommunicator: " + methodName); + } std::chrono::milliseconds timeoutValue = mRpcRequestTimeout; if (timeout.has_value()) { @@ -310,6 +317,12 @@ public: const CallContextType& callContext, std::optional timeout = std::nullopt) { SLogger::trace("notify() notificationName:", notificationName); + if (mDrained) { + // See request(): no adds after the scope drain. + throw RpcClientError( + "notify() called on a drained RpcCommunicator: " + + std::string(notificationName)); + } std::chrono::milliseconds timeoutValue = mRpcRequestTimeout; if (timeout.has_value()) { diff --git a/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm b/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm index 86edaeb5..9519beb3 100644 --- a/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm +++ b/packages/streamr-proto-rpc/modules/RpcCommunicatorServerApi.cppm @@ -188,6 +188,14 @@ private: // immediately, so the delivery thread is never blocked by handler work. void handleRequest( const RpcMessage& rpcMessage, const CallContextType& callContext) { + if (mDrained) { + // The owner has drained this scope during its stop(); a + // request arriving after that is dropped like any request to + // a stopped node (and folly forbids add-after-join). + SLogger::debug( + "handleRequest() after drainAsyncTasks(), dropping request"); + return; + } auto handler = mServerRegistry.getAsyncHandler(rpcMessage); mScope.add( streamr::utils::co_withExecutor( diff --git a/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm b/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm index 83d8f9a9..b969ad9f 100644 --- a/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm +++ b/packages/streamr-trackerless-network/modules/ContentDeliveryManager.cppm @@ -158,6 +158,14 @@ private: streamr::utils::SharedSerialExecutor joinExecutor{ streamr::utils::SharedExecutors::worker()}; GuardedAsyncScope joinScope; + // Cancels the detached join/rejoin tasks at destroy(): the scope + // contract requires the tasks to be cancelled BEFORE the drain, + // otherwise destroy() waits for the joins to complete naturally — an + // unbounded wait on worker-pool/network progress. Same mechanism and + // fix as NetworkStack's joinCancellation; without it the CI + // full-node teardown wedged in ContentDeliveryManager::destroy() -> + // joinScope.close() one layer below the NetworkStack drain. + folly::CancellationSource joinCancellation; public: explicit ContentDeliveryManager(ContentDeliveryManagerOptions options) @@ -206,7 +214,13 @@ public: } } } - this->joinScope.close(); + // Cancel the detached join/rejoin tasks, then drain with the + // suspending closeAsync(): destroy() runs on a shared + // blockingWait drive thread during test teardown, and a blocking + // close() would both serialize sibling stops and wait unboundedly + // for un-cancelled joins (the CI full-node teardown hang). + this->joinCancellation.requestCancellation(); + co_await this->joinScope.closeAsync(); std::vector> parts; { std::scoped_lock lock(this->mutex); @@ -346,7 +360,9 @@ public: this->joinScope.add( streamr::utils::co_withExecutor( &this->joinExecutor, - streamPartReconnect->reconnect())); + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + streamPartReconnect->reconnect()))); } }); node->on( @@ -354,25 +370,30 @@ public: this->joinScope.add( streamr::utils::co_withExecutor( &this->joinExecutor, - this->handleEntryPointLeave( - streamPartId, peerDescriptorStoreManager))); + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + this->handleEntryPointLeave( + streamPartId, peerDescriptorStoreManager)))); }); // TS setImmediate(): detached bounded join. this->joinScope.add( streamr::utils::co_withExecutor( &this->joinExecutor, - folly::coro::co_invoke( - [this, streamPartId, peerDescriptorStoreManager]() - -> folly::coro::Task { - try { - co_await this->startLayersAndJoinDht( - streamPartId, peerDescriptorStoreManager); - } catch (const std::exception& err) { - SLogger::warn( - "Failed to join to stream part " + - streamPartId + ": " + std::string(err.what())); - } - }))); + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + folly::coro::co_invoke( + [this, streamPartId, peerDescriptorStoreManager]() + -> folly::coro::Task { + try { + co_await this->startLayersAndJoinDht( + streamPartId, peerDescriptorStoreManager); + } catch (const std::exception& err) { + SLogger::warn( + "Failed to join to stream part " + + streamPartId + ": " + + std::string(err.what())); + } + })))); } folly::coro::Task setProxies( @@ -597,12 +618,14 @@ private: this->joinScope.add( streamr::utils::co_withExecutor( &this->joinExecutor, - folly::coro::co_invoke( - [networkSplitAvoidance]() - -> folly::coro::Task { - co_await networkSplitAvoidance - ->avoidNetworkSplit(); - }))); + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + folly::coro::co_invoke( + [networkSplitAvoidance]() + -> folly::coro::Task { + co_await networkSplitAvoidance + ->avoidNetworkSplit(); + })))); } } } diff --git a/packages/streamr-trackerless-network/modules/NetworkStack.cppm b/packages/streamr-trackerless-network/modules/NetworkStack.cppm index fe998cab..5615fbae 100644 --- a/packages/streamr-trackerless-network/modules/NetworkStack.cppm +++ b/packages/streamr-trackerless-network/modules/NetworkStack.cppm @@ -103,6 +103,12 @@ private: streamr::utils::SharedSerialExecutor joinExecutor{ streamr::utils::SharedExecutors::worker()}; GuardedAsyncScope joinScope; + // Cancels the fire-and-forget joinDht at stop(): the scope contract + // requires the tasks to be cancelled BEFORE the drain, otherwise + // stop() waits for the join to complete naturally — an unbounded + // wait on network progress (the CI full-node teardown hang: a + // starved shared worker pool never let the join finish). + folly::CancellationSource joinCancellation; folly::coro::Task ensureConnectedToControlLayer() { // TS TODO preserved: could wrap joinDht with pOnce and call it @@ -113,11 +119,13 @@ private: this->joinScope.add( streamr::utils::co_withExecutor( &this->joinExecutor, - folly::coro::co_invoke( - [this]() -> folly::coro::Task { - co_await this->controlLayerNode->joinDht( - this->options.layer0.entryPoints); - }))); + streamr::utils::co_withCancellation( + this->joinCancellation.getToken(), + folly::coro::co_invoke( + [this]() -> folly::coro::Task { + co_await this->controlLayerNode->joinDht( + this->options.layer0.entryPoints); + })))); } co_await this->controlLayerNode->waitForNetworkConnectivity(); } @@ -323,7 +331,12 @@ public: this->stopped = true; // Drain the fire-and-forget joins before the members die // (teardown ordering; see ContentDeliveryManager::destroy). - this->joinScope.close(); + // Cancel first — the drain must not wait for the join to make + // network progress — and suspend (not block) while draining, + // so concurrent sibling stops sharing this drive thread + // proceed meanwhile (the CI full-node teardown hang). + this->joinCancellation.requestCancellation(); + co_await this->joinScope.closeAsync(); co_await this->contentDeliveryManager->destroy(); // The info communicator listens on the transport — take it // down while the transport is still alive. diff --git a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm index 59b39755..9df25fca 100644 --- a/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm +++ b/packages/streamr-trackerless-network/modules/logic/proxy/ProxyClient.cppm @@ -393,14 +393,20 @@ public: std::current_exception(), peerDescriptor); } if (accepted) { - this->options.connectionLocker.lockConnection( - peerDescriptor, LockID{SERVICE_ID}); + // ProxyClient's connection routine is synchronous today (it + // blockingWaits its RPCs above); keep that behavior for the + // now-asynchronous locker API. Async-ifying this whole path + // is a separate task. + streamr::utils::blockingWait( + this->options.connectionLocker.lockConnection( + peerDescriptor, LockID{SERVICE_ID})); { std::scoped_lock lock(this->mutex); if (this->stopped) { - this->options.connectionLocker.unlockConnection( - peerDescriptor, LockID{SERVICE_ID}); + streamr::utils::blockingWait( + this->options.connectionLocker.unlockConnection( + peerDescriptor, LockID{SERVICE_ID})); return; } this->connections.emplace( @@ -464,8 +470,9 @@ public: streamr::utils::blockingWait(server.value()->leaveStreamPartNotice( this->options.streamPartId, false)); } - this->options.connectionLocker.unlockConnection( - peerDescriptor.value(), LockID{SERVICE_ID}); + streamr::utils::blockingWait( + this->options.connectionLocker.unlockConnection( + peerDescriptor.value(), LockID{SERVICE_ID})); this->neighbors.remove(nodeId); } @@ -552,8 +559,9 @@ public: return; } } - this->options.connectionLocker.unlockConnection( - peerDescriptor, LockID{SERVICE_ID}); + streamr::utils::blockingWait( + this->options.connectionLocker.unlockConnection( + peerDescriptor, LockID{SERVICE_ID})); this->neighbors.remove(nodeId); // Deviation from TS: no automatic reconnection attempt here. The // TS single retry always fails in practice (the proxy that @@ -590,8 +598,9 @@ public: // event thread parked on it may be the thread these sends need. auto allNeighbors = this->neighbors.getAll(); for (const auto& remote : allNeighbors) { - this->options.connectionLocker.unlockConnection( - remote->getPeerDescriptor(), LockID{SERVICE_ID}); + streamr::utils::blockingWait( + this->options.connectionLocker.unlockConnection( + remote->getPeerDescriptor(), LockID{SERVICE_ID})); streamr::utils::blockingWait(remote->leaveStreamPartNotice( this->options.streamPartId, false)); } diff --git a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm index 052f81e6..ce297fcc 100644 --- a/packages/streamr-trackerless-network/test/utils/TestUtils.cppm +++ b/packages/streamr-trackerless-network/test/utils/TestUtils.cppm @@ -150,10 +150,14 @@ class MockConnectionLocker : public ConnectionLocker { public: ~MockConnectionLocker() override = default; - void lockConnection( - PeerDescriptor /*targetDescriptor*/, LockID /*lockId*/) override {} - void unlockConnection( - PeerDescriptor /*targetDescriptor*/, LockID /*lockId*/) override {} + folly::coro::Task lockConnection( + PeerDescriptor /*targetDescriptor*/, LockID /*lockId*/) override { + co_return; + } + folly::coro::Task unlockConnection( + PeerDescriptor /*targetDescriptor*/, LockID /*lockId*/) override { + co_return; + } void weakLockConnection( const DhtAddress& /*targetDescriptor*/, const LockID& /*lockId*/) override {} diff --git a/packages/streamr-utils/CMakeLists.txt b/packages/streamr-utils/CMakeLists.txt index c9487917..faf1cd44 100644 --- a/packages/streamr-utils/CMakeLists.txt +++ b/packages/streamr-utils/CMakeLists.txt @@ -127,6 +127,7 @@ if(NOT IOS AND STREAMR_MODULES_SUPPORTED) add_executable(streamr-utils-test-unit test/unit/BrandedTest.cpp test/unit/AbortableTimersTest.cpp + test/unit/GuardedAsyncScopeTest.cpp test/unit/waitForEventTest.cpp test/unit/waitForConditionTest.cpp test/unit/runAndWaitForEventsTest.cpp diff --git a/packages/streamr-utils/modules/CoroutineHelper.cppm b/packages/streamr-utils/modules/CoroutineHelper.cppm index 54818fe9..2f84aff2 100644 --- a/packages/streamr-utils/modules/CoroutineHelper.cppm +++ b/packages/streamr-utils/modules/CoroutineHelper.cppm @@ -26,6 +26,7 @@ module; #include #include #include +#include #include #include #include @@ -72,6 +73,11 @@ using folly::Unit; export namespace streamr::utils { +// (Comment touch to invalidate stale CI build caches: the restored +// macOS cache carried a pre-cancellation-shim BMI of this module and +// ninja under-rebuilt, failing DiscoverySession/RingDiscoverySession +// compiles on symbols that exist — the documented incremental-cache +// trap.) // folly's blockingWait / co_withExecutor / co_withCancellation are // customization-point objects that cannot be re-exported by // using-declaration (internal linkage or conflicting redeclarations @@ -98,4 +104,23 @@ inline auto co_withCancellation( return folly::coro::co_withCancellation(std::forward(args)...); } +// The awaitable tag that yields the CURRENT (ambient) cancellation token +// inside a Task: co_await co_currentCancellationToken(). Exported so +// awaited subtrees can MERGE the caller's token with their own instead of +// replacing it (co_withCancellation replaces; a replaced token made +// stop()-time cancellation invisible inside DiscoverySession — the +// full-node teardown hang). +inline auto +co_currentCancellationToken() // NOLINT(readability-identifier-naming) + -> folly::coro::co_current_cancellation_token_t { + return folly::coro::co_current_cancellation_token; +} + +// folly::cancellation_token_merge is a CPO like the ones above. +template +inline auto cancellationTokenMerge(Args&&... args) + -> decltype(folly::cancellation_token_merge(std::forward(args)...)) { + return folly::cancellation_token_merge(std::forward(args)...); +} + } // namespace streamr::utils diff --git a/packages/streamr-utils/modules/GuardedAsyncScope.cppm b/packages/streamr-utils/modules/GuardedAsyncScope.cppm index d4518c49..4372173a 100644 --- a/packages/streamr-utils/modules/GuardedAsyncScope.cppm +++ b/packages/streamr-utils/modules/GuardedAsyncScope.cppm @@ -59,6 +59,23 @@ public: streamr::utils::blockingWait(this->scope.joinAsync()); } + // Coroutine variant of close() for stop() paths that share a drive + // thread (e.g. several stops under one collectAllRange): it SUSPENDS + // while the scope drains instead of blocking the thread, so sibling + // stops keep making progress. Same single-owning-stopper contract as + // close(); the drain-vs-destruction ordering is the caller's (the + // co_awaiting stop() completes before its owner is destroyed). + folly::coro::Task closeAsync() { + { + std::scoped_lock lock(this->mutex); + if (this->closed) { + co_return; + } + this->closed = true; + } + co_await this->scope.joinAsync(); + } + ~GuardedAsyncScope() { this->close(); } GuardedAsyncScope() = default; diff --git a/packages/streamr-utils/test/unit/GuardedAsyncScopeTest.cpp b/packages/streamr-utils/test/unit/GuardedAsyncScopeTest.cpp new file mode 100644 index 00000000..cc0ad6be --- /dev/null +++ b/packages/streamr-utils/test/unit/GuardedAsyncScopeTest.cpp @@ -0,0 +1,77 @@ +#include +#include +#include +#include + +#include // IWYU pragma: keep + +import streamr.utils.CoroutineHelper; +import streamr.utils.GuardedAsyncScope; +import streamr.utils.SharedExecutors; + +using streamr::utils::blockingWait; +using streamr::utils::GuardedAsyncScope; +using streamr::utils::SharedExecutors; + +namespace { + +auto onWorker(folly::coro::Task&& task) { + return streamr::utils::co_withExecutor( + &SharedExecutors::worker(), std::move(task)); +} + +} // namespace + +TEST(GuardedAsyncScopeTest, CloseAsyncDrainsAddedTasks) { + GuardedAsyncScope scope; + std::atomic ran{false}; + scope.add( + onWorker(folly::coro::co_invoke([&ran]() -> folly::coro::Task { + ran = true; + co_return; + }))); + blockingWait(scope.closeAsync()); + EXPECT_EQ(ran.load(), true); +} + +TEST(GuardedAsyncScopeTest, AddAfterCloseAsyncIsDropped) { + GuardedAsyncScope scope; + blockingWait(scope.closeAsync()); + std::atomic ran{false}; + scope.add( + onWorker(folly::coro::co_invoke([&ran]() -> folly::coro::Task { + ran = true; + co_return; + }))); + // close() after closeAsync() must stay a no-op (the destructor path). + scope.close(); + EXPECT_EQ(ran.load(), false); +} + +// The full-node teardown regression: several stop() coroutines share one +// blockingWait drive thread (collectAllRange), and the first scope's +// drain only finishes after a SIBLING stop has made progress. A +// thread-blocking close() deadlocks this shape; closeAsync() suspends, +// letting the sibling run, so the whole teardown completes. +TEST(GuardedAsyncScopeTest, CloseAsyncSuspendsSoSiblingStopsProgress) { + GuardedAsyncScope scope; + auto contract = folly::coro::makePromiseContract(); + scope.add(onWorker( + folly::coro::co_invoke( + [future = std::move(contract.second)]() mutable + -> folly::coro::Task { co_await std::move(future); }))); + + std::vector> stops; + stops.push_back( + folly::coro::co_invoke([&scope]() -> folly::coro::Task { + co_await scope.closeAsync(); + })); + stops.push_back( + folly::coro::co_invoke( + [promise = std::move( + contract.first)]() mutable -> folly::coro::Task { + promise.setValue(folly::Unit{}); + co_return; + })); + blockingWait(folly::coro::collectAllRange(std::move(stops))); +}