Make RecursiveOperationManager::sendResponse fully asynchronous (remove the last worker-pool park)#83
Merged
Merged
Conversation
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<void> 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 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Follow-up to #81. That PR fixed the permanent worker-pool deadlock in
RecursiveOperationManager::sendResponse(destroying the per-sessionListeningRpcCommunicatoron a pool thread parked the thread in the destructor's scope join, waiting for a send task that itself needed a pool thread) by retiring communicators into a pruned list. One bounded park remained:RecursiveOperationSessionRpcRemote::sendResponsewrapped the generated client notify inblockingWait(), so under pool saturation an RPC handler's shared worker thread was parked for up tosessionRpcTimeout(10 s). This PR removes that park: the response send never blocks any thread anywhere.Design
RecursiveOperationSessionRpcRemote::sendResponseis now afolly::coro::Task<void>coroutine thatco_awaits the generated client notify in-frame (per the repo's lazy-task rule: the request locals live in the awaiting frame) and swallows failures — TS fire-and-forget parity, still bounded bysessionRpcTimeout.RecursiveOperationManager::sendResponseruns the whole send as a detached coroutine on a new manager-ownedGuardedAsyncScope(sendResponseScope, onSharedExecutors::worker()). The task constructs the per-sessionListeningRpcCommunicatorandRpcRemote,co_awaits the notify (suspends instead of parking), then disposes of the communicator.pendingAsyncTaskCount() == 0(an empty-scope join needs no pool thread, and a zero count also proves the coroutine is not inside one of the communicator's tasks;destroy()detaches the transport listeners first, so the zero observation is stable). Otherwise the communicator goes into the existingretiredSessionCommunicatorsbackstop from Fix Layer0 e2e teardown SIGSEGVs: drain RPC scopes before their targets die, roll back started on failed start() #81, pruned exactly as before.sendResponseracingstop()is dropped by the scope'sclose()gate before anything blocking-to-destroy exists: the dropped lambda releases only vectors and protobuf messages at theadd()call site. A dropped response after stop is correct for a fire-and-forget send.stop()closessendResponseScope(owner's thread;DhtNode::stop()calls it while the session transport is still alive) before clearing the retired list, because draining tasks may still retire communicators into it; member declaration order gives the same ordering in the destructor path.Verification (macOS arm64, Debug)
Pool size forced with a temporary
STREAMR_TEST_WORKER_THREADSoverride inSharedExecutors::worker()(removed before commit);DhtNodeExternalApiTest.ExternalStoreDataHappyPathwith a 90 s watchdog per run:Before this change the 1-thread configuration failed after ~10 s per run (the
blockingWaitpark swallowed the only worker until the notify timed out). No hangs in any configuration.Full suites:
streamr-dht-test-unit181/181,streamr-proto-rpc-test-unit24/24,streamr-proto-rpc-test-integration4/4,streamr-dht-test-end-to-end7/7 four times (three consecutive runs plus once on the final clean rebuild),streamr-dht-test-integration42/43 — the one failure wasLayer1ScaleTest.SingleLayer1Dht, the known load-sensitive flake (#80): it exercises onlyjoinDht, which never touches the recursive-operation path, and it passed when rerun in isolation.Test updates: the manager unit test's
transport.sendCountassertions moved aftermanager->stop()(which drains the send scope, making the observation deterministic now that the send is detached), and the session unit testblockingWaits the now-coroutinesendResponseon the test thread.🤖 Generated with Claude Code