ashb commented on PR #66412: URL: https://github.com/apache/airflow/pull/66412#issuecomment-4378712235
<details> <summary>Exercise/test script (expand to see)</summary> ```python #!/usr/bin/env python # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. """ Demonstrates the fix for the "Response read out of order" bug in TriggerCommsDecoder (github.com/apache/airflow/issues/64620). Three trigger types exercise all send() paths concurrently: SendFromThreadTrigger — calls send() from asyncio.to_thread() → run_coroutine_threadsafe() path (original bug: concurrent to_thread() workers raced on StreamReader) GreenbackTrigger — calls send() directly from async def run() on the event loop thread (greenback portal installed by run_trigger) → greenback.await_() path (would deadlock with run_coroutine_threadsafe(...).result()) AsyncTrigger — calls asend() from async def run() → pure async / no-lock path The supervisor responds to each request immediately in arrival order. The bug came from the trigger side (concurrent readers), not the supervisor reordering. Run with: uv run --project airflow-core python dev/demonstrate_trigger_comms_bug.py 2>/dev/null The 2>/dev/null drops asyncio transport warnings from socket teardown. """ from __future__ import annotations import asyncio import os import socket as _socket import sys import threading import traceback from collections.abc import AsyncGenerator from contextlib import suppress import msgspec os.environ.setdefault("AIRFLOW__DATABASE__SQL_ALCHEMY_CONN", "sqlite:////tmp/airflow-demo.db") os.environ.setdefault("AIRFLOW__CORE__FERNET_KEY", "8JmufCx8kB3Xme5oi5OonMoOWF7eKrJR53WCllrxZvs=") import logging logging.disable(logging.CRITICAL) import structlog # noqa: E402 from airflow.jobs.triggerer_job_runner import TriggerCommsDecoder, TriggerRunner, messages # noqa: E402 from airflow.sdk.execution_time import task_runner # noqa: E402 from airflow.sdk.execution_time.comms import _RequestFrame, _ResponseFrame # noqa: E402 from airflow.triggers.base import BaseTrigger, TriggerEvent # noqa: E402 structlog.configure( processors=[structlog.dev.ConsoleRenderer()], logger_factory=structlog.PrintLoggerFactory(file=open(os.devnull, "w")), ) N_EACH = 33 # triggers of each type; total = 99 TIMEOUT = 30 class SendFromThreadTrigger(BaseTrigger): """ Calls send() from asyncio.to_thread() — the original bug scenario. Maps to run_coroutine_threadsafe() path in TriggerCommsDecoder.send(). """ def __init__(self, trigger_idx: int): super().__init__() self.trigger_idx = trigger_idx def serialize(self): return f"{__name__}.SendFromThreadTrigger", {"trigger_idx": self.trigger_idx} async def run(self) -> AsyncGenerator[TriggerEvent, None]: def sync_work(): task_runner.SUPERVISOR_COMMS.send( messages.TriggerStateChanges( events=None, finished=[self.trigger_idx], failures=None, ) ) await asyncio.to_thread(sync_work) yield TriggerEvent(value=self.trigger_idx) class GreenbackTrigger(BaseTrigger): """ Calls send() directly from async def run() — on the event loop thread. run_trigger() installs a greenback portal before iterating triggers, so TriggerCommsDecoder.send() uses greenback.await_() here to avoid deadlock. """ def __init__(self, trigger_idx: int): super().__init__() self.trigger_idx = trigger_idx def serialize(self): return f"{__name__}.GreenbackTrigger", {"trigger_idx": self.trigger_idx} async def run(self) -> AsyncGenerator[TriggerEvent, None]: task_runner.SUPERVISOR_COMMS.send( messages.TriggerStateChanges( events=None, finished=[self.trigger_idx], failures=None, ) ) yield TriggerEvent(value=self.trigger_idx) class AsyncTrigger(BaseTrigger): """ Calls asend() directly — pure async, no threads, no greenback needed. """ def __init__(self, trigger_idx: int): super().__init__() self.trigger_idx = trigger_idx def serialize(self): return f"{__name__}.AsyncTrigger", {"trigger_idx": self.trigger_idx} async def run(self) -> AsyncGenerator[TriggerEvent, None]: await task_runner.SUPERVISOR_COMMS.asend( messages.TriggerStateChanges( events=None, finished=[self.trigger_idx], failures=None, ) ) yield TriggerEvent(value=self.trigger_idx) def _read_frame(sock) -> _RequestFrame | None: lb = b"" while len(lb) < 4: try: chunk = sock.recv(4 - len(lb)) except OSError: return None if not chunk: return None lb += chunk n = int.from_bytes(lb, "big") data = b"" while len(data) < n: chunk = sock.recv(n - len(data)) if not chunk: return None data += chunk return msgspec.msgpack.decode(data, type=_RequestFrame) def run_supervisor(server_sock) -> None: """ Sends StartTriggerer then echoes a TriggerStateSync for every request. Responds in arrival order. Runs until the client closes the connection. """ server_sock.sendall(_ResponseFrame(id=0, body={"type": "StartTriggerer"}).as_bytes()) while True: frame = _read_frame(server_sock) if frame is None: break server_sock.sendall( _ResponseFrame( id=frame.id, body={"type": "TriggerStateSync", "to_create": [], "to_cancel": []}, ).as_bytes() ) async def run_attempt(n_each: int) -> tuple[list[Exception], int]: n_total = n_each * 3 server_sock, client_sock = _socket.socketpair() sup = threading.Thread(target=run_supervisor, args=(server_sock,), daemon=True) sup.start() reader, writer = await asyncio.open_connection(sock=client_sock) decoder = TriggerCommsDecoder(async_writer=writer, async_reader=reader, socket=client_sock) frame = await decoder._aread_frame() start_msg = decoder._from_frame(frame) if not isinstance(start_msg, messages.StartTriggerer): raise RuntimeError(f"Expected StartTriggerer, got {type(start_msg)}") await decoder.start_reader() runner = TriggerRunner() runner.comms_decoder = decoder task_runner.SUPERVISOR_COMMS = decoder trigger_id = 0 for i in range(n_each): for cls in (SendFromThreadTrigger, GreenbackTrigger, AsyncTrigger): trigger = cls(trigger_idx=trigger_id) name = f"{cls.__name__}-{i}" runner.triggers[trigger_id] = {"task": None, "is_watcher": False, "name": name, "events": 0} runner.triggers[trigger_id]["task"] = asyncio.create_task( runner.run_trigger(trigger_id, trigger), name=name ) trigger_id += 1 errors: list[Exception] = [] deadline = asyncio.get_event_loop().time() + TIMEOUT while runner.triggers and asyncio.get_event_loop().time() < deadline: finished_ids = await runner.cleanup_finished_triggers() try: await runner.sync_state_to_supervisor(finished_ids) except Exception as e: errors.append(e) break await asyncio.sleep(0.01) for details in runner.triggers.values(): task = details["task"] if task.done() and not task.cancelled(): exc = task.exception() if exc: errors.append(exc) n_ok = n_total - len(errors) if (reader_task := decoder._reader_task) is not None: reader_task.cancel() with suppress(asyncio.CancelledError, Exception): await reader_task writer.close() server_sock.close() sup.join(timeout=5) return errors, n_ok async def main() -> None: n_total = N_EACH * 3 print(f"Python {sys.version}") print(f"{N_EACH} SendFromThreadTrigger (send via asyncio.to_thread → run_coroutine_threadsafe)") print(f"{N_EACH} GreenbackTrigger (send directly from async def run → greenback.await_)") print(f"{N_EACH} AsyncTrigger (asend directly from async def run)") print(f"Total: {n_total} triggers, supervisor responds in arrival order\n") for attempt in range(1, 21): print(f"Attempt {attempt}/20...", end=" ", flush=True) try: errors, n_ok = await run_attempt(N_EACH) except Exception as e: errors, n_ok = [e], 0 if errors: print(f"BUG TRIGGERED ({n_ok}/{n_total} ok, {len(errors)} errors)") print() for e in errors[:3]: traceback.print_exception(type(e), e, e.__traceback__, limit=5, file=sys.stdout) if len(errors) > 3: print(f" ... and {len(errors) - 3} more") sys.exit(1) else: print(f"ok ({n_ok}/{n_total})") print("\nAll 20 attempts clean.") sys.exit(0) if __name__ == "__main__": asyncio.run(main()) ``` </details> -- 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]
