dabla opened a new pull request, #76545: URL: https://github.com/apache/airflow/pull/76545
`CommsDecoder.asend()` holds the supervisor channel's thread lock while it awaits the event loop for its socket I/O. That has three consequences on main: - a sync `send()` on that loop's thread can never get the lock, so it raises `DeadlockImminentError` (#73521 made it raise instead of freeze when the loop is paused, but it still cannot wait); - `asend()`'s `asyncio.Lock` binds to the first loop that waits on it, so `asyncio.run()` on a worker thread, or any later loop in the same process, fails with "bound to a different event loop" while another loop has calls in flight; - a cancelled `asend()` releases the lock before its response is read, and the next request reads that response as its own. This change runs the whole request of `asend()` (write and read) on a worker thread that needs no loop, behind one queue per event loop (`_LoopQueue`/`_LoopQueues`, held weakly so a closed loop is not kept). `send()` waits for the reentrant channel lock; only a `send()` on a thread that is itself in the middle of a request (a signal handler such as the one running `on_kill`) still raises. A cancelled `asend()` marks its request abandoned: not sent if it had not been yet, and once sent its response is always drained. `TriggerCommsDecoder` overrides `send`/`asend` and is not affected. Benchmarks (CodSpeed simulation, callgrind instructions, 2 rounds each, `main` → this PR): | Benchmark | main | this PR | |---|---|---| | `asend` round trip | 599 201 / 600 637 | 450 867 / 451 749 (−25 %) | | 20 concurrent `asend`s | 28.05 M / 28.14 M | 18.87 M / 18.92 M (−33 %) | | `send` round trip | 117 631 / 117 961 | 114 378 / 115 471 (−2 %) | Wall clock, 50 `asend`s in flight against a supervisor answering in 2 ms: `asyncio.to_thread` work on the same loop waits 0.24–0.32 ms (median), at most 0.70 ms. On main the same script fails with "Lock ... is bound to a different event loop". Extracted from #62922 (AIP-104), which carries the same change and will drop it from its diff once this merges. --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Claude Opus 5.5 Generated-by: Claude Opus 5.5 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
