Continuously service DAG processor IPC while parsers are active

[slice-soupam]

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.

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

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.

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.

Test plan

• TestHandleRequest.test_handle_requests_undecodable_message_sends_error — decode failure sends ErrorResponse instead of dropping the frame
• TestDagFileProcessorProcess.test_handle_request_set_xcom — Dag processor persists SetXCom
• TestDagProcessingMessageTypes.test_to_manager_accepts_set_xcom / test_to_manager_rejects_unsupported_delete_xcom
• TestDagFileProcessor.test_variable_get_serviced_while_manager_is_busy — child Variable.get() is answered while the main thread is not polling
• TestDagFileProcessorManager.test_ipc_service_thread_starts_and_stops / test_ipc_service_thread_polls_while_caller_is_blocked
• Timeout / orphan kill paths assert kill(SIGKILL, wait=False)
• CI: airflow-core + task-sdk unit tests selected by selective-checks

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack