You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "kalluripradeep (via GitHub)" <gi...@apache.org> on 2026/04/07 21:33:49 UTC
[PR] fix(triggerer): fix thread-safety in TriggerCommsDecoder.asend() [airflow]
kalluripradeep opened a new pull request, #64869:
URL: https://github.com/apache/airflow/pull/64869
# #64620 fix: TriggerCommsDecoder.asend() asyncio.Lock is not thread-safe
### Description
The `TriggerCommsDecoder` used an `asyncio.Lock` to protect access to the communication socket. When called from a different event loop (e.g., via `sync_to_async` inside a trigger), `asyncio.Lock` would either fail to provide mutual exclusion or raise a `RuntimeError`.
This PR replaces the `asyncio.Lock` with an asynchronously-acquired `threading.Lock`, which is safe across all threads and loops. It also adds an explicit `drain()` to ensure reliability.
### Changes
- Replaced loop-bound `asyncio.Lock` with `threading.Lock` acquired via `run_in_executor`.
- Added `await self._async_writer.drain()` to `asend()` to ensure reliable message delivery.
--
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(triggerer): fix thread-safety in TriggerCommsDecoder.asend() [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #64869:
URL: https://github.com/apache/airflow/pull/64869#issuecomment-4341564586
I think this is mostly the same as #65622, but without the additional refactoring? Please correct me if I missed anything.
--
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(triggerer): fix thread-safety in TriggerCommsDecoder.asend() [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on PR #64869:
URL: https://github.com/apache/airflow/pull/64869#issuecomment-4356280334
That sounds resonable. I'd favor a bit the #65622 as this also fixes the same problem in base SDK.
--
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(triggerer): fix thread-safety in TriggerCommsDecoder.asend() [airflow]
Posted by "tirkarthi (via GitHub)" <gi...@apache.org>.
tirkarthi commented on code in PR #64869:
URL: https://github.com/apache/airflow/pull/64869#discussion_r3168146275
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -911,13 +911,22 @@ async def _aget_response(self, expect_id: int) -> ToTriggerRunner | None:
return self._from_frame(frame)
async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
- frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
- bytes = frame.as_bytes()
+ loop = asyncio.get_running_loop()
+ await loop.run_in_executor(None, self._thread_lock.acquire)
Review Comment:
`asyncio.to_thread` is available from Python 3.9 and does similar thing.
https://docs.python.org/3/library/asyncio-task.html#asyncio.to_thread
https://github.com/python/cpython/blob/4599335a83a22cf73a0c20e3692c29fd0278d23b/Lib/asyncio/threads.py#L12-L25
--
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(triggerer): fix thread-safety in TriggerCommsDecoder.asend() [airflow]
Posted by "tirkarthi (via GitHub)" <gi...@apache.org>.
tirkarthi commented on PR #64869:
URL: https://github.com/apache/airflow/pull/64869#issuecomment-4352752475
The PR was simpler to apply to our 3.1.x branch as the locking is moved closer to `asend`. There have been no issues after 6-7 hours of applying this patch. Using `await asyncio.to_thread(self._thread_lock.acquire)` waits for the lock and also yields control to event loop so that the other awaitables are completed to release the lock compared to `with self._thread_lock` which blocks the other awaitables to complete resulting in 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(triggerer): fix thread-safety in TriggerCommsDecoder.asend() [airflow]
Posted by "jedcunningham (via GitHub)" <gi...@apache.org>.
jedcunningham closed pull request #64869: fix(triggerer): fix thread-safety in TriggerCommsDecoder.asend()
URL: https://github.com/apache/airflow/pull/64869
--
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(triggerer): fix thread-safety in TriggerCommsDecoder.asend() [airflow]
Posted by "kalluripradeep (via GitHub)" <gi...@apache.org>.
kalluripradeep closed pull request #64869: fix(triggerer): fix thread-safety in TriggerCommsDecoder.asend()
URL: https://github.com/apache/airflow/pull/64869
--
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