dabla opened a new pull request, #68377:
URL: https://github.com/apache/airflow/pull/68377
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull
request.
Write a good git commit message following this guide:
http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the
documentation.
For user-facing UI changes, please attach before/after screenshots (or a
short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for
a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we
sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: [#68375](https://github.com/apache/airflow/pull/68375)
-->
# Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on
event loop
## Summary
Fixes a permanent deadlock that caused async tasks to hang indefinitely when
multiple
mapped task instances attempted to fetch a connection concurrently (e.g. via
`KiotaRequestAdapterHook.get_async_conn` in an MSGraph DAG using
`IterableOperator`).
## Root cause
`CommsDecoder.asend()` used two separate steps to communicate with the
supervisor process:
1. Acquire `_thread_lock` via `loop.run_in_executor()` (a thread-pool hop)
2. Perform the blocking socket read via `asyncio.to_thread(_read_frame)` —
**while still
holding `_thread_lock`**
Between steps 1 and 2 the event loop was free to schedule other coroutines.
If one of
those coroutines called `BaseHook.get_connection()` (the **sync** path),
that call
reached `CommsDecoder.send()`, which tried to acquire `_thread_lock`
**directly on the
event-loop thread**. Because `_thread_lock` was already held (by the
thread-pool thread
from step 1), the event-loop thread blocked — permanently. The
`asyncio.to_thread` future
from step 2 could never be resolved (the event loop was frozen), so
`asend()`'s `finally`
block never ran and the lock was never released. Classic deadlock.
```
Event-loop thread Thread-pool thread
────────────────── ──────────────────
asend(): run_in_executor ← acquires _thread_lock ✓
await to_thread(recv) ← yields; loop free to run other tasks
[other coroutine] send() → tries _thread_lock.acquire() on loop thread
→ BLOCKS loop thread ✗
to_thread future never resolves (loop is frozen)
_thread_lock never released → permanent deadlock
```
This only manifested with hooks that acquire a connection on every call (like
`KiotaRequestAdapterHook` when its cache is cold) and **not** with hooks
that use
native asyncio primitives (like `SFTPAsyncHook`), because asyncio locks
always
`await` and never block the event loop.
## Fix
Consolidate both the socket write **and** the blocking read into a **single**
`asyncio.to_thread()` call. `_thread_lock` is now acquired and released
entirely
within a thread-pool thread, so it is never held by the event-loop thread
across
an `await` boundary.
```python
# Before (deadlock-prone)
await loop.run_in_executor(None, self._thread_lock.acquire) # held by loop
thread…
try:
await loop.sock_sendall(self.socket, frame_bytes) # …across
this await…
frame = await asyncio.to_thread(self._read_frame) # …and this
one
finally:
self._thread_lock.release()
# After (safe)
def _send_and_receive() -> _ResponseFrame:
with self._thread_lock: # held only inside a thread-pool thread
self.socket.sendall(frame_bytes)
return self._read_frame()
frame = await asyncio.to_thread(_send_and_receive)
```
The `_async_lock` (`asyncio.Lock`) is kept to ensure only one `asend()`
coroutine
is in-flight at a time (coroutine-level mutual exclusion).
## Testing
Added a regression test
`test_asend_does_not_deadlock_when_sync_send_called_concurrently`
that:
1. Starts an `asend()` coroutine and waits until `_thread_lock` is held in
the thread pool.
2. Immediately calls the synchronous `send()` from the event-loop thread.
3. Asserts both calls complete successfully within a tight timeout —
previously this
would hang forever.
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR,
please
change below checkbox to `[X]` followed by the name of the tool, uncomment
the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request
Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)**
for more information. Note: commit author/co-author name and email in commits
become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal
([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals))
is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License
Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment:
`{pr_number}.significant.rst`, in
[airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
You can add this file in a follow-up commit after the PR is created so you
know the PR number.
--
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]
