buhuipao commented on code in PR #76423:
URL: https://github.com/apache/airflow/pull/76423#discussion_r4533748660
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -835,6 +837,55 @@ def _handle_request(self, msg: ToTriggerSupervisor, log:
FilteringBoundLogger, r
self.send_msg(resp, request_id=req_id, error=None, **dump_opts)
+ def _send_frame(self, data: bytes | bytearray) -> None:
+ """
+ Send ``data`` to the runner, reading its output sockets while the send
is blocked.
+
+ The runner's event loop can be stuck writing to a full log socket, so
it does not read this
+ response until we read its logs. A plain ``sendall()`` would leave
both processes waiting on
+ each other. EOF on an output socket is left for the main select loop
to handle.
+ """
+ view = memoryview(data)
+
+ def send_available() -> None:
+ nonlocal view
+ with suppress(BlockingIOError):
+ while view:
+ view = view[self.stdin.send(view) :]
+
+ # Non-blocking mode rather than MSG_DONTWAIT: macOS ignores that flag
for large AF_UNIX sends.
+ timeout = self.stdin.gettimeout()
+ self.stdin.setblocking(False)
+ try:
+ send_available()
+ if not view:
+ return
+ with selectors.DefaultSelector() as sel:
+ sel.register(self.stdin, selectors.EVENT_WRITE)
+ for key in list(self.selector.get_map().values()):
+ sock: socket = key.fileobj # type: ignore[assignment]
+ if sock is not self.stdin:
+ # The caller's select results may still list this
socket as readable after
+ # we drain it here; non-blocking makes that stale read
return instead of hang.
+ sock.setblocking(False)
+ sel.register(sock, selectors.EVENT_READ, key.data)
+ while view:
+ for key, _ in sel.select():
+ if key.fileobj is self.stdin:
+ send_available()
+ continue
+ socket_handler, _ = key.data
+ try:
+ need_more = socket_handler(key.fileobj)
+ except BlockingIOError:
+ need_more = True
+ except (BrokenPipeError, ConnectionResetError):
+ need_more = False
+ if not need_more:
Review Comment:
@seanghaeli Sorry, I don't mean that. Thanks for reviewing and checking.
The AI fixed the code following your advice and replied to your message
automatically.
--
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]
