dabla commented on code in PR #62922:
URL: https://github.com/apache/airflow/pull/62922#discussion_r4249272725
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -289,59 +289,68 @@ def send(self, msg: SendMsgType) -> ReceiveMsgType | None:
if not self._thread_lock.acquire(blocking=not self._is_on_loop_thread):
Review Comment:
You're right, and checking it showed that one deadlock is left, so the guard
became precise instead of going away (be426aad12).
Since `asend()` holds the channel only on a worker thread that writes, reads
and releases without the loop, a `send()` that waits for it always gets the
channel back after one round trip. The one wait that cannot end is a `send()`
on a thread that is itself in the middle of a request: the runner's SIGTERM
handler runs `on_kill()` on the main thread, so an `on_kill()` making a sync
SDK call while the main thread waits for a response would wait for its own
thread. The channel now records the thread exchanging a request (set and
cleared inside `_exchange`, which `send()` and `asend()`'s worker thread both
run), and only that thread is refused:
| Situation | Before | Now |
|---|---|---|
| sync `send()` on a running loop while an `asend()` is in flight |
`DeadlockImminentError`, the task fails without retry | waits one round trip,
gets its own response |
| the same on a loop paused between two `run_until_complete()` calls |
`DeadlockImminentError` | waits, gets its own response |
| `send()` on a thread that is itself mid-request (a signal handler, e.g.
`on_kill`) | raised only for async tasks, sync tasks hung | always
`DeadlockImminentError`, at once |
`_is_on_loop_thread`, `_asend_in_flight_on_this_thread` and
`_loop_thread_id` are gone, and the error message names the case that is left.
In `test_comms`:
- running and paused loop with an `asend()` parked on its response: each
caller gets its own response (both fail with the old error on the previous
code);
- re-entry, simulated and with a real `SIGALRM` handler calling `send()`
mid-request: raises at once and the interrupted request still completes; with
the owner check removed both tests hang, so they do catch the deadlock;
- a request that fails mid-exchange leaves no owner behind;
- 20 threads calling `send()` and 20 coroutines calling `asend()` at once:
every caller gets its own response.
`_fail_fast_for` no longer has its own branch for `DeadlockImminentError`;
the generic one for a `BaseException` that is no `Exception` still fails the
task at once. The docs page, the class docstring, the executor docstrings and
`connection.py` now say a sync call on the loop works but holds up every
iteration for its round trip, so async calls are preferred; the sync-to-async
table stays. A local round-trip benchmark (old vs new `comms.py`, median of 5
rounds) shows no change: `send` 151.5 vs 152.7 µs, `asend` 353.2 vs 343.1 µs,
50 concurrent `asend` 17.38 vs 17.55 ms.
Drafted-by: Claude Opus 5.5; reviewed by @dabla before posting
--
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]
