You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dabla (via GitHub)" <gi...@apache.org> on 2026/06/11 06:45:28 UTC
[PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4680879917
> Tested this solution and it doesn't solve the deadlock.
If that doesn't solve the deadlock then I'm not sure that the problem is identified correctly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added aget_hook method on BaseHook and deadlock detection in CommsDecoder [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4698578355
> > > I wonder if we could also catch the deadlock in this case (`_threading_lock.locked() and threading.get_ident() == self._loop_thread_id` style) to throw a nice error message, pointing the user to update their hook?
> >
> >
> > ```
> > [2026-06-12 07:56:11] ERROR - comms.DeadlockImminentError: comms.send() called from the event loop thread for message 'TaskState' — deadlock will occur as soon as another coroutine calls asend(). Likely cause: BaseHook.get_hook() or BaseHook.get_connection() was called from inside an async task. Use the async equivalents instead: await BaseHook.aget_hook() or await BaseHook.aget_connection().
> > [2026-06-12 07:56:11] ERROR - Offending call stack:
> > ...
> > ```
>
> And there's basically 0 chance of this affecting anyone today, as this is only a surfaces when you use the AsyncPythonOperator, right? (Just thinking about if this will suddenly start failing loads of existing workloads that were working before due to dumb luck)
I did exactly the same reflection myself yesterday, but I monkey patched this check to all our environments (even production) to see what comes out of it and until now now exception where raised due to this check (+1200 DAG's), and we use a lot the async PythonOperator with iterate.
A possible solution would be to add a config parameter for this check which is disabled by default for example.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3460392928
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -118,6 +122,39 @@
ReceiveMsgType = TypeVar("ReceiveMsgType", bound=BaseModel)
+class DeadlockImminentError(BaseException):
+ """
+ Raised when ``CommsDecoder.send()`` is called from the event loop thread.
+
+ Inherits from :class:`BaseException` rather than :class:`Exception` so that
+ ``contextlib.suppress(Exception)`` — used in :func:`airflow.sdk.log.mask_secret`
+ and similar helpers — does **not** catch it. This ensures the error always
+ surfaces to the caller, pinpointing the incorrect use of a sync API from an
+ async context (e.g. ``BaseHook.get_hook()`` inside ``aexecute()``).
+
+ Fix: replace ``BaseHook.get_hook()`` with ``await BaseHook.aget_hook()``.
+ """
+
+ def __init__(self, msg: object, *, deadlock_imminent: bool) -> None:
+ import traceback
+
+ detail = (
+ "deadlock is imminent (asend() is concurrently in-flight)"
+ if deadlock_imminent
+ else "deadlock will occur as soon as another coroutine calls asend()"
+ )
+ stack = "".join(traceback.format_stack())
Review Comment:
Yes — it’s a different stack, and in async deadlock/debug scenarios it’s often the more useful one, because it tells you where the illegal call originated, not just where things blew up.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4681082144
@dabla Thanks for digging in more. I'm just very sensitive to adding or changing locks, as I've worked quite hard to fix deadlocks on trigger (by removing locks!)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3490449672
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -118,6 +122,39 @@
ReceiveMsgType = TypeVar("ReceiveMsgType", bound=BaseModel)
+class DeadlockImminentError(BaseException):
+ """
+ Raised when ``CommsDecoder.send()`` is called from the event loop thread.
+
+ Inherits from :class:`BaseException` rather than :class:`Exception` so that
+ ``contextlib.suppress(Exception)`` — used in :func:`airflow.sdk.log.mask_secret`
+ and similar helpers — does **not** catch it. This ensures the error always
+ surfaces to the caller, pinpointing the incorrect use of a sync API from an
+ async context (e.g. ``BaseHook.get_hook()`` inside ``aexecute()``).
+
+ Fix: replace ``BaseHook.get_hook()`` with ``await BaseHook.aget_hook()``.
+ """
+
+ def __init__(self, msg: object, *, deadlock_imminent: bool) -> None:
+ import traceback
+
+ detail = (
+ "deadlock is imminent (asend() is concurrently in-flight)"
+ if deadlock_imminent
+ else "deadlock will occur as soon as another coroutine calls asend()"
Review Comment:
Do we intend to use this anywhere? From what I can tell, all calls to this class uses `deadlock_imminent=True`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3458600169
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -118,6 +122,39 @@
ReceiveMsgType = TypeVar("ReceiveMsgType", bound=BaseModel)
+class DeadlockImminentError(BaseException):
+ """
+ Raised when ``CommsDecoder.send()`` is called from the event loop thread.
+
+ Inherits from :class:`BaseException` rather than :class:`Exception` so that
+ ``contextlib.suppress(Exception)`` — used in :func:`airflow.sdk.log.mask_secret`
+ and similar helpers — does **not** catch it. This ensures the error always
+ surfaces to the caller, pinpointing the incorrect use of a sync API from an
+ async context (e.g. ``BaseHook.get_hook()`` inside ``aexecute()``).
+
+ Fix: replace ``BaseHook.get_hook()`` with ``await BaseHook.aget_hook()``.
+ """
+
+ def __init__(self, msg: object, *, deadlock_imminent: bool) -> None:
+ import traceback
+
+ detail = (
+ "deadlock is imminent (asend() is concurrently in-flight)"
+ if deadlock_imminent
+ else "deadlock will occur as soon as another coroutine calls asend()"
+ )
+ stack = "".join(traceback.format_stack())
Review Comment:
Is the traceback needed in the exception? There’s already a traceback when the exception is raised. Is this a different stack?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3509781386
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -208,19 +249,51 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
_thread_lock: threading.Lock = attrs.field(factory=threading.Lock, repr=False)
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ # Thread ID of the event loop thread, recorded on the first asend() call.
+ # Used by send() to detect an imminent deadlock.
Review Comment:
Also extremely obvious with a quick read, with the danger of being stale if the implementation gets tweaked. Suggest removal.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4681179634
I wonder if we could also catch the deadlock in this case (`_threading_lock.locked() and threading.get_ident() == self._loop_thread_id` style) to throw a nice error message, pointing the user to update their hook?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added aget_hook method on BaseHook and deadlock detection [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4683274335
> I wonder if we could also catch the deadlock in this case (`_threading_lock.locked() and threading.get_ident() == self._loop_thread_id` style) to throw a nice error message, pointing the user to update their hook?
Detection mechanism added
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4680165209
Claude propose to change mask_secret to something like this, but dunno if this is a good idea:
```
def mask_secret(secret: JsonValue, name: str | None = None) -> None:
"""
Mask a secret in both task process and supervisor process.
For secrets loaded from backends (Vault, env vars, etc.), this ensures
they're masked in both the task subprocess AND supervisor's log output.
Works safely in both sync and async contexts.
When called from an async context (event loop running), the supervisor
notification is scheduled as a fire-and-forget coroutine via
:func:`asyncio.ensure_future` so the event loop is never blocked.
"""
import asyncio
from contextlib import suppress
from airflow.sdk._shared.secrets_masker import _secrets_masker
_secrets_masker().add_mask(secret, name)
with suppress(Exception):
# Try to tell supervisor (only if in task execution context)
from airflow.sdk.execution_time import task_runner
from airflow.sdk.execution_time.comms import MaskSecret
if comms := getattr(task_runner, "SUPERVISOR_COMMS", None):
msg = MaskSecret(value=secret, name=name)
try:
asyncio.get_running_loop()
# We're on the event loop thread — must not call blocking send() here.
# Schedule asend() as a fire-and-forget task; MaskSecret needs no response.
asyncio.ensure_future(comms.asend(msg))
except RuntimeError:
# No running event loop — safe to call sync send().
comms.send(msg)
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4681074475
Oh greenback is currently only in the triggerer.
I think the AsyncPython operator might need to do the same.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added aget_hook method on BaseHook to avoid deadlock in async tasks [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4681893608
> I wonder if we could also catch the deadlock in this case (`_threading_lock.locked() and threading.get_ident() == self._loop_thread_id` style) to throw a nice error message, pointing the user to update their hook?
Will try to add the check as well, good idea.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4680550310
> Claude propose to change mask_secret to something like this, but dunno if this is a good idea:
>
> ```
> def mask_secret(secret: JsonValue, name: str | None = None) -> None:
> """
> Mask a secret in both task process and supervisor process.
>
> For secrets loaded from backends (Vault, env vars, etc.), this ensures
> they're masked in both the task subprocess AND supervisor's log output.
> Works safely in both sync and async contexts.
>
> When called from an async context (event loop running), the supervisor
> notification is scheduled as a fire-and-forget coroutine via
> :func:`asyncio.ensure_future` so the event loop is never blocked.
> """
> import asyncio
> from contextlib import suppress
>
> from airflow.sdk._shared.secrets_masker import _secrets_masker
>
> _secrets_masker().add_mask(secret, name)
>
> with suppress(Exception):
> # Try to tell supervisor (only if in task execution context)
> from airflow.sdk.execution_time import task_runner
> from airflow.sdk.execution_time.comms import MaskSecret
>
> if comms := getattr(task_runner, "SUPERVISOR_COMMS", None):
> msg = MaskSecret(value=secret, name=name)
> try:
> asyncio.get_running_loop()
> # We're on the event loop thread — must not call blocking send() here.
> # Schedule asend() as a fire-and-forget task; MaskSecret needs no response.
> asyncio.ensure_future(comms.asend(msg))
> except RuntimeError:
> # No running event loop — safe to call sync send().
> comms.send(msg)
> ```
>
> This means detecting if mask_secret was called from an async context or not.
Tested this solution and it doesn't solve the deadlock.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4681048098
I think I found the issue:
```
@task(show_return_value_in_logs=False, task_concurrency=5)
async def get_user_registered_devices(user_id):
from airflow.exceptions import AirflowNotFoundException
from contextlib import suppress
with suppress(AirflowNotFoundException):
hook = KiotaRequestAdapterHook.get_hook(conn_id=MSGRAPH_CONN_ID)
results = await hook.paginated_run(
url=f"users/{user_id}/registeredDevices",
query_parameters={"$select": "id"},
)
return results
return []
```
We should implement an async get_hook in BaseHook, hence this is why it's causing deadlock, as the get_hook method underneath calls the sync get_connection method.
That was also the mean difference between my MSGraph case and SFTP one, hence why I didn't have the issue with SFTP example, because there I'm using the async SFTPClientPool:
```
@task(
retries=3,
retry_delay=timedelta(seconds=60),
show_return_value_in_logs=False,
)
async def load_xml_files(file, **context):
import logging
from io import BytesIO
from os import cpu_count
from airflow.providers.sftp.pools.sftp import SFTPClientPool
connection = get_connection(sftp_conn)
async with SFTPClientPool(
sftp_conn_id=sftp_conn, pool_size=cpu_count()
).get_sftp_client() as sftp:
logging.info("downloading: %s", file)
buffer = BytesIO()
async with sftp.open(file, encoding=xml_encoding) as remote_file:
data = await remote_file.read()
buffer.write(data.encode(xml_encoding))
buffer.seek(0)
# Convert and return rows for DB insert
return convert_to_rows(
connection,
file,
convert_xml_to_json(
buffer,
xml_namespaces,
xml_encoding,
xml_force_list,
),
)
```
So it's good that you challenged the issue!
Also I can confirm that the latest version of CommDecoder send method in 3.3 (from main) doesn't deadlock anymore with the example of KiotaRequestAdapterHook.get_hook.
I tested with that version through monkey patching on our 3.2.2. Probably something has changed in the meanwhile which solved the deadlock that not present in 3.2.2.
Still I think it's a good thing to add the async aget_hook method on BaseHook.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4681161301
I think adding the `aget_hook` method on `BaseHook` is the only valid and solid solution.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added aget_hook method on BaseHook and deadlock detection in CommsDecoder [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4693114130
> > I wonder if we could also catch the deadlock in this case (`_threading_lock.locked() and threading.get_ident() == self._loop_thread_id` style) to throw a nice error message, pointing the user to update their hook?
>
> ```
> [2026-06-12 07:56:11] ERROR - comms.DeadlockImminentError: comms.send() called from the event loop thread for message 'TaskState' — deadlock will occur as soon as another coroutine calls asend(). Likely cause: BaseHook.get_hook() or BaseHook.get_connection() was called from inside an async task. Use the async equivalents instead: await BaseHook.aget_hook() or await BaseHook.aget_connection().
> [2026-06-12 07:56:11] ERROR - Offending call stack:
> ...
> ```
And there's basically 0 chance of this affecting anyone today, as this is only a surfaces when you use the AsyncPythonOperator, right? (Just thinking about if this will suddenly start failing loads of existing workloads that were working before due to dumb luck)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3509783084
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -118,6 +123,42 @@
ReceiveMsgType = TypeVar("ReceiveMsgType", bound=BaseModel)
+class DeadlockImminentError(BaseException):
+ """
+ Raised when ``CommsDecoder.send()`` is called from the event loop thread while
+ ``_thread_lock`` is already held by a concurrent ``asend()`` round-trip.
+
+ The sync ``send()`` attempts a non-blocking lock acquire when it detects it is
+ running on the event loop thread. If the acquire fails — meaning an ``asend()``
+ coroutine currently owns the lock and is waiting for the event loop to complete
+ its I/O — a true deadlock would result, so this error is raised instead.
+
+ Inherits from :class:`BaseException` rather than :class:`Exception` so that
+ ``contextlib.suppress(Exception)`` — used in :func:`airflow.sdk.log.mask_secret`
+ and similar helpers — does **not** catch it. This ensures the error always
+ surfaces to the caller, pinpointing the incorrect use of a sync API from an
+ async context (e.g. ``BaseHook.get_hook()`` inside ``aexecute()``).
+
+ Fix: replace ``BaseHook.get_hook()`` with ``await BaseHook.aget_hook()``.
Review Comment:
Please review these strings… “Fix” makes no sense here once this PR is merged.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3490437513
##########
task-sdk/tests/task_sdk/execution_time/test_comms.py:
##########
@@ -207,4 +208,142 @@ def _recv_exactly(sock, n):
for idx in range(num_threads):
assert errors[idx] is None, f"Thread {idx} error: {errors[idx]}"
assert results[idx].key == f"key{idx}", f"Out-of-order or missing response for thread {idx}"
- assert results[idx].value == f"value{idx}", f"Incorrect value for thread {idx}"
Review Comment:
This deletion seems accidental?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3490484967
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -118,6 +122,39 @@
ReceiveMsgType = TypeVar("ReceiveMsgType", bound=BaseModel)
+class DeadlockImminentError(BaseException):
+ """
+ Raised when ``CommsDecoder.send()`` is called from the event loop thread.
+
+ Inherits from :class:`BaseException` rather than :class:`Exception` so that
+ ``contextlib.suppress(Exception)`` — used in :func:`airflow.sdk.log.mask_secret`
+ and similar helpers — does **not** catch it. This ensures the error always
+ surfaces to the caller, pinpointing the incorrect use of a sync API from an
+ async context (e.g. ``BaseHook.get_hook()`` inside ``aexecute()``).
+
+ Fix: replace ``BaseHook.get_hook()`` with ``await BaseHook.aget_hook()``.
+ """
+
+ def __init__(self, msg: object, *, deadlock_imminent: bool) -> None:
+ import traceback
Review Comment:
This should be imported at the top. It is cheap.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3490454230
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -118,6 +122,39 @@
ReceiveMsgType = TypeVar("ReceiveMsgType", bound=BaseModel)
+class DeadlockImminentError(BaseException):
+ """
+ Raised when ``CommsDecoder.send()`` is called from the event loop thread.
+
+ Inherits from :class:`BaseException` rather than :class:`Exception` so that
+ ``contextlib.suppress(Exception)`` — used in :func:`airflow.sdk.log.mask_secret`
+ and similar helpers — does **not** catch it. This ensures the error always
+ surfaces to the caller, pinpointing the incorrect use of a sync API from an
+ async context (e.g. ``BaseHook.get_hook()`` inside ``aexecute()``).
+
+ Fix: replace ``BaseHook.get_hook()`` with ``await BaseHook.aget_hook()``.
+ """
+
+ def __init__(self, msg: object, *, deadlock_imminent: bool) -> None:
+ import traceback
+
+ detail = (
+ "deadlock is imminent (asend() is concurrently in-flight)"
+ if deadlock_imminent
+ else "deadlock will occur as soon as another coroutine calls asend()"
+ )
+ stack = "".join(traceback.format_stack())
+ super().__init__(
+ f"comms.send() called from the event loop thread for message '{type(msg).__name__}' "
+ f"— {detail}. "
+ "Likely cause: BaseHook.get_hook() or BaseHook.get_connection() was called "
+ "from inside an async task. "
+ "Use the async equivalents instead: "
+ "await BaseHook.aget_hook() or await BaseHook.aget_connection()."
+ f"\nOffending call stack:\n{stack}"
+ )
Review Comment:
It’s probably cleaner to put the variables as attributes on the class, and implement this string in `__str__` instead.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3490475328
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -118,6 +122,39 @@
ReceiveMsgType = TypeVar("ReceiveMsgType", bound=BaseModel)
+class DeadlockImminentError(BaseException):
+ """
+ Raised when ``CommsDecoder.send()`` is called from the event loop thread.
Review Comment:
This should probably mention it only happens when a lock is _also_ being acquired? The error is not actually always raised when `send` is called.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3509779516
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -208,19 +249,51 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
_thread_lock: threading.Lock = attrs.field(factory=threading.Lock, repr=False)
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ # Thread ID of the event loop thread, recorded on the first asend() call.
+ # Used by send() to detect an imminent deadlock.
+ _loop_thread_id: int | None = attrs.field(default=None, repr=False, init=False)
def _make_frame(self, msg: SendMsgType) -> _RequestFrame:
carrier: dict[str, str] = {}
_trace_propagator.inject(carrier)
return _RequestFrame(id=next(self.id_counter), body=msg.model_dump(), context_carrier=carrier or None)
+ # A deadlock can only occur when the event loop is *currently running* in
+ # this thread. After AsyncOperator.execute() / loop.run_until_complete()
+ # returns, the loop is no longer running; subsequent sync send() calls are
+ # safe. asyncio.get_running_loop() raises RuntimeError when there is no
+ # active loop in the calling thread, so we use it to gate the check.
Review Comment:
Same…
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4679820927
I'm not sold on this fix. sync send on the event loop in an async should go via the existing greenlet trampoline etc to go back to the event loop, no?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added aget_hook method on BaseHook and deadlock detection [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4687930225
> I wonder if we could also catch the deadlock in this case (`_threading_lock.locked() and threading.get_ident() == self._loop_thread_id` style) to throw a nice error message, pointing the user to update their hook?
```
[2026-06-12 07:56:11] ERROR - comms.DeadlockImminentError: comms.send() called from the event loop thread for message 'TaskState' — deadlock will occur as soon as another coroutine calls asend(). Likely cause: BaseHook.get_hook() or BaseHook.get_connection() was called from inside an async task. Use the async equivalents instead: await BaseHook.aget_hook() or await BaseHook.aget_connection().
[2026-06-12 07:56:11] ERROR - Offending call stack:
...
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3412153540
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -216,15 +255,64 @@ def _make_frame(self, msg: SendMsgType) -> _RequestFrame:
def send(self, msg: SendMsgType) -> ReceiveMsgType | None:
"""Send a request to the parent and block until the response is received."""
+ # Two-level detection for sync send() called from the event loop thread.
+ # Raises SyncSendOnEventLoopError (BaseException subclass) so it escapes
+ # contextlib.suppress(Exception) in mask_secret() and other helpers.
+ #
+ # Level 1 — wrong pattern (broader): send() is on the event loop thread at
+ # all, detected via the stored loop thread id from the last asend() call.
+ # Level 2 — imminent deadlock (precise): _thread_lock is also currently held
+ # by a thread-pool worker from a concurrent asend() round-trip.
+ #
+ # Typical cause: BaseHook.get_hook() or BaseHook.get_connection() called
+ # from inside an async task / aexecute().
+ # Fix: use 'await BaseHook.aget_hook()' or 'await BaseHook.aget_connection()'.
+ _on_loop_thread = self._loop_thread_id is not None and threading.get_ident() == self._loop_thread_id
Review Comment:
```suggestion
_on_loop_thread = threading.get_ident() == self._loop_thread_id
```
I think this is enough? `get_ident()` can never return None so if `self._loop_thread_id` is None the part after `and` would never be True anyway.
Actually I think the entire added block can be simplified to
```python
on_loop_thread = False
if threading.get_ident() == self._loop_thread_id:
with contextlib.suppress(RuntimeError):
on_loop_thread = bool(asyncio.get_running_loop())
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3458588696
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -208,42 +245,80 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
_thread_lock: threading.Lock = attrs.field(factory=threading.Lock, repr=False)
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ # Thread ID of the event loop thread, recorded on the first asend() call.
+ # Used by send() to detect an imminent deadlock.
+ _loop_thread_id: int | None = attrs.field(default=None, repr=False, init=False)
def _make_frame(self, msg: SendMsgType) -> _RequestFrame:
carrier: dict[str, str] = {}
_trace_propagator.inject(carrier)
return _RequestFrame(id=next(self.id_counter), body=msg.model_dump(), context_carrier=carrier or None)
+ # A deadlock can only occur when the event loop is *currently running* in
+ # this thread. After AsyncOperator.execute() / loop.run_until_complete()
+ # returns, the loop is no longer running; subsequent sync send() calls are
+ # safe. asyncio.get_running_loop() raises RuntimeError when there is no
+ # active loop in the calling thread, so we use it to gate the check.
+ @property
+ def _is_on_loop_thread(self) -> bool:
+ if threading.get_ident() == self._loop_thread_id:
+ with suppress(RuntimeError):
+ return bool(asyncio.get_running_loop())
+ return False
+
def send(self, msg: SendMsgType) -> ReceiveMsgType | None:
"""Send a request to the parent and block until the response is received."""
+ # Two-level detection for sync send() called from the event loop thread.
+ # Raises DeadlockImminentError (BaseException subclass) so it escapes
+ # contextlib.suppress(Exception) in mask_secret() and other helpers.
+ #
+ # Level 1 — wrong pattern (broader): send() is on the event loop thread at
+ # all, detected via the stored loop thread id from the last asend() call.
+ # Level 2 — imminent deadlock (precise): _thread_lock is also currently held
+ # by a thread-pool worker from a concurrent asend() round-trip.
+ #
+ # Typical cause: BaseHook.get_hook() or BaseHook.get_connection() called
+ # from inside an async task / aexecute().
+ # Fix: use 'await BaseHook.aget_hook()' or 'await BaseHook.aget_connection()'.
frame_bytes = self._make_frame(msg).as_bytes()
# We must make sure sockets aren't intermixed between sync and async calls,
# thus we need a dual locking mechanism to ensure that.
- with self._thread_lock:
+ # When called from the event loop thread, use non-blocking acquire to detect
+ # an imminent deadlock: an asend() coroutine currently holds _thread_lock and
+ # is waiting for the event loop to complete its I/O — a true deadlock.
+ if not self._thread_lock.acquire(blocking=not self._is_on_loop_thread):
+ raise DeadlockImminentError(msg, deadlock_imminent=True)
+ try:
self.socket.sendall(frame_bytes)
if isinstance(msg, ResendLoggingFD):
- if recv_fds is None:
- return None
# We need special handling here! The server can't send us the fd number, as the number on the
# supervisor will be different to in this process, so we have to mutate the message ourselves here.
+ if recv_fds is None:
+ return None
Review Comment:
This does not need to change (I think it makes more sense to put it above the comment since it does not count as special handling)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3509580063
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -118,6 +122,39 @@
ReceiveMsgType = TypeVar("ReceiveMsgType", bound=BaseModel)
+class DeadlockImminentError(BaseException):
+ """
+ Raised when ``CommsDecoder.send()`` is called from the event loop thread.
+
+ Inherits from :class:`BaseException` rather than :class:`Exception` so that
+ ``contextlib.suppress(Exception)`` — used in :func:`airflow.sdk.log.mask_secret`
+ and similar helpers — does **not** catch it. This ensures the error always
+ surfaces to the caller, pinpointing the incorrect use of a sync API from an
+ async context (e.g. ``BaseHook.get_hook()`` inside ``aexecute()``).
+
+ Fix: replace ``BaseHook.get_hook()`` with ``await BaseHook.aget_hook()``.
+ """
+
+ def __init__(self, msg: object, *, deadlock_imminent: bool) -> None:
+ import traceback
+
+ detail = (
+ "deadlock is imminent (asend() is concurrently in-flight)"
+ if deadlock_imminent
+ else "deadlock will occur as soon as another coroutine calls asend()"
+ )
+ stack = "".join(traceback.format_stack())
+ super().__init__(
+ f"comms.send() called from the event loop thread for message '{type(msg).__name__}' "
+ f"— {detail}. "
+ "Likely cause: BaseHook.get_hook() or BaseHook.get_connection() was called "
+ "from inside an async task. "
+ "Use the async equivalents instead: "
+ "await BaseHook.aget_hook() or await BaseHook.aget_connection()."
+ f"\nOffending call stack:\n{stack}"
+ )
Review Comment:
Good point TP will change it you're right
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3509597888
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -118,6 +122,39 @@
ReceiveMsgType = TypeVar("ReceiveMsgType", bound=BaseModel)
+class DeadlockImminentError(BaseException):
+ """
+ Raised when ``CommsDecoder.send()`` is called from the event loop thread.
+
+ Inherits from :class:`BaseException` rather than :class:`Exception` so that
+ ``contextlib.suppress(Exception)`` — used in :func:`airflow.sdk.log.mask_secret`
+ and similar helpers — does **not** catch it. This ensures the error always
+ surfaces to the caller, pinpointing the incorrect use of a sync API from an
+ async context (e.g. ``BaseHook.get_hook()`` inside ``aexecute()``).
+
+ Fix: replace ``BaseHook.get_hook()`` with ``await BaseHook.aget_hook()``.
+ """
+
+ def __init__(self, msg: object, *, deadlock_imminent: bool) -> None:
+ import traceback
+
+ detail = (
+ "deadlock is imminent (asend() is concurrently in-flight)"
+ if deadlock_imminent
+ else "deadlock will occur as soon as another coroutine calls asend()"
Review Comment:
Correct, removed it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3509777418
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -208,19 +249,51 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
_thread_lock: threading.Lock = attrs.field(factory=threading.Lock, repr=False)
# Async lock for async operations
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
+ # Thread ID of the event loop thread, recorded on the first asend() call.
+ # Used by send() to detect an imminent deadlock.
+ _loop_thread_id: int | None = attrs.field(default=None, repr=False, init=False)
def _make_frame(self, msg: SendMsgType) -> _RequestFrame:
carrier: dict[str, str] = {}
_trace_propagator.inject(carrier)
return _RequestFrame(id=next(self.id_counter), body=msg.model_dump(), context_carrier=carrier or None)
+ # A deadlock can only occur when the event loop is *currently running* in
+ # this thread. After AsyncOperator.execute() / loop.run_until_complete()
+ # returns, the loop is no longer running; subsequent sync send() calls are
+ # safe. asyncio.get_running_loop() raises RuntimeError when there is no
+ # active loop in the calling thread, so we use it to gate the check.
+ @property
+ def _is_on_loop_thread(self) -> bool:
+ if threading.get_ident() == self._loop_thread_id:
+ with suppress(RuntimeError):
+ return bool(asyncio.get_running_loop())
+ return False
+
def send(self, msg: SendMsgType) -> ReceiveMsgType | None:
"""Send a request to the parent and block until the response is received."""
+ # Two-level detection for sync send() called from the event loop thread.
+ # Raises DeadlockImminentError (BaseException subclass) so it escapes
+ # contextlib.suppress(Exception) in mask_secret() and other helpers.
+ #
+ # Level 1 — wrong pattern (broader): send() is on the event loop thread at
+ # all, detected via the stored loop thread id from the last asend() call.
+ # Level 2 — imminent deadlock (precise): _thread_lock is also currently held
+ # by a thread-pool worker from a concurrent asend() round-trip.
+ #
+ # Typical cause: BaseHook.get_hook() or BaseHook.get_connection() called
+ # from inside an async task / aexecute().
+ # Fix: use 'await BaseHook.aget_hook()' or 'await BaseHook.aget_connection()'.
Review Comment:
This agent-generated comment is too verbose.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3509778692
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -232,18 +305,24 @@ def send(self, msg: SendMsgType) -> ReceiveMsgType | None:
if TYPE_CHECKING:
assert isinstance(resp, SentFDs)
resp.fds = fds
- # Since we know this is an expliclt SendFDs, and since this class is generic SendFDs might not
+ # Since we know this is an explicit ResendLoggingFD, and since this class is generic SentFDs might not
# always be in the return type union
return resp # type: ignore[return-value]
return self._get_response()
+ finally:
+ self._thread_lock.release()
async def asend(self, msg: SendMsgType) -> ReceiveMsgType | None:
"""
Send a request to the parent without blocking.
Uses async lock for coroutine safety and thread lock for socket safety.
"""
+ # Record which OS thread is running the event loop so that send() can
+ # detect when it is called from this same thread (the deadlock scenario).
Review Comment:
This is too obvious for a comment
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3490475328
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -118,6 +122,39 @@
ReceiveMsgType = TypeVar("ReceiveMsgType", bound=BaseModel)
+class DeadlockImminentError(BaseException):
+ """
+ Raised when ``CommsDecoder.send()`` is called from the event loop thread.
Review Comment:
This should probably mention it only happens when a lock is _also_ being acquired? The error is not actually always raised when `send` is called. (Not sure about this.)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4680155265
> `[other coroutine] send() → tries _thread_lock.acquire()`
>
> That should never be allowed to happen. We need to detect why sync send is being run on the async loop and trampoline back to the async version instead.
This is because of the mask_secret, which is sync, but the get_async_conn connection in msgraph is async, and it also logs some connection details (except password ofc), but the mask_secret is still applied anyway. We could add detection, but as mask_secret is sync, it would then crash anyway, but at least we would know why. That's why it was so difficult to track and why I didn't have the issue with SFTPAsyncHook
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4679835315
`[other coroutine] send() → tries _thread_lock.acquire()`
That should never be allowed to happen. We need to detect why sync send is being run on the async loop and trampoline back to the async version instead.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deadlock in CommsDecoder.asend() when sync send() runs concurrently on event loop [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4680869705
> > `[other coroutine] send() → tries _thread_lock.acquire()`
> > That should never be allowed to happen. We need to detect why sync send is being run on the async loop and trampoline back to the async version instead.
>
> This is because of the mask_secret, which is sync, but the get_async_conn connection in msgraph is async, and it also logs some connection details (except password ofc), but the mask_secret is still applied anyway. We could add detection, but as mask_secret is sync, it would then crash anyway, but at least we would know why. That's why it was so difficult to track and why I didn't have the issue with SFTPAsyncHook
That doesn't make sense to me. `mask_secret` is a function which makes no network requests - it just applies masks based on what it is given.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "dabla (via GitHub)" <gi...@apache.org>.
dabla commented on PR #68377:
URL: https://github.com/apache/airflow/pull/68377#issuecomment-4698619085
I've splitted the PR in two and create a separate [PR](https://github.com/apache/airflow/pull/68506) to add the asynchronous aget_hook method to BaseHook.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Added deadlock detection in CommsDecoder [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #68377:
URL: https://github.com/apache/airflow/pull/68377#discussion_r3412193556
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
Review Comment:
Not sure if I missed anything; the two blocks seem to have mostly the same logic except the former uses `blocking=False`. I think we can just do
```python
try:
if not self._thread_lock.acquire(blocking=not on_loop_thread):
raise DeadlockImminentError(...)
...
finally:
self._thread_lock.release()
```
or, if this is too difficult to understand, with an inner function
```python
def _communicate():
...
if on_loop_thread:
try:
if not acquire(blocking=False):
raise
_communicate()
finally:
release()
else:
with lock:
_communicate()
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org