Auto-adjust timeout based on actual durations

Automatically reduce the "--timeout" value when actual jobs are executed
quicker than its value. Because timings vary hugely in practice, we
multiply measured durations by 10x.

The auto-adjustment can save a lot of time during late stages of the reduction,
when the interestingness test finishes quickly normally, but some unsuccessful
reductions may hang (something we observe happening a lot in practice with
C/C++ compilers).

Author: emaxx-google

cvise-cli.py

@@ -237,6 +237,11 @@ def main():
         default=300,
         help='Interestingness test timeout in seconds',
     )
+    parser.add_argument(
+        '--no-auto-adjust-timeout',
+        type=bool,
+        help='Disable automatic timeout calculation based on actual execution durations.',
+    )
     parser.add_argument('--no-cache', action='store_true', help="Don't cache behavior of passes")
     parser.add_argument(
         '--skip-key-off',
@@ -471,6 +476,7 @@ def do_reduce(args):
             args.start_with_pass,
             args.skip_after_n_transforms,
             args.stopping_threshold,
+            args.no_auto_adjust_timeout,
         ) as test_manager:
             reducer = CVise(test_manager, args.skip_interestingness_test_check)
 

cvise/CMakeLists.txt

@@ -141,6 +141,7 @@ set(SOURCE_FILES
   "utils/readkey_posix.py"
   "utils/readkey_windows.py"
   "utils/readkey.py"
+  "utils/resource.py"
   "utils/sigmonitor.py"
   "utils/statistics.py"
   "utils/testing.py"

cvise/tests/test_test_manager.py

@@ -304,6 +304,7 @@ def manager(tmp_path: Path, input_path: Path, interestingness_script: str, job_t
     START_WITH_PASS = None
     SKIP_AFTER_N_TRANSFORMS = None
     STOPPING_THRESHOLD = 1.0
+    NO_AUTO_ADJUST_TIMEOUT = False
     pass_statistic = statistics.PassStatistic()
 
     script_path = tmp_path / 'check.sh'
@@ -328,6 +329,7 @@ def manager(tmp_path: Path, input_path: Path, interestingness_script: str, job_t
         START_WITH_PASS,
         SKIP_AFTER_N_TRANSFORMS,
         STOPPING_THRESHOLD,
+        NO_AUTO_ADJUST_TIMEOUT,
     )
     test_manager.__enter__()
     try:

cvise/utils/resource.py

@@ -0,0 +1,25 @@
+import time
+from collections import deque
+
+
+class TimeoutEstimator:
+    """Estimates how long a job's timeout should be set, given the recent durations."""
+
+    _HISTORY_LEN = 30
+    _MEASUREMENT_LOWER_BOUND = 1  # seconds
+    _MEASUREMENTS_MULTIPLIER = 10
+
+    def __init__(self, initial_timeout: float):
+        self._initial_timeout = initial_timeout
+        self._recent_durations: deque[float] = deque(maxlen=self._HISTORY_LEN)
+
+    def update(self, start_time: float) -> None:
+        duration = time.monotonic() - start_time
+        self._recent_durations.append(duration)
+
+    def estimate(self) -> float:
+        if not self._recent_durations:
+            return self._initial_timeout  # not enough stats
+        worst = max(self._recent_durations)
+        estimation = self._MEASUREMENTS_MULTIPLIER * max(worst, self._MEASUREMENT_LOWER_BOUND)
+        return min(estimation, self._initial_timeout)

cvise/utils/testing.py

@@ -40,6 +40,7 @@
 from cvise.utils.hint import is_special_hint_type, load_hints
 from cvise.utils.process import MPContextHook, MPTaskLossWorkaround, ProcessEventNotifier, ProcessMonitor
 from cvise.utils.readkey import KeyLogger
+from cvise.utils.resource import TimeoutEstimator
 
 MAX_PASS_INCREASEMENT_THRESHOLD = 3
 
@@ -338,6 +339,15 @@ def should_reinit_after_test_case_update(self, other_contexts: Sequence[PassCont
         return False
 
 
+@dataclass
+class PassGlobalInfo:
+    init_job_timeout_estimator: TimeoutEstimator
+
+    @staticmethod
+    def create(starting_init_job_timeout: float) -> PassGlobalInfo:
+        return PassGlobalInfo(init_job_timeout_estimator=TimeoutEstimator(starting_init_job_timeout))
+
+
 @unique
 class JobType(Enum):
     INIT = auto()
@@ -425,7 +435,7 @@ def __init__(
         self,
         pass_statistic,
         test_script: Path,
-        timeout,
+        user_specified_timeout,
         save_temps,
         test_cases: list[Path],
         parallel_tests,
@@ -440,9 +450,10 @@ def __init__(
         start_with_pass,
         skip_after_n_transforms,
         stopping_threshold,
+        no_auto_adjust_timeout: bool,
     ):
         self.test_script: Path = test_script.absolute()
-        self.timeout = timeout
+        self.user_specified_timeout = user_specified_timeout
         self.save_temps = save_temps
         self.pass_statistic = pass_statistic
         self.test_cases: set[Path] = set()
@@ -459,6 +470,7 @@ def __init__(
         self.start_with_pass = start_with_pass
         self.skip_after_n_transforms = skip_after_n_transforms
         self.stopping_threshold = stopping_threshold
+        self.no_auto_adjust_timeout = no_auto_adjust_timeout
         self.exit_stack = contextlib.ExitStack()
 
         for test_case in test_cases:
@@ -471,6 +483,8 @@ def __init__(
                 raise ScriptInsideTestCaseError(test_case, self.test_script)
             self.test_cases.add(test_case)
 
+        self.transform_job_timeout_estimator = TimeoutEstimator(user_specified_timeout)
+        self.pass_global_infos: dict[str, PassGlobalInfo] = {}
         self.orig_total_file_size = self.total_file_size
         self.cache = None if self.no_cache else cache.Cache(f'{self.TEMP_PREFIX}cache-')
         self.pass_contexts: list[PassContext] = []
@@ -793,6 +807,10 @@ def handle_finished_init_job(self, job: Job) -> None:
         job.temporary_folder = None
 
         self.pass_statistic.add_initialized(job.pass_, job.start_time)
+
+        pass_name = repr(job.pass_)
+        self.pass_global_infos[pass_name].init_job_timeout_estimator.update(job.start_time)
+
         if isinstance(ctx.pass_, HintBasedPass):
             ctx.hint_bundle_paths = {} if ctx.state is None else ctx.state.hint_bundle_paths()
 
@@ -816,6 +834,7 @@ def handle_finished_transform_job(self, job: Job) -> None:
             return
         assert outcome == PassCheckingOutcome.ACCEPT
         self.pass_statistic.add_success(job.pass_)
+        self.transform_job_timeout_estimator.update(job.start_time)
         self.maybe_update_success_candidate(job.order, job.pass_, job.pass_id, env)
         if self.interleaving:
             self.folding_manager.on_transform_job_success(env.state)
@@ -953,6 +972,11 @@ def run_passes(self, passes: Sequence[AbstractPass], interleaving: bool):
             else:
                 return
 
+        for pass_ in passes:
+            pass_name = repr(pass_)
+            if pass_name not in self.pass_global_infos:
+                self.pass_global_infos[pass_name] = PassGlobalInfo.create(self.get_starting_init_job_timeout())
+
         self.order = 1
         self.last_restart_job_order = None
         self.pass_restart_queue = []
@@ -1049,6 +1073,9 @@ def run_passes(self, passes: Sequence[AbstractPass], interleaving: bool):
             self.remove_roots()
             sys.exit(1)
 
+    def get_starting_init_job_timeout(self) -> float:
+        return self.INIT_TIMEOUT_FACTOR * self.transform_job_timeout_estimator.estimate()
+
     def process_result(self) -> None:
         assert self.success_candidate
         new_test_case = self.success_candidate.test_case_path
@@ -1205,6 +1232,13 @@ def schedule_init(self, pass_id: int, ready_hint_types: set[bytes]) -> None:
         ctx = self.pass_contexts[pass_id]
         assert ctx.can_init_now(ready_hint_types)
 
+        pass_name = repr(ctx.pass_)
+        timeout = (
+            self.INIT_TIMEOUT_FACTOR * self.user_specified_timeout
+            if self.no_auto_adjust_timeout
+            else self.pass_global_infos[pass_name].init_job_timeout_estimator.estimate()
+        )
+
         dependee_types = set(ctx.pass_.input_hint_types()) if isinstance(ctx.pass_, HintBasedPass) else set()
         dependee_bundle_paths = []
         for other in self.pass_contexts:
@@ -1223,7 +1257,7 @@ def schedule_init(self, pass_id: int, ready_hint_types: set[bytes]) -> None:
                 pass_new=ctx.pass_.new,
                 test_case=self.current_test_case,
                 tmp_dir=tmp_dir,
-                job_timeout=self.timeout,
+                job_timeout=timeout,
                 pid_queue=self.process_monitor.pid_queue,
                 dependee_bundle_paths=dependee_bundle_paths,
             )
@@ -1234,14 +1268,11 @@ def schedule_init(self, pass_id: int, ready_hint_types: set[bytes]) -> None:
                 pass_previous_state=ctx.state,
                 new_tmp_dir=tmp_dir,
                 pass_succeeded_state=ctx.taken_succeeded_state,
-                job_timeout=self.timeout,
+                job_timeout=timeout,
                 pid_queue=self.process_monitor.pid_queue,
                 dependee_bundle_paths=dependee_bundle_paths,
             )
-        init_timeout = self.INIT_TIMEOUT_FACTOR * self.timeout
-        future = self.worker_pool.schedule(
-            _worker_process_job_wrapper, args=[self.order, env.run], timeout=init_timeout
-        )
+        future = self.worker_pool.schedule(_worker_process_job_wrapper, args=[self.order, env.run], timeout=timeout)
         self.jobs.append(
             Job(
                 type=JobType.INIT,
@@ -1252,7 +1283,7 @@ def schedule_init(self, pass_id: int, ready_hint_types: set[bytes]) -> None:
                 pass_user_visible_name=ctx.pass_.user_visible_name(),
                 pass_job_counter=ctx.pass_job_counter,
                 start_time=time.monotonic(),
-                timeout=init_timeout,
+                timeout=timeout,
                 temporary_folder=tmp_dir,
             )
         )
@@ -1271,6 +1302,11 @@ def schedule_transform(self, pass_id: int) -> None:
         # simply hardcode that hint-based passes are capable of this (and they actually need the original files anyway).
         should_copy_test_cases = not isinstance(ctx.pass_, HintBasedPass)
 
+        timeout = (
+            self.user_specified_timeout
+            if self.no_auto_adjust_timeout
+            else self.transform_job_timeout_estimator.estimate()
+        )
         folder = Path(tempfile.mkdtemp(prefix=self.TEMP_PREFIX, dir=ctx.temporary_root))
         env = TestEnvironment(
             ctx.state,
@@ -1283,9 +1319,7 @@ def schedule_transform(self, pass_id: int) -> None:
             ctx.pass_.transform,
             self.process_monitor.pid_queue,
         )
-        future = self.worker_pool.schedule(
-            _worker_process_job_wrapper, args=[self.order, env.run], timeout=self.timeout
-        )
+        future = self.worker_pool.schedule(_worker_process_job_wrapper, args=[self.order, env.run], timeout=timeout)
         self.jobs.append(
             Job(
                 type=JobType.TRANSFORM,
@@ -1296,7 +1330,7 @@ def schedule_transform(self, pass_id: int) -> None:
                 pass_user_visible_name=ctx.pass_.user_visible_name(),
                 pass_job_counter=ctx.pass_job_counter,
                 start_time=time.monotonic(),
-                timeout=self.timeout,
+                timeout=timeout,
                 temporary_folder=folder,
             )
         )
@@ -1311,6 +1345,11 @@ def schedule_fold(self, folding_state: FoldingStateIn) -> None:
         assert self.interleaving
 
         should_copy_test_cases = False  # the fold transform creates the files itself
+        timeout = (
+            self.user_specified_timeout
+            if self.no_auto_adjust_timeout
+            else self.transform_job_timeout_estimator.estimate()
+        )
         folder = Path(tempfile.mkdtemp(prefix=self.TEMP_PREFIX + 'folding-'))
         env = TestEnvironment(
             folding_state,
@@ -1323,9 +1362,7 @@ def schedule_fold(self, folding_state: FoldingStateIn) -> None:
             FoldingManager.transform,
             self.process_monitor.pid_queue,
         )
-        future = self.worker_pool.schedule(
-            _worker_process_job_wrapper, args=[self.order, env.run], timeout=self.timeout
-        )
+        future = self.worker_pool.schedule(_worker_process_job_wrapper, args=[self.order, env.run], timeout=timeout)
         self.jobs.append(
             Job(
                 type=JobType.TRANSFORM,
@@ -1336,7 +1373,7 @@ def schedule_fold(self, folding_state: FoldingStateIn) -> None:
                 pass_user_visible_name='Folding',
                 pass_job_counter=None,
                 start_time=time.monotonic(),
-                timeout=self.timeout,
+                timeout=timeout,
                 temporary_folder=folder,
             )
         )