dabla commented on code in PR #62922:
URL: https://github.com/apache/airflow/pull/62922#discussion_r4252492106
##########
task-sdk/src/airflow/sdk/execution_time/comms.py:
##########
@@ -241,53 +237,35 @@ class CommsDecoder(Generic[ReceiveMsgType, SendMsgType]):
_thread_lock: threading.Lock = attrs.field(factory=threading.Lock,
repr=False)
_async_lock: asyncio.Lock = attrs.field(factory=asyncio.Lock, repr=False)
- _loop_thread_id: int | None = attrs.field(default=None, repr=False,
init=False)
+ # The thread exchanging a request while it holds _thread_lock, so send()
can tell a call that
+ # would wait for its own thread from one that waits for another.
+ _channel_owner: 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)
- @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
-
- @property
- def _asend_in_flight_on_this_thread(self) -> bool:
+ def send(self, msg: SendMsgType) -> ReceiveMsgType | None:
"""
- Whether an ``asend()`` coroutine of a loop that runs on the current
thread is in flight.
-
- ``asend()`` holds ``_async_lock`` for its whole duration, from before
it takes
- ``_thread_lock`` (in an executor thread) until after it releases it,
so the async lock
- being held is the reliable signal that the thread lock is, or is about
to be, held by a
- coroutine. Such a coroutine only makes progress when its loop runs,
and that loop runs on
- this thread. This is thread-scoped on purpose: a loop running on
*another* thread keeps
- going while this one blocks, so ``send()`` from there can safely wait
for the lock.
+ Send a request to the parent and block until the response is received.
+
+ Called on an event loop's thread while an ``asend()`` is in flight,
this waits for that
+ request's round trip: ``asend()`` runs on a worker thread that needs
no loop to finish, so
+ the wait ends, but it stalls every coroutine on the loop meanwhile.
Async code should
+ prefer ``asend()`` and the async SDK calls built on it. Only a call on
a thread that is
+ itself in the middle of a request raises
:class:`DeadlockImminentError`, as waiting there
+ would never end.
"""
- return threading.get_ident() == self._loop_thread_id and
self._async_lock.locked()
-
- def send(self, msg: SendMsgType) -> ReceiveMsgType | None:
- """Send a request to the parent and block until the response is
received."""
- frame_bytes = self._make_frame(msg).as_bytes()
-
- # An asend() in flight on this thread's loop can only release
_thread_lock once that loop
- # runs again, and the loop cannot run while this thread blocks in
send(). Raise instead of
- # blocking. This covers the loop being paused between two
run_until_complete() calls, where
- # the running-loop check below does not apply:
asyncio.get_running_loop() raises, so send()
- # would otherwise fall through to a blocking acquire and freeze the
process with every
- # thread idle (seen with a sync XCom read pulling the next iterated
sub-task's input while a
- # sub-task's asend() was parked mid-I/O).
- if self._asend_in_flight_on_this_thread:
+ if self._channel_owner == threading.get_ident():
Review Comment:
Right, and it is reachable outside this PR too: the `AirflowTaskTimeout` of
the `execution_timeout` alarm unwinds an exchange the same way for any task. I
kept this PR to documenting it (4e5bd9c895): `_exchange`'s docstring now says
responses are read in order without checking the frame id, which two ways an
exception unwinds a request between its write and its read, and that the next
request then reads the stale response as its own. A real fix needs either a
read that can resume a partly read frame (a signal can land between the 4-byte
header and the body, so skipping by id alone is not enough) or a channel marked
unusable after an interrupted exchange; since it changes the channel for every
task, I would rather do it in its own PR than here.
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]
