slice-soupam opened a new pull request, #72008: URL: https://github.com/apache/airflow/pull/72008
Parser children that call `Variable.get()` or `xcom_push()` during parse or callbacks can pin a Dag-processor slot until `dag_file_processor_timeout`. This PR services IPC continuously while parsers are active, implements `SetXCom` on the Dag processor, and always replies to undecodable requests so a child cannot hang with no response. closes: #65369 ## Problem Two related gaps produce the same symptom: a parser child blocks on `socket.recv()` and occupies a slot so other Dags go stale. ### 1. Supported IPC is serviced too slowly (#65369) When a Dag file calls `Variable.get()` or `Connection.get()` during parse, the forked child sends an IPC request and waits for the parent. The parent only called `_service_processor_sockets()` once per `_run_parsing_loop()` iteration. While the parent is in `_refresh_dag_bundles()`, `_collect_results()`, stale-Dag scans, or other long work, children wait. Airflow 2 did a direct metadata-DB read for those lookups, so this is a 3.x regression. Profiling of the Dag processor shows `_parse_file()` time dominated by a single `socket.recv()` inside `CommsDecoder._get_response()`. ### 2. Unsupported IPC never gets a reply A production failure callback called `ti.xcom_push()`. The child sent `SetXCom`, which was not in the Dag processor `ToManager` union. `WatchedSubprocess.handle_requests` logged `Unable to decode message` and continued **without sending a reply**. The child hung until `dag_file_processor_timeout` (default 1200s), pinning a parser slot so other Dags stayed stale. `SetXCom` is already implemented on the triggerer path. Dag-processor callbacks that used `xcom_push()` in Airflow 2 silently lost that behavior after the IPC split. Any other unsupported type (`DeleteXCom`, `SucceedTask`, `DeferTask`, …) has the same hang-with-no-reply failure mode. ### Why #65370 was not reused #65370 tried a thread only around `_refresh_dag_bundles()`, used an unlocked selector, used a daemon thread with a short join, and had no tests. It was closed for inactivity and static-check failures, not because the issue was rejected. ## Proposed solution Three layers, all required to close the hang: 1. **Continuous IPC service.** A dedicated `dag-processor-ipc` thread polls processor sockets for the entire parsing loop, so children get replies while the main thread is busy. The in-loop `_service_processor_sockets()` call is removed. 2. **Locked selector.** `_LockedSelector` serializes `select` / `register` / `unregister` so the IPC thread and the main thread do not race on the shared selector. Timeout and orphan kills use `kill(..., wait=False)` so they send the signal without stealing that selector. 3. **Always reply.** `SetXCom` is added to `ToManager` and handled with the existing `handle_set_xcom` helper (same as the triggerer). Any other undecodable frame gets `ErrorResponse` with the request id, so the child's `CommsDecoder` raises immediately instead of hanging. `DeleteXCom` and other task-lifecycle messages stay unsupported on purpose; they now fail fast rather than pin a slot. This does not move callbacks off the Dag processor (that remains #44354 / #62887). ## Test plan - [x] `TestHandleRequest.test_handle_requests_undecodable_message_sends_error` — decode failure sends `ErrorResponse` instead of dropping the frame - [x] `TestDagFileProcessorProcess.test_handle_request_set_xcom` — Dag processor persists `SetXCom` - [x] `TestDagProcessingMessageTypes.test_to_manager_accepts_set_xcom` / `test_to_manager_rejects_unsupported_delete_xcom` - [x] `TestDagFileProcessor.test_variable_get_serviced_while_manager_is_busy` — child `Variable.get()` is answered while the main thread is not polling - [x] `TestDagFileProcessorManager.test_ipc_service_thread_starts_and_stops` / `test_ipc_service_thread_polls_while_caller_is_blocked` - [x] Timeout / orphan kill paths assert `kill(SIGKILL, wait=False)` - [ ] CI: airflow-core + task-sdk unit tests selected by selective-checks --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes — Cursor Grok 4.6 Generated-by: Cursor Grok 4.6 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions) Made with [Cursor](https://cursor.com) -- 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]
