From e0d28a7856d8c5111f3712031d60664a7446cdc4 Mon Sep 17 00:00:00 2001 From: Petri Savolainen Date: Sat, 11 Jul 2026 16:50:21 +0300 Subject: [PATCH] Fix Layer1ScaleTest flake: concurrent per-association Simulator delivery Three evidence-backed fixes (baseline: 5/15 isolated runs failed the neighbor-count assertion; after: 20/20 pass): 1. Simulator: the single dispatcher thread executed every delivered operation's entire downstream processing inline, becoming the whole simulated network's serialization point. Under the 49-node layer-1 join wave (~59k deliveries) its queue backed up ~6,000 deep and delivery latency reached 8-14s (measured), blowing every RPC timeout so discovery/ping timeout-pruning removed live neighbors as fast as they were found. Deliveries now run on a per-association serial view of the shared worker pool: per-association FIFO is preserved (the guarantee real transports give), associations deliver concurrently, and stop() drains in-flight handoffs. 2. RoutingSession: consecutive RoutingTablesCache lookups (has, get for getSize, get) each re-evaluate the 15s TTL, so a later lookup could expire the entry and return null (value_or(nullptr)) microseconds after an earlier one succeeded; the null table was then dereferenced (SIGSEGV in getClosestContacts, seen in a crash report). Fetch once and rebuild when null or empty. 3. Layer1ScaleTest: the connection-count/list equality assertions read the SAME live ConnectionsView twice; single-threaded TS evaluates them atomically (tautology) but in C++ background connection churn mutates the view between reads (observed 25 vs 27). Assert the view identity that makes the TS equalities true. Co-Authored-By: Claude Fable 5 --- .../connection/simulator/Simulator.cppm | 189 ++++++++++++------ .../modules/dht/routing/RoutingSession.cppm | 15 +- .../test/integration/Layer1ScaleTest.cpp | 34 ++-- 3 files changed, 150 insertions(+), 88 deletions(-) diff --git a/packages/streamr-dht/modules/connection/simulator/Simulator.cppm b/packages/streamr-dht/modules/connection/simulator/Simulator.cppm index ced18d24..78e8eab6 100644 --- a/packages/streamr-dht/modules/connection/simulator/Simulator.cppm +++ b/packages/streamr-dht/modules/connection/simulator/Simulator.cppm @@ -51,6 +51,7 @@ import streamr.dht.Identifiers; import streamr.dht.RegionPings; import streamr.dht.SimulatorInterfaces; import streamr.logger.SLogger; +import streamr.utils.SharedExecutors; // Hoisted from the former header (file scope, NOT exported); // fully qualified: relative namespace names resolve differently @@ -78,6 +79,20 @@ private: connectedCallback; // only on the connecting side Clock::time_point lastOperationAt{}; bool closing = false; + // Operations on ONE association execute in order on this serial + // view of the shared worker pool; different associations deliver + // concurrently. The dispatcher thread only sequences deadlines — + // it must not execute the receivers' processing itself: with all + // deliveries serialized behind one thread, a 49-node join wave + // (~59k deliveries) backs the queue up thousands deep and delivery + // latency grows to ~8-14s, past every RPC timeout in the system, + // so discovery/ping timeout-pruning tears down live neighbours as + // fast as they are found (the Layer1ScaleTest flake). TS runs the + // deliveries on its single event loop — where they cost little — + // and nothing in the protocol depends on cross-association order: + // the real transports (websocket, WebRTC datachannel) guarantee + // only per-connection FIFO, which is exactly what is kept here. + std::shared_ptr executor; }; enum class OperationType : std::uint8_t { CONNECT, SEND, CLOSE }; @@ -112,6 +127,10 @@ private: std::condition_variable mCondition; bool stopped = false; uint64_t nextSequenceNumber = 0; + // Operations handed off to association executors but not yet finished; + // stop() waits for this to drain so no call-out races teardown (and no + // posted lambda touches a destroyed Simulator). + size_t inFlightOperations = 0; std::map> connectors; std::map> associations; @@ -167,74 +186,93 @@ private: this->mCondition.notify_all(); } - // Called with the lock held; unlocks around call-outs. - void executeConnectOperation( - const Operation& operation, std::unique_lock& lock) { - const auto targetNodeId = Identifiers::getNodeIdFromPeerDescriptor( - operation.targetDescriptor); - const auto connectorIterator = this->connectors.find(targetNodeId); - if (connectorIterator == this->connectors.end()) { + // Runs on the association's serial executor; takes the lock only for + // the shared-state reads and makes the call-outs without it. + void executeConnectOperation(const Operation& operation) { + std::shared_ptr connector; + std::shared_ptr sourceConnection; + std::function&)> errorCallback; + { + std::scoped_lock lock(this->mMutex); + if (this->stopped) { + return; + } + const auto targetNodeId = Identifiers::getNodeIdFromPeerDescriptor( + operation.targetDescriptor); + const auto connectorIterator = this->connectors.find(targetNodeId); + if (connectorIterator == this->connectors.end()) { + errorCallback = operation.association->connectedCallback; + } else { + connector = connectorIterator->second; + sourceConnection = operation.association->sourceConnection; + } + } + if (!connector) { SLogger::error( "Target connector not found when executing connect operation"); - const auto callback = operation.association->connectedCallback; - lock.unlock(); - if (callback) { - callback("Target connector not found"); + if (errorCallback) { + errorCallback("Target connector not found"); } - lock.lock(); return; } - const auto connector = connectorIterator->second; - const auto sourceConnection = operation.association->sourceConnection; - lock.unlock(); connector->handleIncomingConnection(sourceConnection); - lock.lock(); } - // Called with the lock held; unlocks around call-outs. Mirrors the - // TS close/ack sequence: the first CloseOperation disconnects the - // counterpart and schedules a CloseOperation back (the 'ack'), which - // finally removes both associations. - void executeCloseOperation( - const Operation& operation, std::unique_lock& lock) { - const auto target = operation.association->destinationConnection; - std::shared_ptr counterAssociation; - if (target) { - const auto counterIterator = this->associations.find(target.get()); - if (counterIterator != this->associations.end()) { - counterAssociation = counterIterator->second; + // Runs on the association's serial executor. Mirrors the TS close/ack + // sequence: the first CloseOperation disconnects the counterpart and + // schedules a CloseOperation back (the 'ack'), which finally removes + // both associations. + void executeCloseOperation(const Operation& operation) { + std::shared_ptr target; + { + std::scoped_lock lock(this->mMutex); + if (this->stopped) { + return; + } + target = operation.association->destinationConnection; + std::shared_ptr counterAssociation; + if (target) { + const auto counterIterator = + this->associations.find(target.get()); + if (counterIterator != this->associations.end()) { + counterAssociation = counterIterator->second; + } + } + if (!target || !counterAssociation) { + this->associations.erase( + operation.association->sourceConnection.get()); + return; + } + if (counterAssociation->closing) { + // this is the 'ack' of the CloseOperation to the original + // closer + this->associations.erase(target.get()); + this->associations.erase( + operation.association->sourceConnection.get()); + return; } } - if (!target || !counterAssociation) { - this->associations.erase( - operation.association->sourceConnection.get()); - } else if (!counterAssociation->closing) { - lock.unlock(); - target->handleIncomingDisconnection(); - this->close(*target); - lock.lock(); - } else { - // this is the 'ack' of the CloseOperation to the original - // closer - this->associations.erase(target.get()); - this->associations.erase( - operation.association->sourceConnection.get()); - } + target->handleIncomingDisconnection(); + this->close(*target); } - // Called with the lock held; unlocks around call-outs. - void executeSendOperation( - const Operation& operation, std::unique_lock& lock) { - const auto destination = operation.association->destinationConnection; + // Runs on the association's serial executor. + void executeSendOperation(const Operation& operation) { + std::shared_ptr destination; + { + std::scoped_lock lock(this->mMutex); + if (this->stopped) { + return; + } + destination = operation.association->destinationConnection; + } if (!destination) { SLogger::error( "send operation executed on an association with no" " destination, dropping the message"); return; } - lock.unlock(); destination->handleIncomingData(operation.data); - lock.lock(); } void dispatchLoop() { @@ -255,17 +293,33 @@ private: } const Operation operation = this->operationQueue.top(); this->operationQueue.pop(); - switch (operation.type) { - case OperationType::CONNECT: - this->executeConnectOperation(operation, lock); - break; - case OperationType::CLOSE: - this->executeCloseOperation(operation, lock); - break; - case OperationType::SEND: - this->executeSendOperation(operation, lock); - break; - } + // Hand the operation to its association's serial executor: + // per-association FIFO is preserved (this loop is the only + // poster and pops in deadline order), while different + // associations deliver concurrently on the worker pool. + const auto executor = operation.association->executor; + this->inFlightOperations++; + lock.unlock(); + executor->add([this, operation]() { + switch (operation.type) { + case OperationType::CONNECT: + this->executeConnectOperation(operation); + break; + case OperationType::CLOSE: + this->executeCloseOperation(operation); + break; + case OperationType::SEND: + this->executeSendOperation(operation); + break; + } + // Notify under the lock: once stop()'s drain-wait sees the + // count hit zero the Simulator may be destroyed, so this + // lambda must not touch members after releasing it. + std::scoped_lock finishLock(this->mMutex); + this->inFlightOperations--; + this->mCondition.notify_all(); + }); + lock.lock(); } } @@ -337,6 +391,9 @@ public: auto targetAssociation = std::make_shared(); targetAssociation->sourceConnection = targetConnection; targetAssociation->destinationConnection = sourceConnection; + targetAssociation->executor = + std::make_shared( + streamr::utils::SharedExecutors::worker()); this->associations.emplace( targetConnection.get(), std::move(targetAssociation)); callback = sourceIterator->second->connectedCallback; @@ -359,6 +416,9 @@ public: auto association = std::make_shared(); association->sourceConnection = sourceConnection; association->connectedCallback = std::move(connectedCallback); + association->executor = + std::make_shared( + streamr::utils::SharedExecutors::worker()); this->associations[sourceConnection.get()] = association; const auto executionTime = this->generateExecutionTime( @@ -451,6 +511,15 @@ public: std::this_thread::get_id() != this->dispatcherThread.get_id()) { this->dispatcherThread.join(); } + // Drain the operations already handed to association executors — + // their lambdas skip the call-out once `stopped` is set, so this + // completes promptly, and afterwards nothing references this + // Simulator (safe to destroy). + { + std::unique_lock lock(this->mMutex); + this->mCondition.wait( + lock, [this]() { return this->inFlightOperations == 0; }); + } } }; diff --git a/packages/streamr-dht/modules/dht/routing/RoutingSession.cppm b/packages/streamr-dht/modules/dht/routing/RoutingSession.cppm index 6dfb1358..0ed95605 100644 --- a/packages/streamr-dht/modules/dht/routing/RoutingSession.cppm +++ b/packages/streamr-dht/modules/dht/routing/RoutingSession.cppm @@ -280,13 +280,14 @@ public: const DhtAddress targetId = Identifiers::getDhtAddressFromRaw( DhtAddressRaw{this->options.routedMessage.target()}); - RoutingTable routingTable; - if (this->options.routingTablesCache.has(targetId, previousId) && - this->options.routingTablesCache.get(targetId, previousId) - ->getSize() > 0) { - routingTable = - this->options.routingTablesCache.get(targetId, previousId); - } else { + // Fetch the cached table ONCE: every lookup re-evaluates the cache's + // 15s TTL, so a second get() microseconds after a successful has()/ + // get() can expire the entry and return null — dereferencing that + // null table SEGV'd here (observed crash: updateAndGetRoutablePeers + // → getClosestContacts on a null RoutingTable). + RoutingTable routingTable = + this->options.routingTablesCache.get(targetId, previousId); + if (routingTable == nullptr || routingTable->getSize() == 0) { routingTable = std::make_shared>( SortedContactListOptions{ diff --git a/packages/streamr-dht/test/integration/Layer1ScaleTest.cpp b/packages/streamr-dht/test/integration/Layer1ScaleTest.cpp index 3157a61f..f9f6e6e1 100644 --- a/packages/streamr-dht/test/integration/Layer1ScaleTest.cpp +++ b/packages/streamr-dht/test/integration/Layer1ScaleTest.cpp @@ -38,18 +38,6 @@ namespace { constexpr size_t nodeCount = 48; -std::vector toNodeIds( - const std::vector& descriptors) { - std::vector ids; - ids.reserve(descriptors.size()); - for (const auto& descriptor : descriptors) { - ids.push_back( - static_cast( - Identifiers::getNodeIdFromPeerDescriptor(descriptor))); - } - return ids; -} - // Promise.all(nodes.map((node) => node.joinDht([entryPoint]))) void joinAllInParallel( const std::vector>& nodes, @@ -118,14 +106,17 @@ TEST_F(Layer1ScaleTest, SingleLayer1Dht) { const auto& layer0Node = this->nodes[i]->node; const auto& layer1Node = layer1Nodes[i]->node; EXPECT_EQ(layer1Node->getNodeId(), layer0Node->getNodeId()); + // TS compares the two views' connection counts and connection + // lists; both are reads of the SAME live view object (the layer-1 + // node is constructed with the layer-0 node's ConnectionsView), so + // in single-threaded TS the equalities hold trivially. In C++ the + // background connection churn (e.g. trailing k-bucket pings) can + // mutate the view between two reads, so assert the identity that + // makes the TS equalities true instead of racing two snapshots. EXPECT_EQ( - layer1Node->getConnectionsView()->getConnectionCount(), - layer0Node->getConnectionsView()->getConnectionCount()); + layer1Node->getConnectionsView(), layer0Node->getConnectionsView()); EXPECT_GE( layer1Node->getNeighborCount(), numberOfNodesPerKBucketDefault / 2); - EXPECT_EQ( - toNodeIds(layer1Node->getConnectionsView()->getConnections()), - toNodeIds(layer0Node->getConnectionsView()->getConnections())); } } @@ -155,11 +146,12 @@ TEST_F(Layer1ScaleTest, MultipleLayer1Dht) { for (size_t i = 0; i < nodeCount; i++) { const auto& layer0Node = this->nodes[i]->node; for (size_t s = 0; s < serviceIds.size(); s++) { + // Same live-view identity as in SingleLayer1Dht: TS compares + // counts of the same object; two C++ reads would race the + // background connection churn. EXPECT_EQ( - layer0Node->getConnectionsView()->getConnectionCount(), - streams[s][i] - ->node->getConnectionsView() - ->getConnectionCount()); + layer0Node->getConnectionsView(), + streams[s][i]->node->getConnectionsView()); } } }