Fix dead worker draining shared task queue after restart_limit exhausted#946
Conversation
📝 WalkthroughWalkthroughAdds cooperative worker-death tracking to the block-allocation scheduler: an Changes
Sequence DiagramsequenceDiagram
participant Client
participant Worker
participant SharedState as "Shared State\n(alive_workers, lock)"
participant Queue as "Task Queue"
Client->>Queue: enqueue task (future)
Queue->>Worker: deliver task
Worker->>SharedState: check alive count (acquire lock)
SharedState-->>Worker: allowed / denied (release lock)
alt worker permanently dies / restart limit reached
Worker->>SharedState: decrement alive_workers (with lock)
SharedState-->>Worker: updated count
alt alive_workers > 0
Worker->>Queue: recycle queued tasks for other workers
else no healthy workers
Worker->>Queue: fail queued futures with ExecutorlibSocketError
end
else worker restarts normally
Worker->>Client: complete task future
end
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Tip Try Coding Plans. Let us write the prompt for your AI agent so you can ship faster (with fewer bugs). Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #946 +/- ##
==========================================
+ Coverage 94.10% 94.13% +0.02%
==========================================
Files 39 39
Lines 2054 2080 +26
==========================================
+ Hits 1933 1958 +25
- Misses 121 122 +1 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
|
@IlgarBaghishov Can you ask Claude to fix the issue? |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/executorlib/task_scheduler/interactive/blockallocation.py`:
- Around line 86-87: The liveness counter alive_workers created at startup isn’t
kept in sync when max_workers is changed, so newly spawned workers don’t receive
the shared counter and scaled-down removals don’t update the count; update the
max_workers setter (the method that adds/removes threads) to modify the shared
counter stored on the scheduler (self._alive_workers[0]) whenever workers are
intentionally added or removed, ensure you hold the existing alive_workers_lock
while updating, and pass the same alive_workers (and its Lock) into any newly
spawned worker threads (the code that creates startup worker threads) so all
workers share and update the same counter.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 89f3f5c5-b8c4-4a5b-8ae4-4d1eb520e1b3
📒 Files selected for processing (1)
src/executorlib/task_scheduler/interactive/blockallocation.py
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
src/executorlib/task_scheduler/interactive/blockallocation.py (1)
359-359: Simplify key check withdict.get().As flagged by static analysis (RUF019), the pattern
"shutdown" in task_dict and task_dict["shutdown"]can be simplified.♻️ Proposed fix
- if "shutdown" in task_dict and task_dict["shutdown"]: + if task_dict.get("shutdown"):🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/executorlib/task_scheduler/interactive/blockallocation.py` at line 359, Replace the explicit key-existence check "shutdown" in task_dict and task_dict["shutdown"] with the simpler dict.get() form to read truthiness in one step; locate the occurrence in blockallocation.py where the variable task_dict is inspected (the conditional using "shutdown") and change it to use task_dict.get("shutdown") so it returns False/None when missing and avoids the redundant membership test.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/executorlib/task_scheduler/interactive/blockallocation.py`:
- Around line 354-371: The loop uses a stale snapshot of has_healthy_workers
which can cause requeued tasks to be orphaned; modify the while True loop in
blockallocation.py to recompute/validate has_healthy_workers under the shared
lock before deciding to requeue: acquire the same lock used to manage
alive_workers, compute has_healthy_workers = (alive_workers is not None and
len(alive_workers) > 0) (or equivalent), and if it is now False, pop the
"future" from task_dict and call f.set_exception(ExecutorlibSocketError(...))
and task_done(...) instead of re-putting the task into future_queue; keep the
existing fallback for alive_workers is None behavior.
---
Nitpick comments:
In `@src/executorlib/task_scheduler/interactive/blockallocation.py`:
- Line 359: Replace the explicit key-existence check "shutdown" in task_dict and
task_dict["shutdown"] with the simpler dict.get() form to read truthiness in one
step; locate the occurrence in blockallocation.py where the variable task_dict
is inspected (the conditional using "shutdown") and change it to use
task_dict.get("shutdown") so it returns False/None when missing and avoids the
redundant membership test.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 2755c938-c8dd-4d0a-8c51-7b0f6158c8b0
📒 Files selected for processing (1)
src/executorlib/task_scheduler/interactive/blockallocation.py
| while True: | ||
| try: | ||
| task_dict = future_queue.get(timeout=1) | ||
| except queue.Empty: | ||
| continue | ||
| if "shutdown" in task_dict and task_dict["shutdown"]: | ||
| task_done(future_queue=future_queue) | ||
| break | ||
| elif "fn" in task_dict and "future" in task_dict: | ||
| if has_healthy_workers: | ||
| future_queue.put(task_dict) | ||
| task_done(future_queue=future_queue) | ||
| else: | ||
| f = task_dict.pop("future") | ||
| f.set_exception( | ||
| ExecutorlibSocketError("SocketInterface crashed during execution.") | ||
| ) | ||
| task_done(future_queue=future_queue) |
There was a problem hiding this comment.
Race condition can orphan tasks when all workers are dead.
When has_healthy_workers=True is captured at entry (Line 351), this value becomes stale if the remaining healthy workers subsequently die. A dead worker with stale has_healthy_workers=True will keep requeuing tasks, while the last dead worker (with has_healthy_workers=False) may consume its shutdown message and exit before processing all requeued items.
Example scenario with 2 workers both exhausting restarts:
- Worker A enters drain first:
alive_workers2→1,has_healthy_workers=True - Worker B enters drain:
alive_workers1→0,has_healthy_workers=False - Worker A requeues a task, Worker B fails another task
- Worker B gets its shutdown message and exits
- Worker A gets its shutdown message and exits
- The task requeued by A after B exited is orphaned (future never completes)
Consider re-checking has_healthy_workers under the lock on each iteration, or having workers with has_healthy_workers=True fail tasks if they detect they're the last active thread in the drain.
💡 Suggested fix direction
while True:
try:
task_dict = future_queue.get(timeout=1)
except queue.Empty:
continue
- if "shutdown" in task_dict and task_dict["shutdown"]:
+ if task_dict.get("shutdown"):
task_done(future_queue=future_queue)
break
elif "fn" in task_dict and "future" in task_dict:
- if has_healthy_workers:
+ # Re-check under lock to avoid stale reads
+ with alive_workers_lock:
+ current_healthy = alive_workers[0] > 0 if alive_workers else False
+ if current_healthy:
future_queue.put(task_dict)
task_done(future_queue=future_queue)
else:This requires the lock and list to be non-None in this path, which they will be for initial workers. For the fallback case (alive_workers is None), the current behavior (fail immediately) is correct.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| while True: | |
| try: | |
| task_dict = future_queue.get(timeout=1) | |
| except queue.Empty: | |
| continue | |
| if "shutdown" in task_dict and task_dict["shutdown"]: | |
| task_done(future_queue=future_queue) | |
| break | |
| elif "fn" in task_dict and "future" in task_dict: | |
| if has_healthy_workers: | |
| future_queue.put(task_dict) | |
| task_done(future_queue=future_queue) | |
| else: | |
| f = task_dict.pop("future") | |
| f.set_exception( | |
| ExecutorlibSocketError("SocketInterface crashed during execution.") | |
| ) | |
| task_done(future_queue=future_queue) | |
| while True: | |
| try: | |
| task_dict = future_queue.get(timeout=1) | |
| except queue.Empty: | |
| continue | |
| if task_dict.get("shutdown"): | |
| task_done(future_queue=future_queue) | |
| break | |
| elif "fn" in task_dict and "future" in task_dict: | |
| # Re-check under lock to avoid stale reads | |
| with alive_workers_lock: | |
| current_healthy = alive_workers[0] > 0 if alive_workers else False | |
| if current_healthy: | |
| future_queue.put(task_dict) | |
| task_done(future_queue=future_queue) | |
| else: | |
| f = task_dict.pop("future") | |
| f.set_exception( | |
| ExecutorlibSocketError("SocketInterface crashed during execution.") | |
| ) | |
| task_done(future_queue=future_queue) |
🧰 Tools
🪛 Ruff (0.15.4)
[warning] 359-359: Unnecessary key check before dictionary access
Replace with dict.get
(RUF019)
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/executorlib/task_scheduler/interactive/blockallocation.py` around lines
354 - 371, The loop uses a stale snapshot of has_healthy_workers which can cause
requeued tasks to be orphaned; modify the while True loop in blockallocation.py
to recompute/validate has_healthy_workers under the shared lock before deciding
to requeue: acquire the same lock used to manage alive_workers, compute
has_healthy_workers = (alive_workers is not None and len(alive_workers) > 0) (or
equivalent), and if it is now False, pop the "future" from task_dict and call
f.set_exception(ExecutorlibSocketError(...)) and task_done(...) instead of
re-putting the task into future_queue; keep the existing fallback for
alive_workers is None behavior.
…kers race condition
|
@jan-janssen Fixed both "Too many statements (56 > 50)" and "Race condition can orphan tasks when all workers are dead." |
There was a problem hiding this comment.
♻️ Duplicate comments (1)
src/executorlib/task_scheduler/interactive/blockallocation.py (1)
86-87:⚠️ Potential issue | 🟠 MajorPersist the shared liveness state on the scheduler.
alive_workers/alive_workers_lockonly exist as locals here, so themax_workerssetter later spawns threads fromself._process_kwargswithout the same shared state and never updates the count on intentional scale-downs. That leaves_drain_dead_worker()making recycle-vs-fail decisions from stale membership, which can still cause endless requeueing or premature task failure after resizing.Also applies to: 104-105
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/executorlib/task_scheduler/interactive/blockallocation.py` around lines 86 - 87, The alive_workers and alive_workers_lock are defined as local variables so other methods/threads (e.g., the max_workers setter which spawns threads via self._process_kwargs and _drain_dead_worker) cannot share updated liveness state, causing stale membership and incorrect drain/requeue behavior; refactor by promoting alive_workers and alive_workers_lock to scheduler instance attributes (e.g., self.alive_workers, self.alive_workers_lock), update all references (including where threads are spawned in the max_workers setter and in _drain_dead_worker) to use these instance-level symbols, and ensure any thread creation receives self (or the instance attributes) rather than creating independent locals so scale-downs correctly update the shared liveness state.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@src/executorlib/task_scheduler/interactive/blockallocation.py`:
- Around line 86-87: The alive_workers and alive_workers_lock are defined as
local variables so other methods/threads (e.g., the max_workers setter which
spawns threads via self._process_kwargs and _drain_dead_worker) cannot share
updated liveness state, causing stale membership and incorrect drain/requeue
behavior; refactor by promoting alive_workers and alive_workers_lock to
scheduler instance attributes (e.g., self.alive_workers,
self.alive_workers_lock), update all references (including where threads are
spawned in the max_workers setter and in _drain_dead_worker) to use these
instance-level symbols, and ensure any thread creation receives self (or the
instance attributes) rather than creating independent locals so scale-downs
correctly update the shared liveness state.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 5740a1cf-1d8a-4bd1-add1-c0b1d621cd91
📒 Files selected for processing (1)
src/executorlib/task_scheduler/interactive/blockallocation.py
|
@IlgarBaghishov Could you also add a unit test in particular to the |
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@tests/unit/task_scheduler/interactive/test_blockallocation.py`:
- Around line 12-33: The test only covers the branch where _drain_dead_worker
decrements alive_workers to 0 and fails the task; add a new unit test that
exercises the healthy-worker requeue path by providing alive_workers with more
than one entry (e.g., alive_workers = [1, 2]) and verifying that when
_drain_dead_worker encounters a shutdown sentinel it re-queues the task instead
of failing it: call _drain_dead_worker with that setup, then assert the task is
either requeued in future_queue (or that future.result() returns the expected
value) and that alive_workers was decremented appropriately for the dead worker;
reference the _drain_dead_worker helper and the future/
future_queue/future.result checks to locate where to add the test.
- Around line 31-33: The assertion awaiting the future currently calls
future.result() without a timeout; update the test that expects
ExecutorlibSocketError to call future.result with a short timeout (e.g., pass
timeout= a small number) so the test fails fast if _drain_dead_worker() does not
set the exception; locate the assertion around future.result() in
tests/unit/task_scheduler/interactive/test_blockallocation.py and modify that
call to include a timeout and still assert that ExecutorlibSocketError is
raised.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: dd01bab9-0bdc-438b-bd64-ac4b708fc9d5
📒 Files selected for processing (1)
tests/unit/task_scheduler/interactive/test_blockallocation.py
| def test_fail_tasks_when_no_workers_remain(self): | ||
| future_queue = queue.Queue() | ||
| alive_workers = [1] | ||
| alive_workers_lock = Lock() | ||
| future = Future() | ||
|
|
||
| # Add a task and then the shutdown sentinel | ||
| future_queue.put({"fn": lambda: 42, "future": future}) | ||
| future_queue.put({"shutdown": True}) | ||
|
|
||
| _drain_dead_worker( | ||
| future_queue=future_queue, | ||
| alive_workers=alive_workers, | ||
| alive_workers_lock=alive_workers_lock, | ||
| ) | ||
|
|
||
| # Worker count should be decremented | ||
| self.assertEqual(alive_workers[0], 0) | ||
|
|
||
| # Task should fail with ExecutorlibSocketError | ||
| with self.assertRaises(ExecutorlibSocketError): | ||
| future.result() |
There was a problem hiding this comment.
Add a test for the healthy-worker requeue path.
This only covers the alive_workers -> 0 branch. The regression fixed in src/executorlib/task_scheduler/interactive/blockallocation.py:335-374 was the case where a dead worker must put work back when another worker is still alive, and that behavior is still untested here. Without that case, the original queue-drain bug can come back while this file still passes.
🧪 Suggested follow-up test
class TestDrainDeadWorker(unittest.TestCase):
+ def test_requeue_tasks_when_other_workers_remain(self):
+ future_queue = queue.Queue()
+ alive_workers = [2]
+ alive_workers_lock = Lock()
+ future = Future()
+ task = {"fn": lambda: 42, "future": future}
+
+ future_queue.put(task)
+ future_queue.put({"shutdown": True})
+
+ _drain_dead_worker(
+ future_queue=future_queue,
+ alive_workers=alive_workers,
+ alive_workers_lock=alive_workers_lock,
+ )
+
+ self.assertEqual(alive_workers[0], 1)
+ self.assertFalse(future.done())
+ self.assertEqual(future_queue.get_nowait(), task)
+ with self.assertRaises(queue.Empty):
+ future_queue.get_nowait()
+
def test_fail_tasks_when_no_workers_remain(self):
future_queue = queue.Queue()
alive_workers = [1]📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def test_fail_tasks_when_no_workers_remain(self): | |
| future_queue = queue.Queue() | |
| alive_workers = [1] | |
| alive_workers_lock = Lock() | |
| future = Future() | |
| # Add a task and then the shutdown sentinel | |
| future_queue.put({"fn": lambda: 42, "future": future}) | |
| future_queue.put({"shutdown": True}) | |
| _drain_dead_worker( | |
| future_queue=future_queue, | |
| alive_workers=alive_workers, | |
| alive_workers_lock=alive_workers_lock, | |
| ) | |
| # Worker count should be decremented | |
| self.assertEqual(alive_workers[0], 0) | |
| # Task should fail with ExecutorlibSocketError | |
| with self.assertRaises(ExecutorlibSocketError): | |
| future.result() | |
| def test_requeue_tasks_when_other_workers_remain(self): | |
| future_queue = queue.Queue() | |
| alive_workers = [2] | |
| alive_workers_lock = Lock() | |
| future = Future() | |
| task = {"fn": lambda: 42, "future": future} | |
| future_queue.put(task) | |
| future_queue.put({"shutdown": True}) | |
| _drain_dead_worker( | |
| future_queue=future_queue, | |
| alive_workers=alive_workers, | |
| alive_workers_lock=alive_workers_lock, | |
| ) | |
| self.assertEqual(alive_workers[0], 1) | |
| self.assertFalse(future.done()) | |
| self.assertEqual(future_queue.get_nowait(), task) | |
| with self.assertRaises(queue.Empty): | |
| future_queue.get_nowait() | |
| def test_fail_tasks_when_no_workers_remain(self): | |
| future_queue = queue.Queue() | |
| alive_workers = [1] | |
| alive_workers_lock = Lock() | |
| future = Future() | |
| # Add a task and then the shutdown sentinel | |
| future_queue.put({"fn": lambda: 42, "future": future}) | |
| future_queue.put({"shutdown": True}) | |
| _drain_dead_worker( | |
| future_queue=future_queue, | |
| alive_workers=alive_workers, | |
| alive_workers_lock=alive_workers_lock, | |
| ) | |
| # Worker count should be decremented | |
| self.assertEqual(alive_workers[0], 0) | |
| # Task should fail with ExecutorlibSocketError | |
| with self.assertRaises(ExecutorlibSocketError): | |
| future.result() |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tests/unit/task_scheduler/interactive/test_blockallocation.py` around lines
12 - 33, The test only covers the branch where _drain_dead_worker decrements
alive_workers to 0 and fails the task; add a new unit test that exercises the
healthy-worker requeue path by providing alive_workers with more than one entry
(e.g., alive_workers = [1, 2]) and verifying that when _drain_dead_worker
encounters a shutdown sentinel it re-queues the task instead of failing it: call
_drain_dead_worker with that setup, then assert the task is either requeued in
future_queue (or that future.result() returns the expected value) and that
alive_workers was decremented appropriately for the dead worker; reference the
_drain_dead_worker helper and the future/ future_queue/future.result checks to
locate where to add the test.
| # Task should fail with ExecutorlibSocketError | ||
| with self.assertRaises(ExecutorlibSocketError): | ||
| future.result() |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n tests/unit/task_scheduler/interactive/test_blockallocation.py | head -50Repository: pyiron/executorlib
Length of output: 1864
🏁 Script executed:
find . -name "test_blockallocation.py" -type fRepository: pyiron/executorlib
Length of output: 125
🏁 Script executed:
wc -l tests/unit/task_scheduler/interactive/test_blockallocation.pyRepository: pyiron/executorlib
Length of output: 126
🏁 Script executed:
cat -n tests/unit/task_scheduler/interactive/test_blockallocation.pyRepository: pyiron/executorlib
Length of output: 1417
🏁 Script executed:
find . -name "blockallocation.py" -type f | grep -v testRepository: pyiron/executorlib
Length of output: 125
🏁 Script executed:
cat -n executorlib/task_scheduler/interactive/blockallocation.pyRepository: pyiron/executorlib
Length of output: 151
🏁 Script executed:
cat -n src/executorlib/task_scheduler/interactive/blockallocation.pyRepository: pyiron/executorlib
Length of output: 20515
Add a timeout to the future assertion for test robustness.
While _drain_dead_worker() correctly completes the future via f.set_exception(), a timeout safeguards against regressions where this path might be skipped, which would cause the test to hang indefinitely instead of failing fast.
Suggested change
# Task should fail with ExecutorlibSocketError
with self.assertRaises(ExecutorlibSocketError):
- future.result()
+ future.result(timeout=1)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # Task should fail with ExecutorlibSocketError | |
| with self.assertRaises(ExecutorlibSocketError): | |
| future.result() | |
| # Task should fail with ExecutorlibSocketError | |
| with self.assertRaises(ExecutorlibSocketError): | |
| future.result(timeout=1) |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@tests/unit/task_scheduler/interactive/test_blockallocation.py` around lines
31 - 33, The assertion awaiting the future currently calls future.result()
without a timeout; update the test that expects ExecutorlibSocketError to call
future.result with a short timeout (e.g., pass timeout= a small number) so the
test fails fast if _drain_dead_worker() does not set the exception; locate the
assertion around future.result() in
tests/unit/task_scheduler/interactive/test_blockallocation.py and modify that
call to include a timeout and still assert that ExecutorlibSocketError is
raised.
2 participants
Problem
Two bugs in
_execute_multiple_tasks()inblockallocation.py:Bug 1: Dead worker drains the shared task queue
When
block_allocation=Trueand a worker process exhausts itsrestart_limit, the permanently dead worker thread setsinterface.status = True(faking that the worker is alive) and stores anExecutorlibSocketErrorasinterface_initialization_exception, but continues thewhile Trueloop. On the next iteration, it falls into theelsebranch (sinceinterface.statusis nowTrue), callsfuture_queue.get(), and immediately fails the task with the stored exception. This repeats indefinitely.Since the dead worker "processes" tasks instantly (just sets an exception on the future), it pulls from the shared queue orders of magnitude faster than healthy workers doing real computation. A single dead worker effectively drains the entire queue, starving all healthy workers.
Evidence from a real run: 12 workers (2 nodes x 2 GPUs x 6 jobs/GPU), 159,782 total tasks. One worker crashed and exhausted its
restart_limit. It drained the entire shared queue — each of the 11 healthy workers only completed ~1 task before the queue was empty. All 159,782 futures were reported as "FINISHED" (nearly all were errors from the dead worker).Bug 2: Off-by-one in restart_limit
The condition was
restart_counter > restart_limit, butrestart_counteris incremented afterbootup(). So withrestart_limit=2:> 2?restart_limit=Nwas allowingN+1restarts.restart_limit=0was allowing 1 restart instead of none.Fix
Fix 1: Dead worker recycling
When a worker is permanently dead (
restart_counter >= restart_limit), instead of faking the interface as alive and continuing the main loop, it now enters a dedicated inner loop where it:queue.join()hang inshutdown())A shared
alive_workerscounter (a single-element list[N]for pass-by-reference) protected by athreading.Locktracks how many worker threads are still alive. Each permanently dead worker atomically decrements this counter and checks whether healthy workers remain.Why not just
break?shutdown()sends one{"shutdown": True}message per worker thread, then callsqueue.join(). If a dead thread exits without consuming its shutdown message,queue.join()waits forever (unfinished_tasksnever reaches 0). The fix ensures each thread — alive or dead — consumes exactly one shutdown message.Why the counter?
Shutdown messages and real tasks share the same queue, so the dead worker must keep calling
queue.get()to find its shutdown message. When it pulls a real task, it must decide: recycle (healthy workers exist to pick it up) or fail (all workers dead, prevent infinite bouncing). Thealive_workerscounter is the minimal mechanism to make this decision correctly in all cases.The new parameters (
alive_workers,alive_workers_lock) areOptionalwithNonedefaults for backward compatibility. WhenNone, tasks are failed immediately (safe fallback).Fix 2: Off-by-one
Changed
restart_counter > restart_limittorestart_counter >= restart_limit. Nowrestart_limit=Nmeans exactly N restarts.Note: This is a behavior change — users who previously set
restart_limit=0and got 1 restart will now get 0 restarts. This is the correct/expected semantic.Test plan
Tested on TACC Vista HPC (2 nodes, Flux scheduler):
restart_limit-> all 20 tasks completed by the remaining 3 healthy workers (dead worker's tasks recycled)ExecutorlibSocketError, no hangrestart_counter >= restart_limitrestart_limitrestartsSummary by CodeRabbit
Bug Fixes
Documentation
Tests