andrei-gromov opened a new issue, #74575:
URL: https://github.com/apache/airflow/issues/74575
### Under which category would you file this issue?
Airflow Core
### Apache Airflow version
3.3.2
### What happened and how to reproduce it?
**Issue description**
With the LocalExecutor, when the idle worker that currently holds the read
lock of the executor's
activity queue is killed (SIGKILL, which is what the kernel's OOM killer
sends), the executor stops
starting tasks. Every task scheduled afterwards stays `queued`. The
scheduler keeps running and
reports healthy, so nothing flags the problem. Only restarting the scheduler
recovers.
**Why it happens** (Airflow 3.3.2,
`airflow-core/src/airflow/executors/local_executor.py`)
- `:158-159`: `activity_queue` and `result_queue` are
`multiprocessing.SimpleQueue()`.
- `:167-170`: with the `fork` start method, `start()` forks all
`parallelism` workers at once.
- `:78-81`: each idle worker loops on `workload = input.get()`.
- In CPython 3.13 (`Lib/multiprocessing/queues.py`), `SimpleQueue.__init__`
creates
`self._rlock = ctx.Lock()` (line 361) and `get()` is
`with self._rlock: res = self._reader.recv_bytes()` (lines 383-385). So
exactly one idle worker
holds `_rlock` while it blocks reading the pipe, and every other idle
worker waits for the lock.
On the stand, one worker was in `anon_pipe_read` and the others in
`futex_do_wait`.
- `_rlock` is a semaphore-based `multiprocessing` lock. If its holder is
killed, nothing releases
it.
- `_process_workloads()` (`:280-294`) only takes the write lock to `put()`
the workload, so the
scheduler's `put()` succeeds, and nobody can read it.
- `_check_workers()` (`:172-202`) reaps the dead worker and, because work is
outstanding, forks a
replacement. The replacement calls `input.get()` too and waits for the
same dead lock, so the
executor believes it has `parallelism` workers. I saw the three other idle
workers and the
replacement all in `futex_do_wait`.
The Python documentation warns about killing a process that is using a
`Queue` (the data can become
corrupted); `SimpleQueue` is documented as very close to a locked `Pipe`.
**Steps to reproduce**
1. Run `airflow scheduler` with `[core] executor = LocalExecutor`, Python
3.13 on Linux (start
method `fork`) and `parallelism = 4`. I used the official image, Docker
Compose.
2. Wait until all workers are idle. On the host, `ps -eo pid,args | grep
'airflow worker -- LocalExecutor'`
shows `<idle>` for each of them (the processes of the container are
visible there).
3. On the host, look at the kernel stack of each idle worker (for example
`cat /proc/<pid>/stack`
as root). One is blocked reading the pipe (`anon_pipe_read`); the others
wait on the lock
(`futex_do_wait`).
4. `kill -9` the worker that is in `anon_pipe_read`.
5. Trigger any DAG with one trivial task, for example `airflow dags trigger
<dag_id>`.
6. The task stays `queued` (I watched it for 180 s) while the scheduler
reports healthy. After
`docker compose restart airflow-scheduler` the same task runs.
Control: killing an idle worker that waits on the lock (`futex_do_wait`)
instead does not matter. The
next task ran within 5 s, because a replacement worker is forked when work
arrives.
**The same mechanism in plain Python, without Airflow**
```python
"""A SimpleQueue reader that dies while it holds the read lock.
python simplequeue_dead_reader.py holder # kill the reader that holds the
lock
python simplequeue_dead_reader.py waiter # control: kill one that waits
for it
"""
import multiprocessing as mp
import os
import signal
import sys
import time
def reader(q, name):
# what an idle LocalExecutor worker does: SimpleQueue.get()
print(name, "got", q.get(), flush=True)
if __name__ == "__main__":
mp.set_start_method("fork")
q = mp.SimpleQueue()
w0 = mp.Process(target=reader, args=(q, "w0"))
w0.start()
time.sleep(1) # w0 now holds q._rlock and blocks reading the pipe
rest = [mp.Process(target=reader, args=(q, f"w{i}")) for i in (1, 2, 3)]
for p in rest:
p.start() # these block on q._rlock
time.sleep(1)
victim = w0 if sys.argv[1] == "holder" else rest[0]
os.kill(victim.pid, signal.SIGKILL) # as the OOM killer would
victim.join()
q.put("job")
time.sleep(3)
taken = [p for p in [w0, *rest] if p is not victim and not p.is_alive()]
print("job taken by a live reader:", bool(taken))
for p in [w0, *rest]:
p.kill()
With `holder` it prints `job taken by a live reader: False`: the three live
readers never get the
job. With `waiter` the holder gets it and it prints `True`. I ran this on
macOS with Python 3.13.5
and 3.14.x (`fork`); I have not run it on Linux.
### What you think should happen instead?
The death of one worker should not stop the executor. After
`_check_workers()` reaps a dead worker it
cannot know whether that worker held `_rlock`, so the executor stays exposed
as long as the workers
share one lock-protected `SimpleQueue`. I have not prototyped a fix. Two
directions I can see:
- When a worker dies, re-create `activity_queue` and its workers (and
re-queue the workloads that
were in the old pipe and not yet read), instead of only replacing the one
process.
- Do not make idle workers share a lock-protected read end at all, for
example one pipe per worker
with the executor handing each workload to an idle worker.
Short of that, the state could at least be detected and logged: workloads
are outstanding
(`_unread_messages` > 0), the queue is not empty, and no worker has picked
anything up for a while,
instead of tasks sitting in `queued` while everything reports healthy. I
waited 180 s and not
`[scheduler] task_queued_timeout` (600 s by default), so I do not know what
the stuck-in-queued
handling would do with such a task, or whether it could recover while the
workers stay blocked.
### Operating System
Host: Ubuntu 26.04 LTS (arm64), kernel 7.0.0-30-generic. Container: the
official
`apache/airflow:3.3.2-python3.13` image (Debian-based; I did not record
`/etc/os-release` from
inside it).
### Deployment
Docker-Compose
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
Not relevant (core executor issue).
### Official Helm Chart version
Not Applicable
### Kubernetes Version
Not Applicable
### Helm Chart configuration
Not Applicable
### Docker Image customizations
`apache/airflow:3.3.2-python3.13` plus `dlt[postgres]` and
`astronomer-cosmos` (and a few small
dependencies) installed into Airflow's environment, a separate virtualenv
for dbt, and one Airflow
listener plugin that writes `/proc/self/oom_score_adj` of task processes.
None of it touches the
executor or its queues, but I did not repeat the test on an unmodified image.
### Anything else?
- I found this while testing which process the kernel's OOM killer should
prefer in a
memory-limited container (1280 MiB, `parallelism = 4`). SIGKILL is what
the OOM killer sends, and
the workers have the default `oom_score_adj` of 0 in my setup, so any of
them can be picked. I
reproduced the wedge with `kill -9` from the host; I did not catch the OOM
killer choosing a
lock-holding worker by itself.
- It needs a worker to die at the moment it holds the lock, and with idle
workers that is the one
that is blocked reading the pipe. I tested only the `fork` start method.
- `main` (4ca1f6693bb, 2026-10-11) still has the `SimpleQueue` pair and
`input.get()`
(`local_executor.py:170-171` and `:87`). Its `_check_workers()`
(`:186-198`) now also fails the
dead worker's task, but it does not touch the lock. I did not run `main`.
- Related but different failures of the same queues: #67881 and #70526 (a
full pipe blocks a
write), #67870 (unbounded `join()` after a worker's `BrokenPipeError`). I
did not find an
existing issue about a dead lock holder.
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
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]
