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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
