SakshamKapoor2911 opened a new issue, #70543:
URL: https://github.com/apache/airflow/issues/70543
#### Context
The LocalExecutor uses `multiprocessing.SimpleQueue` for both
`activity_queue` (scheduler → workers) and `result_queue` (workers →
scheduler). `SimpleQueue` is backed by an OS-level pipe with a ~64 KB kernel
buffer. When that buffer fills, the producer blocks in `put()`.
Two deadlocks have surfaced because of this:
- **#67881 (shutdown deadlock in `end()`):** Workers filled the
`result_queue` pipe, blocked on `put()`, and could not receive poison pills
from `activity_queue`. The scheduler's `proc.join()` blocked forever. Fixed by
draining the result queue during join.
- **#70526 (dispatch deadlock in `_process_workloads()`):** Same mechanism
during normal operation. Workers filled `result_queue` → blocked on `put()` →
stopped reading `activity_queue` → `activity_queue` pipe filled → scheduler
blocked on its own `put()`. Fix in progress: drain before each put.
(I authored #67881 and helped diagnose #70526, so I have been following both
paths.)
Both fixes use the same approach: call `_read_results()` before a
potentially-blocking operation. The spot-fixes work. But they share a pattern
(guard a specific `put()` call), and every future code path that touches either
queue would need the same treatment. I am not sure whether the current approach
is "good enough" or whether a more structural fix is warranted, so I wanted to
get input.
#### The Question
**Would a continuous background drain thread be a better long-term
approach?**
Instead of spot-fixing each blocking path, a lightweight daemon thread would
continuously move items from the `result_queue` OS pipe into a thread-safe
in-process `queue.Queue` buffer. The drain thread would **never** touch
application state (`event_buffer`, `running`, `change_state()`). Only the main
scheduler thread (via `sync()` / `_read_results()`) would process items from
that buffer into application state. All mutation stays single-threaded.
```
┌─ Drain Thread ────────┐ ┌─ Main Thread ──────────────┐ ┌─ Workers
────────┐
│ (daemon, isolated) │ │ (only thread that mutates │ │
│
│ │ │ event_buffer / running) │ │
│
│ result_queue ──get──→ │ │ │ │ workload
= │
│ (OS pipe, drained) │ │ _read_results(): │ │
activity.get() │
│ │ │ _drain_buffer ──get──→ │ │
│
│ _drain_buffer ──put─→ │ │ .change_state() │ │
result_queue │
│ (queue.Queue, safe) │ │ │ │
.put(state) ───┤
│ │ │ _process_workloads(): │ │
│
│ sleep 10ms / stop │ │ activity_queue.put() │ │ (never
blocks, │
│ (Event.wait) │ │ (safe, pipe drained │ │ pipe
always has │
└───────────────────────┘ │ continuously) │ │ room)
│
└────────────────────────────┘
└──────────────────┘
```
The drain thread is a pure pipe drainer. It only touches `result_queue`
(read) and `_drain_buffer` (write). All application-state mutation remains
single-threaded in the main scheduler loop. If the drain thread dies, `sync()`
catches up via the existing fallback drain path. Nothing is lost.
#### Trade-offs
| Pro | Con |
|-----|-----|
| Eliminates the precondition for both deadlock families (pipe can never
fill) | Introduces a daemon thread to the scheduler process |
| Removes per-workload `_read_results()` call (less overhead per dispatch) |
~55 lines of code + tests to review |
| Future `put()` call sites automatically safe, no audit needed | The
spot-fixes already work; this may be overkill |
| Only stdlib primitives (`threading`, `queue`), no new dependencies | Drain
thread is a new moving part in shutdown sequence |
| Drain thread never touches application state, single-threaded invariant
preserved | `queue.Queue` is unbounded; if main thread stalls, memory grows
(mitigated: main thread drains buffer every heartbeat) |
| Fork-safe; workers inherit nothing, only touch write end of pipe | Two
readers of `result_queue`, narrow TOCTOU window but only benign consequence
(brief block) |
| Can be implemented, tested, and removed independently of heartbeat loop |
Adds ~16 KB memory per executor instance |
#### Non-Goals
- Does not change the scheduler heartbeat loop
- Does not affect other executors (Celery, Kubernetes, etc.)
- Does not replace `SimpleQueue` with `Queue`
- No dependencies beyond stdlib
#### Alternative: Keep Spot-Fixes
The spot-fixes from #67881 and #70526 work. Workers that complete tasks
produce 1 to 3 queue items each. Even at 256 concurrent tasks × 3 items × ~200
bytes = ~150 KB per heartbeat window, the spot-fixes handle this fine. The only
risk is a future contributor adding a new `queue.put()` in a code path that
does not drain first, but this could be caught in code review.
If maintainers prefer this approach, that is perfectly valid. I just wanted
to surface the alternative before anyone files a third deadlock.
#### Questions for Maintainers
1. Is adding a daemon thread to the scheduler process acceptable in
principle, or is that a non-starter?
2. If the drain thread approach is acceptable, should the per-workload
`_read_results()` call from #70526 be removed (cleaner) or kept (defense in
depth)?
3. Is the complexity of a background thread worth eliminating the entire
deadlock class, or are the two spot-fixes sufficient?
Happy to implement this if there is appetite, or to drop it if the spot-fix
approach is preferred. No strong attachment either way. Just wanted to get
guidance before writing code.
--
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]
