[potiuk]

related: #70526, #70545

#70545 drains the result queue once before each activity_queue.put(), but the put itself can still block forever. A worker can write a result between the drain and the put; if the activity pipe is full at that moment the put blocks, and the worker keeps reading workloads and writing results until the result pipe is full too, so each side waits on the other. Linux pipes free space only a whole page at a time, so this shows on loaded Linux CI runners (the new test_process_workloads_progresses_under_concurrent_result_refill hung at its 30s limit in job 117757985706 on an unrelated PR) and not on macOS.

The put now runs on a short-lived helper thread while the main thread keeps reading results until it completes, so the result pipe cannot 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. If the dispatch is interrupted (SIGTERM raises SystemExit), draining continues until the pending put completes before re-raising, so end() does not queue its shutdown messages behind an abandoned writer. If no worker is alive the interruption is re-raised at once, since nothing could take the pending write and end() sends nothing.

A workload is counted and registered as running before its put; if the put itself fails, that bookkeeping is undone and the workload stays queued, as before. While a put is blocked and fewer than parallelism workers are alive (the spawn start method has none before the first batch and starts one per check, or workers died), workers are spawned from the drain loop, since only a live worker can take the pending write. On SIGTERM during a blocked put the executor now waits until a live worker has made room, consistent with end() waiting for running tasks; a second signal still interrupts.

Trade-offs worth a look from executor maintainers:

• one thread start per dispatched workload; the worst case for dispatching 200 workloads went from about 8 ms to about 50 ms,
• end() still sends its shutdown messages with a plain blocking put; left as is to keep this change focused,
• an mp.Queue (background writer thread, risky to fork workers under) or non-blocking pipe writes (multiprocessing internals) were considered and rejected.

This could not be reproduced on macOS, where pipes free space byte by byte; the Linux behaviour was checked against a model of the kernel's pipe rules, where the previous code still deadlocked in 50-97% of interleavings with a lagging worker and the new code in none.

---

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

• Yes — Claude Code (Opus 5.5)

Generated-by: Claude Code (Opus 5.5) following the guidelines

[potiuk]

A question for executor maintainers before this leaves draft: this starts a short-lived thread for every workload put, and the edge cases around it (interruption, no live workers, failed puts) needed handling of their own.

An alternative would be one long-running feeder thread per executor that owns all writes to the activity queue: _process_workloads would hand workloads to it through an in-process queue and keep draining results while it is busy. That removes the per-put thread start and the 50 ms polling, and most of the interruption handling moves into one place (stopping the feeder in end()).

The cost is a thread living for the executor's whole lifetime, which needs care around forking workers while it runs. Would the long-running thread be the better approach here, or is the per-put thread acceptable for the volumes LocalExecutor handles?

---

Drafted-by: Claude Code (Opus 5.5); reviewed by @potiuk before posting

[FrankYang0529]

The test case test_process_workloads_undoes_dispatch_when_put_fails only covers a key with no earlier dispatch, so the test never reaches the remaining > 0 branch of _undo_dispatch(). How about changing the test case like following?

    @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

[o-nikolas]

I'd keep the per-put thread. A long-lived feeder would be alive whenever a
replacement worker is forked, so every such fork would happen in a
multi-threaded process. With the per-put thread that only overlaps when a put
is blocked and a worker has died. ~50 ms per 200 workloads seems fine for
LocalExecutor volumes.
+1 to @FrankYang0529's suggestion to cover the remaining > 0 branch of
_undo_dispatch; with that, LGTM.

---

Drafted-by: Kiro (claude-opus-5.5); reviewed by @o-nikolas before posting