[Fix] serialize worker job submissions to preserve worker_id ordering (#935)

* fix: serialize worker job submissions to preserve worker_id ordering

  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.

* Update blockallocation.py

* Update spawner_flux.py

* Update spawner_flux.py

---------

Co-authored-by: Ilgar Baghishov <ilgar@login15.chn.perlmutter.nersc.gov>
Co-authored-by: Jan Janssen <jan-janssen@users.noreply.github.com>

Author: 3 people

src/executorlib/task_scheduler/interactive/blockallocation.py

@@ -1,7 +1,7 @@
 import queue
 import random
 from concurrent.futures import Future
-from threading import Thread
+from threading import Event, Thread
 from typing import Callable, Optional
 
 from executorlib.standalone.command import get_interactive_execute_command
@@ -83,6 +83,8 @@ def __init__(
         self_id = random.getrandbits(128)
         self._self_id = self_id
         _interrupt_bootup_dict[self._self_id] = False
+        bootup_events = [Event() for _ in range(self._max_workers)]
+        bootup_events[0].set()
         self._set_process(
             process=[
                 Thread(
@@ -91,6 +93,12 @@ def __init__(
                     | {
                         "worker_id": worker_id,
                         "stop_function": lambda: _interrupt_bootup_dict[self_id],
+                        "bootup_event": bootup_events[worker_id],
+                        "next_bootup_event": (
+                            bootup_events[worker_id + 1]
+                            if worker_id + 1 < self._max_workers
+                            else None
+                        ),
                     },
                 )
                 for worker_id in range(self._max_workers)
@@ -217,6 +225,8 @@ def _execute_multiple_tasks(
     worker_id: int = 0,
     stop_function: Optional[Callable] = None,
     restart_limit: int = 0,
+    bootup_event: Optional[Event] = None,
+    next_bootup_event: Optional[Event] = None,
     **kwargs,
 ) -> None:
     """
@@ -245,7 +255,12 @@ def _execute_multiple_tasks(
                          distribution.
         stop_function (Callable): Function to stop the interface.
         restart_limit (int): The maximum number of restarting worker processes.
+        bootup_event (Event): Event to wait on before submitting the job to the scheduler, ensuring workers are
+                              submitted in worker_id order.
+        next_bootup_event (Event): Event to signal after job submission, unblocking the next worker.
     """
+    if bootup_event is not None:
+        bootup_event.wait()
     interface = interface_bootup(
         command_lst=get_interactive_execute_command(
             cores=cores,
@@ -256,6 +271,8 @@ def _execute_multiple_tasks(
         worker_id=worker_id,
         stop_function=stop_function,
     )
+    if next_bootup_event is not None:
+        next_bootup_event.set()
     interface_initialization_exception = _set_init_function(
         interface=interface,
         init_function=init_function,

src/executorlib/task_scheduler/interactive/spawner_flux.py

@@ -146,6 +146,8 @@ def bootup(
             )
         else:
             self._future = self._flux_executor.submit(jobspec=jobspec)
+        if self._future is not None:
+            self._future.jobid()
         return self.poll()
 
     def shutdown(self, wait: bool = True):