apache / apache/airflow

DagFileProcessorManager IPC bottleneck: child processes block on Variable.get() during parsing

Open
#65,369 1 comment 0 reactions 0 assignees View on GitHub
affected_version:3.2 area:core area:DAG-processing kind:bug priority:upgrade_to_airflow3
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

### Apache Airflow version

3.0.0+

### What happened

When DAG files call `Variable.get()` during parsing, the forked child process blocks on `socket.recv()` waiting for the parent `DagFileProcessorManager` to service the IPC request. However, IPC is only serviced once per main loop iteration in `_service_processor_sockets()` (manager.py line 398). This means child processes can be blocked for the entire duration of the parent's loop — which includes `_refresh_dag_bundles()`, `_collect_results()`, `_scan_stale_dags()`, etc.

With multiple parsing processes running and a non-trivial number of DAG files, the parent loop takes significant time per iteration. During this time, any child process that calls `Variable.get()` is completely blocked, turning a sub-second DAG parse into a multi-minute wait.

**Profiling confirms the bottleneck**: cProfile of the dag-processor shows `_parse_file()` time is dominated by a single `socket.recv()` call inside `CommsDecoder._get_response()`, waiting for the parent to respond to the `GetVariable` IPC request.

This is a regression from Airflow 2.x where `Variable.get()` was a direct DB query with no IPC dependency.

### What you think should happen instead

The dag-processor should continuously service IPC requests from child processes, not just once per main loop iteration.

For comparison, `WatchedSubprocess._monitor_subprocess()` in the task execution path has a tight loop that services IPC continuously — the dag-processor should have similar behavior.

One approach: run `_service_processor_sockets()` in a dedicated background thread during the phases of the main loop where child processes may be actively parsing (i.e., between `_start_new_processes()` and `_collect_results()`). This ensures child IPC requests are serviced immediately regardless of what the parent's main loop is doing.

### How to reproduce

1. Create multiple DAG files that call `Variable.get()` during module-level or top-level DAG definition
2. Run the standalone dag-processor with multiple `parsing_processes`
3. Profile the dag-processor — observe that `_parse_file()` duration is dominated by `socket.recv()` waiting for IPC response
4. The more DAG files and parsing processes, the worse the contention becomes

### Operating System

Any

### Versions of Apache Airflow Providers

N/A

### Deployment

Other

### Deployment details

Standalone dag-processor with multiple parsing processes

### Anything else

**Root cause in code:**

In `_run_parsing_loop()` (manager.py), `_service_processor_sockets()` is called once at line 398. Between `_start_new_processes()` (line 396) and `_service_processor_sockets()` (line 398) there is no gap, but the issue is that after servicing sockets once, the loop continues with `_collect_results()`, `_scan_stale_dags()`, `_cleanup_stale_bundle_versions()`, etc. — and during all of that, no IPC is being serviced. If any of these steps take time (especially `_refresh_dag_bundles()` on the next iteration), child processes are blocked.

The problem compounds because `_service_processor_sockets()` only does a single `select()` call — it doesn't drain all pending requests, just whatever is ready at that instant.

**Contrast with task execution path:**

`WatchedSubprocess._monitor_subprocess()` in `supervisor.py` has a dedicated tight loop that continuously services IPC for running tasks. The dag-processor lacks this — its IPC servicing is interleaved with many other operations in the main loop.

Related issues: #56446, #42205

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.