Uh oh!
There was an error while loading. Please reload this page.
[Fix] serialize worker job submissions to preserve worker_id ordering - #935
Conversation
Worker threads were submitting Flux jobs concurrently, causing executorlib_worker_id to not correspond to the Flux scheduling order. This made worker_id unreliable for resource mapping (e.g. GPU assignment). Two changes: - Add threading.Event chain in BlockAllocationTaskScheduler so each worker waits for the previous worker to finish submitting before starting its own submission. - Call self._future.jobid() after FluxExecutor.submit() to block until the job is actually registered with the Flux broker, not just queued in the async FluxExecutor.
📝 WalkthroughWalkthroughAdds ordered per-worker boot-up synchronization using threading.Event objects and updates Changes
Sequence Diagram(s)sequenceDiagram
participant Scheduler as Scheduler (main)
participant WorkerA as Worker-0 thread
participant WorkerB as Worker-1 thread
participant FluxSpawner as FluxSpawner
participant Future as Future
Scheduler->>WorkerA: start thread with bootup_event A, next_bootup_event B
Scheduler->>WorkerB: start thread with bootup_event B, next_bootup_event C
WorkerA-->>Scheduler: bootup_event A.wait()
WorkerA->>WorkerA: create interface / submit tasks
WorkerA->>Scheduler: signal next_bootup_event B (set)
WorkerA->>FluxSpawner: submit jobspec
FluxSpawner->>Future: (if present) call jobid()
WorkerB-->>Scheduler: bootup_event B.wait()
WorkerB->>WorkerB: create interface / submit tasks
WorkerB->>Scheduler: signal next_bootup_event C (set)
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)
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 |
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/executorlib/task_scheduler/interactive/blockallocation.py (1)
254-267:⚠️ Potential issue | 🔴 CriticalDeadlock if
interface_bootupraises: subsequent workers hang forever.If
interface_bootup(line 256) throws an exception for worker N, the thread terminates without callingnext_bootup_event.set(). Workers N+1, N+2, … remain blocked onbootup_event.wait()indefinitely, andshutdown(wait=True)will hang onprocess.join().Wrap the bootup + signal in
try/finallyso the chain always progresses:🔒 Proposed fix using try/finally
if bootup_event is not None: bootup_event.wait() - interface = interface_bootup(- command_lst=get_interactive_execute_command(- cores=cores,- ),- connections=spawner(cores=cores, worker_id=worker_id, **kwargs),- hostname_localhost=hostname_localhost,- log_obj_size=log_obj_size,- worker_id=worker_id,- stop_function=stop_function,- )- if next_bootup_event is not None:- next_bootup_event.set()+ try:+ interface = interface_bootup(+ command_lst=get_interactive_execute_command(+ cores=cores,+ ),+ connections=spawner(cores=cores, worker_id=worker_id, **kwargs),+ hostname_localhost=hostname_localhost,+ log_obj_size=log_obj_size,+ worker_id=worker_id,+ stop_function=stop_function,+ )+ finally:+ if next_bootup_event is not None:+ next_bootup_event.set()Note: if bootup fails and the chain unblocks the next worker, subsequent workers will also likely fail. But that's preferable to a silent deadlock. You may also want to consider using
bootup_event.wait(timeout=...)as additional protection.🤖 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 254 - 267, The chain of bootstrapped workers can deadlock if interface_bootup raises because next_bootup_event.set() is skipped; modify the block around the call to interface_bootup (the code that waits on bootup_event, calls interface_bootup with get_interactive_execute_command and spawner, and then calls next_bootup_event.set()) to ensure next_bootup_event.set() is executed in a finally clause: keep the bootup_event.wait() before trying to create the interface, perform the interface_bootup call (using get_interactive_execute_command and spawner) inside a try, and always call next_bootup_event.set() in the finally so the next worker is unblocked even if interface_bootup raises; optionally re-raise or log the exception inside the try/except around interface_bootup if you want error visibility.
🧹 Nitpick comments (1)
src/executorlib/task_scheduler/interactive/blockallocation.py (1)
116-126: Dynamic worker addition skips boot-up ordering (andworker_id).The
max_workerssetter creates new threads using onlyself._process_kwargs, which doesn't includebootup_event,next_bootup_event, orworker_id. TheNonedefaults prevent a crash, but dynamically added workers won't have ordered boot-up or correctworker_idassignment.This is pre-existing behavior and not a regression from this PR, but worth noting if the setter is expected to be used in contexts where ordering matters.
🤖 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 116 - 126, The dynamic worker-add path in the max_workers setter builds new Thread instances using only self._process_kwargs which omits bootup_event, next_bootup_event, and worker_id, so newly added threads skip boot-up ordering and get wrong worker IDs; update the branch that creates new_process_lst (and references _execute_multiple_tasks and self._process_kwargs) to construct per-thread kwargs mirroring the initial worker creation: set worker_id = self._max_workers + i for each new thread, pass the correct bootup_event and next_bootup_event (chain them the same way the initial spawn logic does), and then start and append those threads to self._process so dynamically added workers follow the same boot ordering and ID assignment as existing workers.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@src/executorlib/task_scheduler/interactive/blockallocation.py`:
- Around line 254-267: The chain of bootstrapped workers can deadlock if
interface_bootup raises because next_bootup_event.set() is skipped; modify the
block around the call to interface_bootup (the code that waits on bootup_event,
calls interface_bootup with get_interactive_execute_command and spawner, and
then calls next_bootup_event.set()) to ensure next_bootup_event.set() is
executed in a finally clause: keep the bootup_event.wait() before trying to
create the interface, perform the interface_bootup call (using
get_interactive_execute_command and spawner) inside a try, and always call
next_bootup_event.set() in the finally so the next worker is unblocked even if
interface_bootup raises; optionally re-raise or log the exception inside the
try/except around interface_bootup if you want error visibility.
---
Nitpick comments:
In `@src/executorlib/task_scheduler/interactive/blockallocation.py`:
- Around line 116-126: The dynamic worker-add path in the max_workers setter
builds new Thread instances using only self._process_kwargs which omits
bootup_event, next_bootup_event, and worker_id, so newly added threads skip
boot-up ordering and get wrong worker IDs; update the branch that creates
new_process_lst (and references _execute_multiple_tasks and
self._process_kwargs) to construct per-thread kwargs mirroring the initial
worker creation: set worker_id = self._max_workers + i for each new thread, pass
the correct bootup_event and next_bootup_event (chain them the same way the
initial spawn logic does), and then start and append those threads to
self._process so dynamically added workers follow the same boot ordering and ID
assignment as existing workers.
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/executorlib/task_scheduler/interactive/blockallocation.py (1)
256-269:⚠️ Potential issue | 🔴 CriticalDeadlock if
interface_bootupraises an exception.If
interface_bootupthrows (e.g., spawner failure, connection timeout),next_bootup_event.set()on line 268 is never reached. Every subsequent worker will block forever onbootup_event.wait(), andshutdown(wait=True)will also hang becauseThread.join()blocks on those stuck threads.Wrap the boot-up + signal in
try/finallyso the chain always advances.🔒 Proposed fix
if bootup_event is not None: bootup_event.wait() - interface = interface_bootup(- command_lst=get_interactive_execute_command(- cores=cores,- ),- connections=spawner(cores=cores, worker_id=worker_id, **kwargs),- hostname_localhost=hostname_localhost,- log_obj_size=log_obj_size,- worker_id=worker_id,- stop_function=stop_function,- )- if next_bootup_event is not None:- next_bootup_event.set()+ try:+ interface = interface_bootup(+ command_lst=get_interactive_execute_command(+ cores=cores,+ ),+ connections=spawner(cores=cores, worker_id=worker_id, **kwargs),+ hostname_localhost=hostname_localhost,+ log_obj_size=log_obj_size,+ worker_id=worker_id,+ stop_function=stop_function,+ )+ finally:+ if next_bootup_event is not None:+ next_bootup_event.set()Note: after the
finallyblock, ifinterface_bootupdid raise, the exception will propagate and terminate this worker thread. The rest of the function (init_function, task loop) will be skipped — but at least the remaining workers won't deadlock.🤖 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 256 - 269, The boot-up chain can deadlock if interface_bootup (called with get_interactive_execute_command(...) and connections=spawner(...)) raises, because next_bootup_event.set() is never reached; wrap the call to interface_bootup and the subsequent signaling in a try/finally so that next_bootup_event.set() is always executed in the finally block (still allowing the exception to propagate and terminate this worker thread), ensuring bootup_event.wait()/next_bootup_event.set() chain always advances.
🧹 Nitpick comments (1)
src/executorlib/task_scheduler/interactive/blockallocation.py (1)
256-257:bootup_event.wait()is not interruptible byshutdown().If
shutdown()is called while workers are still queued behindbootup_event.wait(), those threads cannot observe the stop signal —stop_functionis only evaluated insideinterface_bootup, not here. With thetry/finallyfix above, the chain will unblock naturally once the currently booting worker finishes (or fails), so this is mostly a latency concern rather than a hard deadlock. However, for a fast-exit guarantee you could add a timeout loop:ifbootup_eventisnotNone: whilenotbootup_event.wait(timeout=1.0): ifstop_functionisnotNoneandstop_function(): returnThis is optional and can be deferred.
🤖 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 256 - 257, The current unconditional bootup_event.wait() call can block threads from observing shutdown; change it to a timeout loop that repeatedly waits with a short timeout and checks stop_function between waits so threads can return early on shutdown. Replace the direct call to bootup_event.wait() (in blockallocation where bootup_event is used alongside interface_bootup) with a loop that calls bootup_event.wait(timeout=...) and after each timeout calls stop_function() (if provided) and returns if it indicates shutdown; keep behavior of eventually proceeding when bootup_event is set. Ensure you reference bootup_event and stop_function and preserve existing cleanup/try/finally semantics around interface_bootup.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@src/executorlib/task_scheduler/interactive/blockallocation.py`:
- Around line 256-269: The boot-up chain can deadlock if interface_bootup
(called with get_interactive_execute_command(...) and connections=spawner(...))
raises, because next_bootup_event.set() is never reached; wrap the call to
interface_bootup and the subsequent signaling in a try/finally so that
next_bootup_event.set() is always executed in the finally block (still allowing
the exception to propagate and terminate this worker thread), ensuring
bootup_event.wait()/next_bootup_event.set() chain always advances.
---
Nitpick comments:
In `@src/executorlib/task_scheduler/interactive/blockallocation.py`:
- Around line 256-257: The current unconditional bootup_event.wait() call can
block threads from observing shutdown; change it to a timeout loop that
repeatedly waits with a short timeout and checks stop_function between waits so
threads can return early on shutdown. Replace the direct call to
bootup_event.wait() (in blockallocation where bootup_event is used alongside
interface_bootup) with a loop that calls bootup_event.wait(timeout=...) and
after each timeout calls stop_function() (if provided) and returns if it
indicates shutdown; keep behavior of eventually proceeding when bootup_event is
set. Ensure you reference bootup_event and stop_function and preserve existing
cleanup/try/finally semantics around interface_bootup.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@## main #935 +/- ##
==========================================
+ Coverage 94.07% 94.09% +0.02%
==========================================
Files 39 39 Lines 2041 2049 +8 ==========================================
+ Hits 1920 1928 +8
Misses 121 121 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
Uh oh!
There was an error while loading. Please reload this page.
Worker threads were submitting Flux jobs concurrently, causing
executorlib_worker_id to not correspond to the Flux scheduling order.
This made worker_id unreliable for resource mapping (e.g. GPU assignment).
Two changes:
Summary by CodeRabbit