dabla commented on code in PR #65622:
URL: https://github.com/apache/airflow/pull/65622#discussion_r3131628890
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -912,14 +912,20 @@ async def _aget_response(self, expect_id: int) ->
ToTriggerRunner | None:
raise RuntimeError(f"Response read out of order! Got {frame.id=},
{expect_id=}")
return self._from_frame(frame)
- async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
+ async def _asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
frame = _RequestFrame(id=next(self.id_counter), body=msg.model_dump())
bytes = frame.as_bytes()
+ self._async_writer.write(bytes)
+ return await self._aget_response(frame.id)
+ async def asend(self, msg: ToTriggerSupervisor) -> ToTriggerRunner | None:
Review Comment:
We had the same issue with CommsDecoder, I also had to apply a fix lately
reported by @uranusjr.
The solution brought by @renat-sagut is similar to the asend method in
CommsDecoder which looks like this:
`
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.
"""
frame = _RequestFrame(id=next(self.id_counter),
body=msg.model_dump())
frame_bytes = frame.as_bytes()
async with self._async_lock:
# Acquire the threading lock without blocking the event loop
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._thread_lock.acquire)
try:
# Async write to socket
await loop.sock_sendall(self.socket, frame_bytes)
if isinstance(msg, ResendLoggingFD):
if recv_fds is None:
return None
# Blocking read in a thread
frame, fds = await asyncio.to_thread(self._read_frame,
maxfds=1)
resp = self._from_frame(frame)
if TYPE_CHECKING:
assert isinstance(resp, SentFDs)
resp.fds = fds
return resp # type: ignore[return-value]
# Normal blocking read in a thread
frame = await asyncio.to_thread(self._read_frame)
return self._from_frame(frame)
finally:
self._thread_lock.release()
`
Maybe we should try to make the critical part generic in CommDecoder (dunno
if possible), so we don't encounter this regression at different places.
--
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]
