Should we replace spot-fixes with a background drain thread in LocalExecutor?

[SakshamKapoor2911]

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:

• Drain result_queue in LocalExecutor to prevent process join deadlock #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.

• Scheduler deadlocks in pipe write during task dispatch (LocalExecutor) — runtime variant of #67881 #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 Scheduler deadlocks in pipe write during task dispatch (LocalExecutor) — runtime variant of #67881 #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.