[SakshamKapoor2911]

Fixes a runtime deadlock in LocalExecutor where the scheduler blocks indefinitely on activity_queue.put() during task dispatch, a gap left by #67881 which only addressed the shutdown path.

Workers fill the result_queue pipe buffer and block on put(). Blocked workers stop consuming activity_queue, so the scheduler blocks on activity_queue.put() and never reaches _read_results() to drain the pipe — a circular deadlock.

The fix: drain result_queue via _read_results() before each activity_queue.put() in _process_workloads, using the same drain-before-write pattern already applied to end().

closes: #70526

---

Was generative AI tooling used to co-author this PR?

• Yes — OpenCode (DeepSeek v4 Pro)

Generated-by: OpenCode (DeepSeek v4 Pro) following the guidelines

[SakshamKapoor2911]

@eladkal @amoghrajesh — all CI checks pass. Fixes a LocalExecutor shutdown deadlock. Ready for maintainer review.

[vatsrahul1001]

LGTM! should be good to merged post code owner review

[vatsrahul1001]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this as it not critical

[SakshamKapoor2911]

@vatsrahul1001 thanks for the approval and for moving this to the 3.3.2 milestone.

@XD-DENG @ashb @o-nikolas @pierrejeambrun @hussein-awala @dheerajturaga CI is green; PR needs a code owner sign-off to merge. It is a relatively small fix; could one of you take a look when you have a moment?

[darren-cheung-sa]

Moving to 3.3.2 as this is still pending code owner review and do not want to rush on merging this as it not critical

Seems it wasn't included in 3.3.2. And looks like there are no deadlines for 3.3.3 or 3.4.0

For our system (using LocalExecutor), it is a bit urgent since it means there is always a risk of a deadlock happening (stopping all our DAGs)

[darren-cheung-sa]

@vatsrahul1001 CC on this

[SakshamKapoor2911]

Rebased onto current main (62917eab) and re-verified.

Correction: an earlier comment described this as a shutdown deadlock fix. That was inaccurate — this PR addresses the runtime / dispatch path (_process_workloads()), which is the variant reported in #70526 and which #67881 (the end() drain) does not cover. #67881 fixed the shutdown/join path; this adds a result_queue drain before each activity_queue.put() during normal dispatch.

Why: workers can block writing results into a full result_queue pipe → they stop consuming activity_queue → the scheduler blocks in activity_queue.put() and, being single-threaded at that point, never reaches _read_results() → circular wait.

Change: one line in local_executor.py — self._read_results() before self.activity_queue.put(workload) in _process_workloads. The branch had gone stale against main's workload-API refactor, so I rebased and adapted the tests to the new API (get_workload_key / executor_queues / _dispatch_counts).

Tests (run locally on this branch, fork start method, Python 3.12):

• pytest airflow-core/tests/unit/executors/test_local_executor.py → 59 passed, 1 skipped.
• Added test_process_workloads_drains_results_before_put (asserts drain-before-dispatch ordering).
• Added a bounded real-queue regression test test_process_workloads_progresses_under_concurrent_result_refill that runs the actual _process_workloads with real multiprocessing.SimpleQueues and a worker concurrently refilling results. Under identical conditions it fails (times out) without the patch and completes with it (measured pipe capacity F_GETPIPE_SZ = 65536, kernel 6.8.0).

I have not verified a production deployment — this is a synthetic reproduction that passes with the patch. Feedback on the approach is welcome.

[github-actions]

Backport successfully created: v3-4-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-4-test