fix bug in hung processes / timeout

Author: sbryngelson

.github/workflows/bench.yml

@@ -101,6 +101,9 @@ jobs:
         run: |
           set -e
           
+          # Capture path to monitoring script from PR checkout (it doesn't exist in master)
+          MONITOR_SCRIPT="$(pwd)/pr/.github/scripts/monitor_slurm_job.sh"
+          
           # Function to submit and monitor using extracted script
           submit_and_monitor() {
             local dir=$1
@@ -124,8 +127,8 @@ jobs:
               return 1
             fi
             
-            # Use the monitoring script
-            bash .github/scripts/monitor_slurm_job.sh "$job_id" "$output_file"
+            # Use the monitoring script from PR checkout
+            bash "$MONITOR_SCRIPT" "$job_id" "$output_file"
           }
           
           # Run both jobs with monitoring

toolchain/mfc/test/case.py

@@ -123,7 +123,7 @@ def __init__(self, trace: str, mods: dict, ppn: int = None, override_tol: float
         self.override_tol = override_tol
         super().__init__({**BASE_CFG.copy(), **mods})
 
-    def run(self, targets: List[Union[str, MFCTarget]], gpus: Set[int]) -> subprocess.CompletedProcess:
+    def run(self, targets: List[Union[str, MFCTarget]], gpus: Set[int], timeout: Optional[float] = None) -> subprocess.CompletedProcess:
         if gpus is not None and len(gpus) != 0:
             gpus_select = ["--gpus"] + [str(_) for _ in gpus]
         else:
@@ -146,7 +146,7 @@ def run(self, targets: List[Union[str, MFCTarget]], gpus: Set[int]) -> subproces
             *jobs, "-t", *target_names, *gpus_select, *ARG("--")
         ]
 
-        return common.system(command, print_cmd=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        return common.system(command, print_cmd=False, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout)
 
     def get_trace(self) -> str:
         return self.trace

toolchain/mfc/test/test.py

@@ -1,4 +1,4 @@
-import os, typing, shutil, time, itertools, threading
+import os, typing, shutil, time, itertools, threading, subprocess
 from random import sample, seed
 
 import rich, rich.table
@@ -36,9 +36,6 @@
 # from worker threads which could leave the scheduler in an undefined state.
 abort_tests = threading.Event()
 
-class TestTimeoutError(MFCException):
-    pass
-
 # pylint: disable=too-many-branches, trailing-whitespace
 def __filter(cases_) -> typing.List[TestCase]:
     cases = cases_[:]
@@ -309,32 +306,16 @@ def _handle_case(case: TestCase, devices: typing.Set[int]):
     global current_test_number
     start_time = time.time()
 
-    # Set timeout using threading.Timer (works in worker threads)
-    # Note: we intentionally do not use signal.alarm() here because signals
-    # only work in the main thread; sched.sched runs tests in worker threads.
-    # threading.Timer works correctly in this threaded context.
-    timeout_flag = threading.Event()
-    timeout_timer = threading.Timer(TEST_TIMEOUT_SECONDS, timeout_flag.set)
-    timeout_timer.start()
-
     tol = case.compute_tolerance()
     case.delete_output()
     case.create_directory()
 
     if ARG("dry_run"):
         cons.print(f"  [bold magenta]{case.get_uuid()}[/bold magenta]     SKIP     {case.trace}")
-        timeout_timer.cancel()
         return
 
     try:
-        # Check timeout before starting
-        if timeout_flag.is_set():
-            raise TestTimeoutError("Test case exceeded 1 hour timeout")
-        cmd = case.run([PRE_PROCESS, SIMULATION], gpus=devices)
-
-        # Check timeout after simulation
-        if timeout_flag.is_set():
-            raise TestTimeoutError("Test case exceeded 1 hour timeout")
+        cmd = case.run([PRE_PROCESS, SIMULATION], gpus=devices, timeout=TEST_TIMEOUT_SECONDS)
 
         out_filepath = os.path.join(case.get_dirpath(), "out_pre_sim.txt")
 
@@ -380,13 +361,9 @@ def _handle_case(case: TestCase, devices: typing.Set[int]):
             # Clean up output from golden comparison run
             case.delete_output()
 
-            # Check timeout before full pipeline
-            if timeout_flag.is_set():
-                raise TestTimeoutError("Test case exceeded 1 hour timeout")
-            
             # Run full pipeline: this ensures simulation and post_process use
             # consistent configuration (parallel_io, file_per_process, etc.)
-            cmd = case.run([PRE_PROCESS, SIMULATION, POST_PROCESS], gpus=devices)
+            cmd = case.run([PRE_PROCESS, SIMULATION, POST_PROCESS], gpus=devices, timeout=TEST_TIMEOUT_SECONDS)
 
             out_post_filepath = os.path.join(case.get_dirpath(), "out_post.txt")
             common.file_write(out_post_filepath, cmd.stdout)
@@ -419,7 +396,8 @@ def _handle_case(case: TestCase, devices: typing.Set[int]):
         progress_str = f"({current_test_number:3d}/{total_test_count:3d})"
         cons.print(f"  {progress_str}    [bold magenta]{case.get_uuid()}[/bold magenta]    {duration:6.2f}    {case.trace}")
 
-    except TestTimeoutError as exc:
+    except subprocess.TimeoutExpired as exc:
+        # Subprocess timeout - the process was actually killed
         log_path = os.path.join(case.get_dirpath(), 'out_pre_sim.txt')
         if os.path.exists(log_path):
             log_msg = f"Check the log at: {log_path}"
@@ -428,12 +406,10 @@ def _handle_case(case: TestCase, devices: typing.Set[int]):
                 f"Log file ({log_path}) may not exist if the timeout occurred early."
             )
         raise MFCException(
-            f"Test {case} exceeded 1 hour timeout. "
+            f"Test {case} exceeded 1 hour timeout (process was killed). "
             f"This may indicate a hung simulation or misconfigured case. "
             f"{log_msg}"
         ) from exc
-    finally:
-        timeout_timer.cancel()  # Cancel timeout timer
 
 
 def handle_case(case: TestCase, devices: typing.Set[int]):