This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 769d0700b88c9e0c2436681067d1bd924f56f58b
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sun Nov 1 14:02:33 2026 +0100
[v3-4-test] Fix LocalExecutor deadlock when a workload put blocks (#75086)
(#76223)
* Keep draining LocalExecutor results while a workload put is blocked
Draining the result queue once before each put did not remove the
deadlock reported in #70526. A worker can write a result between the
drain and the put; if the activity pipe is full at that point the put
blocks, and the worker keeps writing results until the result pipe is
full too, so each side waits on the other. Linux pipes free space only a
page at a time, which is why the regression test from #70545 hangs on
loaded CI runners but not on macOS.
Make the put on a short-lived helper thread and keep reading results on
the main thread until it completes, so the result pipe can never fill up
behind a blocked put. The workload is registered as running before it is
sent, so a "running" result read while the put is in flight is not
dropped.
Generated-by: Claude Opus 5
* Undo a failed LocalExecutor put and spawn workers while one is blocked
Registering a workload before its put meant a put that raised left the
key running and counted although nothing was sent; restore the queued
workload and its bookkeeping instead. Counting each workload before its
put keeps a worker from decrementing the outstanding count below zero
when a batch is interrupted part way.
A blocked put can only complete once a live worker reads the activity
pipe, but workers were only spawned or replaced after the batch, so with
the spawn start method (or once every worker had died) a batch larger
than the pipe waited forever. Spawn them from the drain loop whenever
fewer than parallelism are alive: spawn starts one per call, so a single
busy worker would otherwise hold the put until its task finished.
Generated-by: Claude Opus 5
* Cover undoing a failed put while an earlier dispatch of the key is running
The test only reached the no-earlier-dispatch path, so the remaining > 0
branch of _undo_dispatch had no coverage.
(cherry picked from commit 1d1d17bb718dbb1362dbfabcefc37755c4074763)
Generated-by: Claude Opus 5
Co-authored-by: Jarek Potiuk <[email protected]>
---
.../src/airflow/executors/local_executor.py | 66 +++++-
.../tests/unit/executors/test_local_executor.py | 232 ++++++++++++++++++++-
2 files changed, 293 insertions(+), 5 deletions(-)
diff --git a/airflow-core/src/airflow/executors/local_executor.py
b/airflow-core/src/airflow/executors/local_executor.py
index aebadbf66ad..6bbc9274a8b 100644
--- a/airflow-core/src/airflow/executors/local_executor.py
+++ b/airflow-core/src/airflow/executors/local_executor.py
@@ -31,6 +31,7 @@ import multiprocessing
import multiprocessing.sharedctypes
import os
import sys
+import threading
from multiprocessing import Queue, SimpleQueue
from typing import TYPE_CHECKING
@@ -337,17 +338,74 @@ class LocalExecutor(BaseExecutor):
def _process_workloads(self, workload_list):
for workload in workload_list:
key = self.get_workload_key(workload)
- self._read_results()
- self.activity_queue.put(workload)
removed = self.executor_queues[workload.type].pop(key, None)
if not removed:
raise KeyError(f"Workload {key} was not found in any queue")
+ # Register before the put: results read while the put is in flight
are dropped for keys
+ # that are not running yet.
self.running.add(key)
self._dispatch_counts[key] = self._dispatch_counts.get(key, 0) + 1
- with self._unread_messages:
- self._unread_messages.value += len(workload_list)
+ # Counted before the put so a worker taking the message never
decrements below zero, and so
+ # _check_workers sees it as outstanding while the put is blocked.
+ with self._unread_messages:
+ self._unread_messages.value += 1
+ self._read_results()
+ if (put_error := self._put_workload(workload)) is not None:
+ self._undo_dispatch(key, workload)
+ raise put_error
self._check_workers()
+ def _undo_dispatch(self, key: WorkloadKey, workload: ExecutorWorkload) ->
None:
+ """Reverse the bookkeeping of a workload whose put failed, leaving it
queued as it was before."""
+ self.executor_queues[workload.type][key] = workload
+ remaining = self._dispatch_counts.pop(key, 1) - 1
+ if remaining > 0:
+ self._dispatch_counts[key] = remaining
+ else:
+ self.running.discard(key)
+ with self._unread_messages:
+ self._unread_messages.value -= 1
+
+ def _put_workload(self, workload: ExecutorWorkload) -> BaseException |
None:
+ """
+ Put ``workload`` on the activity queue while continuing to drain the
result queue.
+
+ The put blocks while the activity pipe is full. If the workers are at
the same time blocked
+ writing to a full result pipe, a put on this thread alone would never
return, so it runs on a
+ helper thread while this thread keeps reading results.
+
+ :return: the error the put itself raised, in which case the workload
was not sent.
+ """
+ put_errors: list[BaseException] = []
+
+ def _put() -> None:
+ try:
+ self.activity_queue.put(workload)
+ except BaseException as e:
+ put_errors.append(e)
+
+ putter = threading.Thread(target=_put, name="local-executor-put",
daemon=True)
+ try:
+ putter.start()
+ putter.join(timeout=0.05)
+ while putter.is_alive():
+ if sum(proc.is_alive() for proc in self.workers.values()) <
self.parallelism:
+ # Only a live worker can take the pending write, and
workers are otherwise only
+ # spawned or replaced after the batch: with spawn there
are none before the first one,
+ # and each call starts just one, so a single busy worker
must not stop the rest.
+ self._check_workers()
+ self._read_results()
+ putter.join(timeout=0.05)
+ except BaseException:
+ # end() sends a shutdown message on the same pipe for each live
worker, so an interruption
+ # (SIGTERM raises SystemExit here) must not leave this writer
blocked behind a full result
+ # pipe. Without a live worker nothing can take the pending write,
and end() sends nothing.
+ while putter.is_alive() and any(proc.is_alive() for proc in
self.workers.values()):
+ self._read_results()
+ putter.join(timeout=0.05)
+ raise
+ return put_errors[0] if put_errors else None
+
def _finish_dispatch(self, key: WorkloadKey, state: WorkloadState) -> None:
# A resumed attempt reuses its key, so the previous dispatch can
finish while the next one is live.
remaining = self._dispatch_counts.pop(key, 1) - 1
diff --git a/airflow-core/tests/unit/executors/test_local_executor.py
b/airflow-core/tests/unit/executors/test_local_executor.py
index 92572b9844e..d7614c32ac6 100644
--- a/airflow-core/tests/unit/executors/test_local_executor.py
+++ b/airflow-core/tests/unit/executors/test_local_executor.py
@@ -23,6 +23,7 @@ import os
import signal
import threading
import time
+from multiprocessing.queues import SimpleQueue as SimpleQueueType
from pathlib import Path
from unittest import mock
@@ -47,6 +48,8 @@ from airflow.utils.state import State
from tests_common.test_utils.config import conf_vars
from tests_common.test_utils.markers import
skip_if_force_lowest_dependencies_marker
+_ORIGINAL_THREAD_START = threading.Thread.start
+
pytestmark = pytest.mark.db_test
# Mock patching doesn't work across process boundaries with 'spawn' (default
on macOS)
@@ -477,7 +480,234 @@ class TestLocalExecutor:
executor._process_workloads(workloads_list)
- assert call_order == ["_read_results", "put"] * num_workloads
+ put_indices = [i for i, call in enumerate(call_order) if call == "put"]
+ assert len(put_indices) == num_workloads
+ assert all(i > 0 and call_order[i - 1] == "_read_results" for i in
put_indices)
+
+ def test_put_workload_drains_results_while_put_is_blocked(self):
+ """A put blocked on a full activity pipe must not stop the result
queue from being drained."""
+ executor = LocalExecutor(parallelism=1)
+ executor.activity_queue = mock.MagicMock(spec=SimpleQueueType)
+ executor.workers = {1: mock.Mock(spec=multiprocessing.Process,
**{"is_alive.return_value": True})}
+ put_started = threading.Event()
+ drained_during_put = threading.Event()
+
+ def _put_blocked_until_drained(_workload):
+ put_started.set()
+ if not drained_during_put.wait(timeout=10):
+ raise TimeoutError("result queue was not drained while the put
was blocked")
+
+ executor.activity_queue.put.side_effect = _put_blocked_until_drained
+ executor._read_results = mock.MagicMock(
+ side_effect=lambda: put_started.is_set() and
drained_during_put.set()
+ )
+
+ executor._put_workload(_make_task_workload())
+
+ assert drained_during_put.is_set()
+
+ def
test_put_workload_keeps_draining_until_put_completes_when_interrupted(self):
+ """An interruption during a blocked put must not abandon the writer
that end() would queue behind."""
+ executor = LocalExecutor(parallelism=1)
+ executor.activity_queue = mock.MagicMock(spec=SimpleQueueType)
+ executor.workers = {1: mock.Mock(spec=multiprocessing.Process,
**{"is_alive.return_value": True})}
+ drains = []
+ put_finished = threading.Event()
+
+ def _put_blocked_until_drained_twice(_workload):
+ deadline = time.monotonic() + 10
+ while len(drains) < 2 and time.monotonic() < deadline:
+ time.sleep(0.01)
+ put_finished.set()
+
+ def _read_results():
+ drains.append(None)
+ if len(drains) == 1:
+ raise SystemExit(1)
+
+ executor.activity_queue.put.side_effect =
_put_blocked_until_drained_twice
+ executor._read_results = mock.MagicMock(spec=executor._read_results,
side_effect=_read_results)
+
+ with pytest.raises(SystemExit):
+ executor._put_workload(_make_task_workload())
+
+ assert put_finished.is_set()
+ assert len(drains) >= 2
+
+ @mock.patch.object(threading.Thread, "start", autospec=True)
+ def
test_put_workload_keeps_draining_when_interrupted_while_starting_the_writer(self,
mock_start):
+ """An interruption raised while the writer is starting must not skip
the drain either."""
+ executor = LocalExecutor(parallelism=1)
+ executor.activity_queue = mock.MagicMock(spec=SimpleQueueType)
+ executor.workers = {1: mock.Mock(spec=multiprocessing.Process,
**{"is_alive.return_value": True})}
+ drains = []
+ put_finished = threading.Event()
+
+ def _put_blocked_until_drained(_workload):
+ deadline = time.monotonic() + 10
+ while not drains and time.monotonic() < deadline:
+ time.sleep(0.01)
+ put_finished.set()
+
+ executor.activity_queue.put.side_effect = _put_blocked_until_drained
+ executor._read_results = mock.MagicMock(
+ spec=executor._read_results, side_effect=lambda:
drains.append(None)
+ )
+
+ def _start_then_interrupt(thread):
+ _ORIGINAL_THREAD_START(thread)
+ raise SystemExit(1)
+
+ mock_start.side_effect = _start_then_interrupt
+
+ with pytest.raises(SystemExit):
+ executor._put_workload(_make_task_workload())
+
+ assert put_finished.is_set()
+ assert drains
+
+ def
test_put_workload_reraises_interruption_when_no_worker_can_take_the_put(self):
+ """Without a live worker the blocked write can never complete, so the
interruption is re-raised."""
+ executor = LocalExecutor(parallelism=1)
+ executor.activity_queue = mock.MagicMock(spec=SimpleQueueType)
+ executor.workers = {}
+ release_put = threading.Event()
+ put_done = threading.Event()
+
+ def _put_blocked_until_released(_workload):
+ release_put.wait(timeout=10)
+ put_done.set()
+
+ executor.activity_queue.put.side_effect = _put_blocked_until_released
+ executor._read_results = mock.MagicMock(
+ spec=executor._read_results, side_effect=[SystemExit(1), *[None] *
1000]
+ )
+
+ try:
+ with pytest.raises(SystemExit):
+ executor._put_workload(_make_task_workload())
+ assert not put_done.is_set()
+ finally:
+ release_put.set()
+
+ def test_put_workload_returns_put_error(self):
+ executor = LocalExecutor(parallelism=1)
+ executor.activity_queue = mock.MagicMock(spec=SimpleQueueType)
+ error = ValueError("unpicklable")
+ executor.activity_queue.put.side_effect = error
+
+ assert executor._put_workload(_make_task_workload()) is error
+
+ @pytest.mark.parametrize(
+ ("earlier_dispatches", "still_running"),
+ [
+ pytest.param(0, False, id="no-earlier-dispatch"),
+ pytest.param(1, True, id="earlier-dispatch-still-running"),
+ ],
+ )
+ def test_process_workloads_undoes_dispatch_when_put_fails(self,
earlier_dispatches, still_running):
+ """A failed put undoes only its own dispatch, so an earlier dispatch
of the same key stays running."""
+ executor = LocalExecutor(parallelism=1)
+ executor.workers = {}
+ executor.activity_queue = mock.MagicMock(spec=SimpleQueueType)
+ executor.activity_queue.put.side_effect = ValueError("unpicklable")
+ executor._read_results = mock.MagicMock(spec=executor._read_results)
+ executor._unread_messages = multiprocessing.Value("I", 0)
+ executor._check_workers = mock.MagicMock(spec=executor._check_workers)
+ workload = _make_task_workload()
+ key = executor.get_workload_key(workload)
+ executor.executor_queues.setdefault(workload.type, {})[key] = workload
+ if earlier_dispatches:
+ # A resumed attempt reuses the key while the earlier dispatch has
not reported its result yet.
+ executor.running.add(key)
+ executor._dispatch_counts[key] = earlier_dispatches
+
+ with pytest.raises(ValueError, match="unpicklable"):
+ executor._process_workloads([workload])
+
+ assert (key in executor.running) is still_running
+ assert executor._dispatch_counts.get(key, 0) == earlier_dispatches
+ assert executor.executor_queues[workload.type][key] is workload
+ assert executor._unread_messages.value == 0
+
+ def test_process_workloads_counts_workload_before_put(self):
+ """A worker decrements the count when it takes the message, so it must
be counted first."""
+ executor = LocalExecutor(parallelism=1)
+ executor.activity_queue = mock.MagicMock(spec=SimpleQueueType)
+ executor._read_results = mock.MagicMock(spec=executor._read_results)
+ executor._unread_messages = multiprocessing.Value("I", 0)
+ executor._check_workers = mock.MagicMock(spec=executor._check_workers)
+ workload = _make_task_workload()
+ executor.executor_queues.setdefault(workload.type,
{})[executor.get_workload_key(workload)] = workload
+ counted_at_put = []
+ executor.activity_queue.put.side_effect = lambda _workload:
counted_at_put.append(
+ executor._unread_messages.value
+ )
+
+ executor._process_workloads([workload])
+
+ assert counted_at_put == [1]
+
+ def
test_put_workload_spawns_workers_up_to_parallelism_while_one_is_busy(self):
+ """A busy worker must not stop more workers from being spawned while a
put is blocked."""
+ executor = LocalExecutor(parallelism=2)
+ executor.activity_queue = mock.MagicMock(spec=SimpleQueueType)
+ executor._read_results = mock.MagicMock(spec=executor._read_results)
+ busy_worker = mock.Mock(spec=multiprocessing.Process,
**{"is_alive.return_value": True})
+ executor.workers = {1: busy_worker}
+ second_worker_started = threading.Event()
+
+ def _spawn_worker():
+ executor.workers = {
+ 1: busy_worker,
+ 2: mock.Mock(spec=multiprocessing.Process,
**{"is_alive.return_value": True}),
+ }
+ second_worker_started.set()
+
+ executor._check_workers = mock.MagicMock(spec=executor._check_workers,
side_effect=_spawn_worker)
+ executor.activity_queue.put.side_effect = lambda _workload:
second_worker_started.wait(timeout=10)
+
+ assert executor._put_workload(_make_task_workload()) is None
+
+ assert second_worker_started.is_set()
+
+ def test_put_workload_spawns_a_worker_when_none_is_alive(self):
+ """With no live worker the blocked put can never complete, so workers
are spawned while it waits."""
+ executor = LocalExecutor(parallelism=1)
+ executor.activity_queue = mock.MagicMock(spec=SimpleQueueType)
+ executor._read_results = mock.MagicMock(spec=executor._read_results)
+ executor.workers = {}
+ worker_started = threading.Event()
+
+ def _spawn_worker():
+ executor.workers = {1: mock.Mock(spec=multiprocessing.Process,
**{"is_alive.return_value": True})}
+ worker_started.set()
+
+ executor._check_workers = mock.MagicMock(spec=executor._check_workers,
side_effect=_spawn_worker)
+ executor.activity_queue.put.side_effect = lambda _workload:
worker_started.wait(timeout=10)
+
+ assert executor._put_workload(_make_task_workload()) is None
+
+ assert worker_started.is_set()
+
+ def test_process_workloads_registers_running_before_put(self):
+ """A result read while the put is in flight is dropped unless the key
is already running."""
+ executor = LocalExecutor(parallelism=1)
+ executor.activity_queue = mock.MagicMock(spec=SimpleQueueType)
+ executor._read_results = mock.MagicMock(spec=executor._read_results)
+ executor._unread_messages = multiprocessing.Value("I", 0)
+ executor._check_workers = mock.MagicMock(spec=executor._check_workers)
+ workload = _make_task_workload()
+ key = executor.get_workload_key(workload)
+ executor.executor_queues.setdefault(workload.type, {})[key] = workload
+ running_at_put = []
+ executor.activity_queue.put.side_effect = lambda _workload:
running_at_put.append(
+ key in executor.running
+ )
+
+ executor._process_workloads([workload])
+
+ assert running_at_put == [True]
def test_process_workloads_progresses_under_concurrent_result_refill(self,
mocker):
"""#70545: with real queues, _process_workloads must keep draining so
a concurrently
