diff --git a/.gitignore b/.gitignore index 0ee97f0..ce799a1 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ __pycache__/ *.log* *out +*.swp +*.swo diff --git a/ats/atsMachines/fluxScheduled.py b/ats/atsMachines/fluxScheduled.py index 48c4cab..4542a78 100755 --- a/ats/atsMachines/fluxScheduled.py +++ b/ats/atsMachines/fluxScheduled.py @@ -175,6 +175,15 @@ def get_physical_node(self, rel_index): raise IndexError(f"Relative index {rel_index} out of range (0-{len(nodes)-1})") return nodes[rel_index] + def getNumberOfProcessors(self): + """Return total schedulable processor slots in the Flux allocation. + + Returns: + int: Total processor capacity ATS may schedule inside the current + Flux allocation. + """ + return self.maxCores + def kill(self, test): """ diff --git a/ats/completion_detector.py b/ats/completion_detector.py new file mode 100644 index 0000000..280cac9 --- /dev/null +++ b/ats/completion_detector.py @@ -0,0 +1,143 @@ +"""Completion-detection strategy base class and factory helpers.""" + +from abc import ABC, abstractmethod + +from ats.atsut import AtsError + + +def normalize_completion_detection_mode(mode): + """Normalize a completion-detection mode string. + + Args: + mode (str|None): Requested mode name. + + Returns: + str: Normalized lowercase mode name, defaulting to + ``"per_test_watcher"``. + """ + return str(mode or "per_test_watcher").strip().lower() or "per_test_watcher" + + +def _validate_completion_detection_mode_for_machine(machine, normalized_mode): + """Reject detector modes unsupported by one machine implementation. + + Args: + machine: Machine instance that would own the detector. + normalized_mode (str): Normalized detector mode name. + + Returns: + None: Validation succeeds without modifying ``machine``. + + Raises: + AtsError: If ``normalized_mode`` is unsupported for ``machine``. + """ + machine_class = machine.__class__.__name__ + machine_module = machine.__class__.__module__ + if ( + normalized_mode in ("waitpid_reaper", "per_test_watcher") + and machine_class == "FluxDirect" + and machine_module.endswith("flux_direct") + ): + raise AtsError( + "threaded completion detection is unsupported for FluxDirect. " + "Use 'poll' for this experimental machine." + ) + + +class CompletionDetector(ABC): + """Abstract policy object with shared ATS completion-detection helpers.""" + + mode_name = "" + + def __init__(self, machine): + """Bind one completion detector to one ATS machine. + + Args: + machine: Machine instance that owns completion helpers and running + test state. + + Returns: + None: The detector stores a reference to ``machine``. + """ + self.machine = machine + + def register_launched_test(self, test): + """Prepare one launched test for detector-specific completion work. + + Args: + test: ATS test object whose child process has just been launched. + + Returns: + None: The default detector implementation needs no launch-time + setup. + """ + + def unregister_finished_test(self, test): + """Release detector-owned state associated with one finished test. + + Args: + test: ATS test object that may own detector-specific wait state. + + Returns: + None: The default detector implementation needs no cleanup. + """ + + def owns_child_reaping(self): + """Return whether this detector is responsible for child reaping. + + Returns: + bool: ``False`` for the default detector behavior. + """ + return False + + @abstractmethod + def logCompletionWarnings(self, logger): + """Print any detector-specific end-of-run completion warnings. + + Args: + logger (callable): Logging function compatible with ``ats.log.log``. + + Returns: + None: Implementations may emit zero or more warnings. + """ + + @abstractmethod + def check_running(self): + """Update machine running state according to one detector strategy. + + Returns: + None: Implementations may finish tests and update ``machine.running``. + """ + + +def create_completion_detector(machine, mode): + """Create one completion detector instance for the requested mode. + + Args: + machine: Machine instance that will own the detector. + mode (str|None): Requested detector mode. + + Returns: + CompletionDetector: Strategy instance for the requested mode. + + Raises: + AtsError: If ``mode`` is not one of the supported detector modes. + """ + normalized_mode = normalize_completion_detection_mode(mode) + _validate_completion_detection_mode_for_machine(machine, normalized_mode) + if normalized_mode == "waitpid_reaper": + from ats.completion_waitpid_reaper import WaitpidReaperCompletionDetector + + return WaitpidReaperCompletionDetector(machine) + if normalized_mode == "per_test_watcher": + from ats.completion_per_test_watcher import PerTestWatcherCompletionDetector + + return PerTestWatcherCompletionDetector(machine) + if normalized_mode == "poll": + from ats.completion_poll import PollingCompletionDetector + + return PollingCompletionDetector(machine) + raise AtsError( + "Unknown completion detection mode %r. Expected one of: " + "'waitpid_reaper', 'per_test_watcher', 'poll'." % mode + ) diff --git a/ats/completion_per_test_watcher.py b/ats/completion_per_test_watcher.py new file mode 100644 index 0000000..f692fb9 --- /dev/null +++ b/ats/completion_per_test_watcher.py @@ -0,0 +1,57 @@ +"""Per-test watcher completion detector for ATS machines.""" + +import threading + +from ats.completion_waitpid_reaper import WaitpidReaperCompletionDetector + + +class PerTestWatcherCompletionDetector(WaitpidReaperCompletionDetector): + """Use one ``Popen.wait`` watcher thread per launched test. + + Each watcher + waits on the exact child process owned by its test, then enqueues that + test for the shared fast-path drain. + This detector is thread safe, but may not scale beyond a few nodes of + concurrently running tests. + """ + + mode_name = "per_test_watcher" + + def owns_child_reaping(self): + """Simple queue mode leaves child reaping to ``subprocess.Popen``.""" + return False + + def register_launched_test(self, test): + """Start one watcher thread that waits for this child to exit.""" + child = getattr(test, "child", None) + if child is None: + return + watcher = getattr(test, "_completionWatcher", None) + if watcher is not None: + self.machine._incrementCompletionStat("per_test_watcher_already_registered") + return + + def watch_for_completion(): + """Wait for one child to exit and then enqueue its completion.""" + try: + child.wait() + except Exception: + self.machine._incrementCompletionStat("per_test_watcher_wait_error") + return + self.machine._incrementCompletionStat("per_test_watcher_wait_completed") + self.record_completion_signal(test) + + watcher = threading.Thread( + target=watch_for_completion, + name=f"ats-completion-{getattr(child, 'pid', 'unknown')}", + daemon=True, + ) + test._completionWatcher = watcher + watcher.start() + self.machine._incrementCompletionStat("per_test_watcher_registered") + + def unregister_finished_test(self, test): + """Clear watcher bookkeeping for one finished test.""" + if hasattr(test, "_completionWatcher"): + test._completionWatcher = None + self.machine._incrementCompletionStat("per_test_watcher_cleared") diff --git a/ats/completion_poll.py b/ats/completion_poll.py new file mode 100644 index 0000000..9468727 --- /dev/null +++ b/ats/completion_poll.py @@ -0,0 +1,126 @@ +"""Polling completion detector for ATS machines.""" + +import time + +from ats.completion_detector import CompletionDetector +from ats.atsut import AtsError, PASSED + + +class PollingCompletionDetector(CompletionDetector): + """Sleep-then-poll completion behavior.""" + + mode_name = "poll" + + def preserve_new_running_tests(self, remaining, seen_ids): + """Keep tests appended to ``machine.running`` during completion callbacks. + + Args: + remaining (list): Running tests that should remain after the current + polling pass. + seen_ids (set): Object ids already considered in the polling pass. + + Returns: + None: ``remaining`` is updated in place. + """ + remaining_ids = {id(test) for test in remaining} + for test in self.machine.running: + test_id = id(test) + if test_id in seen_ids or test_id in remaining_ids: + continue + remaining.append(test) + remaining_ids.add(test_id) + + def poll_running_tests( + self, + allow_running_checks, + completion_limit=None, + ): + """Poll running tests in scheduler order. + + Args: + allow_running_checks (bool): When ``False``, skip timeout and + runtime error checks for children that have not yet exited. + completion_limit (int|None): Maximum number of completions to + process before returning control to the scheduler. + + Returns: + int: Number of completed tests processed in this polling pass. + """ + from ats import configuration + + machine = self.machine + start_us = time.time_ns() // 1000 + machine._incrementCompletionStat("_pollRunningTests_called") + if allow_running_checks: + machine._incrementCompletionStat("_pollRunningTests_allow_running_checks_true") + else: + machine._incrementCompletionStat("_pollRunningTests_allow_running_checks_false") + + ordered_count = 0 + completed = 0 + result_kind = "completed_none" + try: + ordered = list(machine.running) + seen_ids = {id(test) for test in ordered} + ordered_count = len(ordered) + machine._incrementCompletionStat("_pollRunningTests_total_ordered", ordered_count) + + remaining = [] + for index, test in enumerate(ordered): + done = machine.getStatus(test, allow_running_checks=allow_running_checks) + if not done: + remaining.append(test) + continue + completed += 1 + if test.status is not PASSED and configuration.options.oneFailure: + raise AtsError("Test failed in oneFailure mode.") + if completion_limit is not None and completed >= completion_limit: + remaining.extend(ordered[index + 1:]) + self.preserve_new_running_tests(remaining, seen_ids) + machine.running = remaining + result_kind = "stopped_after_completion_limit" + machine._incrementCompletionStat("_pollRunningTests_stopped_after_completion") + machine._incrementCompletionStat("_pollRunningTests_total_completed", completed) + return completed + + self.preserve_new_running_tests(remaining, seen_ids) + machine.running = remaining + machine._incrementCompletionStat("_pollRunningTests_total_completed", completed) + if completed: + result_kind = "completed" + machine._incrementCompletionStat("_pollRunningTests_completed") + else: + machine._incrementCompletionStat("_pollRunningTests_completed_none") + return completed + finally: + machine._recordCompletionInternalSpan( + "_pollRunningTests", + start_us, + time.time_ns() // 1000, + metadata={ + "mode": getattr(machine, "completion_detection_mode", ""), + "allow_running_checks": bool(allow_running_checks), + "ordered_count": ordered_count, + "completion_limit": completion_limit, + "completed_count": completed, + "result": result_kind, + }, + ) + + + def check_running(self): + """Advance machine state using plain sleep-then-poll behavior. + + Returns: + None: Completed tests may be finalized and removed from + ``machine.running``. + """ + machine = self.machine + time.sleep(machine.naptime) + self.poll_running_tests( + allow_running_checks=True, + ) + + def logCompletionWarnings(self, logger): + """Polling mode has no detector-specific completion warnings.""" + return None diff --git a/ats/completion_waitpid_reaper.py b/ats/completion_waitpid_reaper.py new file mode 100644 index 0000000..cfdef9f --- /dev/null +++ b/ats/completion_waitpid_reaper.py @@ -0,0 +1,417 @@ +"""Waitpid-reaper completion detector for ATS machines.""" + +import os +import threading +import time + +from ats.atsut import AtsError, PASSED +from ats.completion_detector import CompletionDetector + +# TODO: When we move to RHEL 5 or RHEL 6, migrate this to use pidfd instead of a reaper thread strategy. +class WaitpidReaperCompletionDetector(CompletionDetector): + """Drain explicitly signaled completions from a machine-owned queue.""" + + mode_name = "waitpid_reaper" + + def __init__(self, machine): + """Initialize queue-mode reaper state for one ATS machine. + + Args: + machine: Machine instance that owns queue-mode running tests. + """ + super().__init__(machine) + self._reaper_thread = None + self._reaper_stop = False + self._reaper_lock = threading.Lock() + self._reaper_condition = threading.Condition(self._reaper_lock) + self._registered_tests_by_pid = {} + self._unexpected_reaps = [] + self._unexpected_reaps_count = 0 + self._unexpected_reaps_dropped = 0 + self._unexpected_reaps_lock = threading.Lock() + + def owns_child_reaping(self): + """Queue mode owns child reaping through the detector reaper.""" + return True + + def _ensure_reaper_started(self): + """Start the queue-mode reaper thread on first child registration.""" + if self._reaper_thread is not None: + return + self._reaper_thread = threading.Thread( + target=self._reaper_loop, + name="ats-completion-reaper", + daemon=True, + ) + self._reaper_thread.start() + + def _wait_status_to_returncode(self, wait_status): + """Convert one raw wait status to subprocess-style return codes.""" + if os.WIFEXITED(wait_status): + return os.WEXITSTATUS(wait_status) + if os.WIFSIGNALED(wait_status): + return -os.WTERMSIG(wait_status) + return wait_status + + def _handle_reaped_pid(self, pid, wait_status): + """Route one reaped child to the matching registered test if present.""" + machine = self.machine + with self._reaper_condition: + test = self._registered_tests_by_pid.pop(pid, None) + + if test is None: + machine._incrementCompletionStat("completion_queue_reaper_unknown_pid") + self._recordUnexpectedCompletionReap(pid, wait_status) + return + + child = getattr(test, "child", None) + if child is None: + machine._incrementCompletionStat("completion_queue_reaper_missing_child") + return + + child.returncode = self._wait_status_to_returncode(wait_status) + machine._incrementCompletionStat("completion_queue_reaper_reaped") + self.record_completion_signal(test) + + def _recordUnexpectedCompletionReap(self, pid, wait_status): + """Remember one unexpectedly reaped child for end-of-run warnings.""" + observed_us = time.time_ns() // 1000 + if os.WIFEXITED(wait_status): + outcome = "exit %d" % os.WEXITSTATUS(wait_status) + elif os.WIFSIGNALED(wait_status): + outcome = "signal %d" % os.WTERMSIG(wait_status) + else: + outcome = "wait_status %d" % wait_status + sample = { + "pid": pid, + "wait_status": wait_status, + "outcome": outcome, + "observed_us": observed_us, + } + with self._unexpected_reaps_lock: + self._unexpected_reaps_count += 1 + if len(self._unexpected_reaps) < 8: + self._unexpected_reaps.append(sample) + else: + self._unexpected_reaps_dropped += 1 + + def _unexpectedReapsSnapshot(self): + """Return a snapshot of unexpected completion-reaper activity.""" + with self._unexpected_reaps_lock: + return { + "count": self._unexpected_reaps_count, + "samples": list(self._unexpected_reaps), + "dropped": self._unexpected_reaps_dropped, + } + + def logCompletionWarnings(self, logger): + """Print end-of-run warnings for suspicious queue-reaper events.""" + snapshot = self._unexpectedReapsSnapshot() + if snapshot["count"] <= 0: + logger( + "WARNING: waitpid_reaper did not hit its known waitpid(-1) race in this run, " + "but the risk remains. Use per_test_watcher if you need the safer path." + ) + return + logger( + "WARNING: waitpid_reaper reaped %d child process(es) that were not " + "registered ATS tests. This indicates the queue-mode reaper hit the " + "known waitpid(-1) race and may have consumed another ATS subprocess " + "exit status. This is especially an issue if wait_status!=0 for the pid." % snapshot["count"] + ) + for sample in snapshot["samples"]: + logger( + "WARNING: unexpected reaped pid=%d outcome=%s wait_status=%d observed_us=%d" + % ( + sample["pid"], + sample["outcome"], + sample["wait_status"], + sample["observed_us"], + ) + ) + if snapshot["dropped"]: + logger( + "WARNING: %d additional unexpected reaped child event(s) were not " + "listed individually." % snapshot["dropped"] + ) + + def _reaper_loop(self): + """Reap registered queue-mode children and enqueue their completions.""" + machine = self.machine + while True: + with self._reaper_condition: + while not self._reaper_stop and not self._registered_tests_by_pid: + self._reaper_condition.wait() + if self._reaper_stop and not self._registered_tests_by_pid: + return + + try: + pid, wait_status = os.waitpid(-1, 0) + except ChildProcessError: + machine._incrementCompletionStat("completion_queue_reaper_child_process_error") + with self._reaper_condition: + if not self._registered_tests_by_pid: + continue + time.sleep(0.01) + continue + except OSError: + machine._incrementCompletionStat("completion_queue_reaper_waitpid_error") + time.sleep(0.01) + continue + + self._handle_reaped_pid(pid, wait_status) + + def completion_drain_limit(self): + """Return the configured maximum completions drained per wakeup. + + Returns: + int: Positive completion drain limit. + """ + limit = getattr(self.machine, "completion_fast_path_drain_limit", 128) + try: + limit = int(limit) + except (TypeError, ValueError): + limit = 128 + return max(1, limit) + + def wait_for_completion_signal(self): + """Wait one polling interval for queued completion signals. + + Returns: + None: Will have waited for {machine.naptime} seconds, and updated appropriate + statistics and timing data. + """ + start_us = time.time_ns() // 1000 + machine = self.machine + machine._incrementCompletionStat("_waitForCompletionSignal_called") + result_kind = "queue_event_wait" + try: + machine._incrementCompletionStat("_waitForCompletionSignal_queue_event_wait") + machine._completionEvent.wait(machine.naptime) + finally: + machine._recordCompletionInternalSpan( + "_waitForCompletionSignal", + start_us, + time.time_ns() // 1000, + metadata={ + "mode": getattr(machine, "completion_detection_mode", ""), + "running_count": len(machine.running), + "registered": False, + "registered_count": 0, + "ready_count": 0, + "used_queue_event_wait": True, + "result": result_kind, + }, + ) + + def register_launched_test(self, test): + """Register one launched child with the queued completion reaper. + + Args: + test: ATS test object whose child process has just been launched. + + Returns: + None: The queue-mode reaper is started lazily and the child pid is + registered for reaping. + """ + child = getattr(test, "child", None) + pid = getattr(child, "pid", None) + if child is None or pid is None: + return + with self._reaper_condition: + self._ensure_reaper_started() + self._registered_tests_by_pid[pid] = test + self._reaper_condition.notify() + self.machine._incrementCompletionStat("completion_queue_reaper_registered") + + def unregister_finished_test(self, test): + """Clear reaper bookkeeping for one finished test. + + Args: + test: ATS test object that may still be registered with the queue + reaper. + + Returns: + None: Any stale pid registration is removed when present. + """ + child = getattr(test, "child", None) + pid = getattr(child, "pid", None) + if pid is None: + return + with self._reaper_condition: + registered = self._registered_tests_by_pid.get(pid) + if registered is test: + self._registered_tests_by_pid.pop(pid, None) + + def check_running(self): + """Advance machine state by draining the queued completion set first. + + Returns: + None: Completed tests may be finalized and removed from + ``machine.running``. + """ + machine = self.machine + completion_limit = self.completion_drain_limit() + machine._incrementCompletionStat("check_running_completion_queue_mode") + if self.poll_queued_completion_tests(completion_limit=completion_limit): + machine._incrementCompletionStat("check_running_queue_pre_drain_completed") + else: + machine._incrementCompletionStat("check_running_queue_pre_drain_empty") + machine._incrementCompletionStat("check_running_wait_for_completion_signal") + self.wait_for_completion_signal() + if self.poll_queued_completion_tests(completion_limit=completion_limit): + machine._incrementCompletionStat("check_running_queue_post_wait_completed") + else: + machine._incrementCompletionStat("check_running_queue_post_wait_empty") + machine.scan_running_tests_for_health() + + def record_completion_signal(self, test): + """Record a likely completion signal and enqueue it for later draining. + + Args: + test: ATS test object associated with the completion signal. + + Returns: + None: Internal timestamps, queue state, and statistics are updated. + """ + machine = self.machine + observed_us = time.time_ns() // 1000 + if getattr(test, "ats_completion_signal_us", None) is None: + test.ats_completion_signal_us = observed_us + machine._incrementCompletionStat("completion_signal_recorded") + with machine._completionQueueLock: + test_id = id(test) + if test_id in machine._completionQueueIds: + machine._incrementCompletionStat("completion_queue_duplicate_signal") + return + machine._completionQueue.append(test) + machine._completionQueueIds.add(test_id) + machine._completionEvent.set() + machine._incrementCompletionStat("completion_queue_enqueued") + depth = len(machine._completionQueue) + machine._recordCompletionQueueSnapshot( + depth, + "completion_queue_enqueue", + timestamp_us=observed_us, + ) + + def drain_completion_queue(self, completion_limit=None): + """Remove queued completion candidates up to the configured limit. + + Args: + completion_limit (int|None): Maximum number of queued tests to + return. ``None`` drains the entire queue. + + Returns: + list: Queued tests selected for completion re-checking. + """ + machine = self.machine + queued = [] + with machine._completionQueueLock: + while machine._completionQueue: + if completion_limit is not None and len(queued) >= completion_limit: + break + test = machine._completionQueue.popleft() + machine._completionQueueIds.discard(id(test)) + queued.append(test) + remaining_depth = len(machine._completionQueue) + if not machine._completionQueue: + machine._completionEvent.clear() + if queued: + machine._recordCompletionQueueSnapshot( + remaining_depth, + "completion_queue_drain", + metadata={ + "drained_count": len(queued), + "completion_limit": completion_limit, + }, + ) + return queued + + def poll_queued_completion_tests(self, completion_limit=None): + """Handle completion candidates from the queued completion path. + + Args: + completion_limit (int|None): Maximum number of queued candidates to + process in this pass. + + Returns: + int: Number of running tests confirmed completed in this pass. + """ + from ats import configuration + + machine = self.machine + start_us = time.time_ns() // 1000 + machine._incrementCompletionStat("_pollQueuedCompletionTests_called") + queued_count = 0 + selected_count = 0 + stale_count = 0 + completed = 0 + result_kind = "empty" + try: + queued = self.drain_completion_queue(completion_limit=completion_limit) + queued_count = len(queued) + machine._incrementCompletionStat("_pollQueuedCompletionTests_total_queued", queued_count) + if not queued: + machine._incrementCompletionStat("_pollQueuedCompletionTests_empty") + return 0 + + selected = [] + selected_ids = set() + running_ids = {id(test) for test in machine.running} + for test in queued: + test_id = id(test) + if test_id in selected_ids: + continue + if test_id not in running_ids: + stale_count += 1 + continue + selected.append(test) + selected_ids.add(test_id) + + selected_count = len(selected) + machine._incrementCompletionStat("_pollQueuedCompletionTests_total_selected", selected_count) + machine._incrementCompletionStat("_pollQueuedCompletionTests_total_stale", stale_count) + if stale_count: + machine._incrementCompletionStat("_pollQueuedCompletionTests_saw_stale_entries") + if not selected: + result_kind = "stale_only" + machine._incrementCompletionStat("_pollQueuedCompletionTests_selected_none") + return 0 + + completed_ids = set() + for test in selected: + done = machine.getStatus(test, allow_running_checks=False) + if not done: + continue + completed_ids.add(id(test)) + completed += 1 + if test.status is not PASSED and configuration.options.oneFailure: + raise AtsError("Test failed in oneFailure mode.") + + machine._incrementCompletionStat("_pollQueuedCompletionTests_total_completed", completed) + if completed_ids: + machine.running = [ + test for test in machine.running if id(test) not in completed_ids + ] + result_kind = "completed" + machine._incrementCompletionStat("_pollQueuedCompletionTests_completed") + else: + result_kind = "selected_none_completed" + machine._incrementCompletionStat("_pollQueuedCompletionTests_selected_none_completed") + return completed + finally: + machine._recordCompletionInternalSpan( + "_pollQueuedCompletionTests", + start_us, + time.time_ns() // 1000, + metadata={ + "mode": getattr(machine, "completion_detection_mode", ""), + "completion_limit": completion_limit, + "queued_count": queued_count, + "selected_count": selected_count, + "stale_count": stale_count, + "completed_count": completed, + "result": result_kind, + }, + ) diff --git a/ats/configuration.py b/ats/configuration.py index 25a2d4f..f5ca89e 100644 --- a/ats/configuration.py +++ b/ats/configuration.py @@ -11,6 +11,7 @@ from argparse import ArgumentParser from glob import glob import importlib +import inspect import os import re import sys @@ -474,7 +475,69 @@ def get_machine_factory(module_name, machine_class, log(f"Importing {module_name} from {machine_package} caused the following error:\n{e}", echo=True) return None -def get_machine(file_text, file_name, is_batch=False): +def _instantiate_machine(machine_factory, + machine_name, + npMaxH, + completion_detection_mode=None): + """Instantiate a machine factory with optional detector selection. + + Args: + machine_factory (callable): Factory or class used to build the machine. + machine_name (str): Machine name passed to the constructor. + npMaxH (int|str): Hardware processor-slot limit for the machine. + completion_detection_mode (str|None): Requested completion detector + mode. When the factory supports it, the mode is passed as a keyword + argument. Older two-argument constructors remain supported. + + Returns: + object: Instantiated machine object. + """ + npMaxH = int(npMaxH) + if completion_detection_mode is None: + return machine_factory(machine_name, npMaxH) + + try: + signature = inspect.signature(machine_factory) + except (TypeError, ValueError): + signature = None + + if signature is not None: + parameters = signature.parameters.values() + supports_mode_kwarg = any( + parameter.kind == inspect.Parameter.VAR_KEYWORD + or parameter.name == "completion_detection_mode" + for parameter in parameters + ) + if not supports_mode_kwarg: + return machine_factory(machine_name, npMaxH) + + try: + return machine_factory( + machine_name, + npMaxH, + completion_detection_mode=completion_detection_mode, + ) + except TypeError: + return machine_factory(machine_name, npMaxH) + +def get_machine(file_text, + file_name, + is_batch=False, + completion_detection_mode=None): + """Create the machine declared in one machine-specification file. + + Args: + file_text (str): Full text of the candidate machine module. + file_name (str): Basename of the candidate machine module. + is_batch (bool): When ``True``, search ``#BATS:`` declarations instead + of ``#ATS:`` declarations. + completion_detection_mode (str|None): Requested completion detector + mode to pass through machine construction when supported. + + Returns: + object|None: Matching machine instance, or ``None`` when the file does + not define the active machine type. + """ header = '#BATS:' if is_batch else '#ATS:' machine_type = BATCH_TYPE if is_batch else MACHINE_TYPE ats_lines = (ats_line for ats_line in file_text.splitlines() @@ -503,7 +566,12 @@ def get_machine(file_text, file_name, is_batch=False): f"import {machine_class} as Machine") if machine_factory: - machine = machine_factory(machine_name, int(npMaxH)) + machine = _instantiate_machine( + machine_factory, + machine_name, + npMaxH, + completion_detection_mode=completion_detection_mode, + ) break else: @@ -511,7 +579,7 @@ def get_machine(file_text, file_name, is_batch=False): return machine -def get_machine_entry_points(machine_class): +def get_machine_entry_points(machine_class, completion_detection_mode=None): """ Looks for custom machine type via entry_points plugins installed by ats wrappers. @@ -521,6 +589,16 @@ def get_machine_entry_points(machine_class): Batch mode not really supported this way? -> would want to tag batch vs ats headers as an instance/class variable instead in this mode rather than rely on the header comments + + Args: + machine_class (str): Machine type to resolve through installed entry + points. + completion_detection_mode (str|None): Requested completion detector + mode to pass through machine construction when supported. + + Returns: + object|None: Machine instance loaded from an entry-point plugin, or + ``None`` when no plugin matches. """ log("Machine Factory: looping over available machine plugins:", echo=False) @@ -531,7 +609,12 @@ def get_machine_entry_points(machine_class): for machine_factory in ats_machines: if machine_class in machine_factory.value: log(f"Machine Factory: Found machine {machine_factory.name} of class {machine_factory.value}: {machine_factory}") - return machine_factory.load()(machine_class, -1) + return _instantiate_machine( + machine_factory.load(), + machine_class, + -1, + completion_detection_mode=completion_detection_mode, + ) else: ats_machines = {machine.name: machine for group, machines in entry_points().items() @@ -541,17 +624,33 @@ def get_machine_entry_points(machine_class): for name, machine_factory in ats_machines.items(): if machine_class in machine_factory.value: log(f"Machine Factory: Found machine {name} of class {machine_class}: {machine_factory}") - return machine_factory.load()(machine_class, -1) + return _instantiate_machine( + machine_factory.load(), + machine_class, + -1, + completion_detection_mode=completion_detection_mode, + ) # Downstream needs to be able to detect if machine isn't found return None -def init(clas = '', adder = None, examiner=None): - """Called by manager.init(class, adder, examiner) - Initialize configuration and process command-line options; create log, - options, inputFiles, timelimit, machine, and batchmatchine. - Call backs to machine and to adder/examiner for options. +def init(clas = '', adder = None, examiner=None, + completion_detection_mode=None): + """Initialize ATS configuration, options, and machine instances. + + Args: + clas (str): ATS command-line string to parse instead of + ``sys.argv[1:]``. + adder (callable|None): Optional callback that adds parser options + before ATS parses the command line. + examiner (callable|None): Optional callback that inspects parsed + options after initialization. + completion_detection_mode (str|None): Requested completion detector + mode to pass through machine construction when supported. + + Returns: + None: Module-level ATS configuration state is updated in place. """ global log, options, inputFiles, timelimit, machine, batchmachine,\ defaultExecutable, ATSROOT, cuttime @@ -599,25 +698,41 @@ def init(clas = '', adder = None, examiner=None): file_name = os.path.basename(full_path) if not machine and re.search(ATS_PATTERN, file_text): - machine = get_machine(file_text, file_name) + machine = get_machine( + file_text, + file_name, + completion_detection_mode=completion_detection_mode, + ) specFoundIn = full_path if not batchmachine and re.search(BATS_PATTERN, file_text): - batchmachine = get_machine(file_text, file_name, is_batch=True) + batchmachine = get_machine( + file_text, + file_name, + is_batch=True, + completion_detection_mode=completion_detection_mode, + ) bspecFoundIn = full_path if machine and batchmachine: break # Check entry_points plugins to override built-in machines - machine_plugin = get_machine_entry_points(MACHINE_TYPE) + machine_plugin = get_machine_entry_points( + MACHINE_TYPE, + completion_detection_mode=completion_detection_mode, + ) if machine_plugin: machine = machine_plugin if machine is None: terminal("No machine specifications for", SYS_TYPE, "found, using generic.") - machine = machines.Machine('generic', -1) + machine = machines.Machine( + 'generic', + -1, + completion_detection_mode=completion_detection_mode, + ) # create the option set usage = "usage: %(prog)s [options] [input files]" diff --git a/ats/machines.py b/ats/machines.py index 2915f49..ce12218 100644 --- a/ats/machines.py +++ b/ats/machines.py @@ -1,6 +1,8 @@ """Definition of class Machine for overriding. """ -import subprocess, sys, os, time, shlex +from collections import deque +import subprocess, sys, os, threading, time, shlex +from ats.completion_detector import create_completion_detector from ats.atsut import RUNNING, TIMEDOUT, PASSED, FAILED, LSFERROR, \ SKIPPED, HALTED, AtsError from ats.log import log, terminal @@ -21,6 +23,27 @@ class MachineCore(object): printExperimentalNotice = False printSleepBeforeSrunNotice = True + def __init__(self, completion_detection_mode=None): + """Initialize machine-owned completion-detection state. + + Args: + completion_detection_mode (str|None): Requested completion-detector + mode. When omitted, ATS falls back to ``"per_test_watcher"``. + + Returns: + None: Completion detector state and hooks are initialized. + """ + self._completionEvent = threading.Event() + self._completionQueue = deque() + self._completionQueueIds = set() + self._completionQueueLock = threading.Lock() + self._completionStats = {} + self._completionStatsLock = threading.Lock() + self._completion_span_hooks = [] + self._completion_queue_snapshot_hooks = [] + self._completionDetector = None + self.configureCompletionDetector(completion_detection_mode) + # self.numberTestsRunningMax is not really the max number of tests running # but is rather the max number of processors which can run tests. def label(self): @@ -82,195 +105,480 @@ def checkForTimeOut(self, test): # return -1, fraction return 0, fraction + def configureCompletionDetector(self, completion_detection_mode=None): + """Instantiate and install the requested completion detector. + + Args: + completion_detection_mode (str|None): Requested detector mode. When + omitted, ATS falls back to ``"per_test_watcher"``. + + Returns: + object: Newly created completion detector strategy instance. + """ + self._completionDetector = create_completion_detector( + self, + completion_detection_mode, + ) + self.completion_detection_mode = self._completionDetector.mode_name + return self._completionDetector + def checkRunning(self): - """Find those tests still running. getStatus checks for timeout. + """Update ``self.running`` after checking for finished child processes. + + Returns: + None: ``self.running`` is rewritten in place and completion + callbacks may run for newly finished tests. """ - # print("DEBUG checkRunning 100\n") - from ats import configuration - time.sleep(self.naptime) - stillRunning = [] - for test in self.running: - done = self.getStatus(test) - if not done: - stillRunning.append(test) - else: # test has finished - if test.status is not PASSED: - if configuration.options.oneFailure: - raise AtsError("Test failed in oneFailure mode.") - self.running = stillRunning + self._completionDetector.check_running() def remainingCapacity(self): """How many processors are free? Could be overriden to answer the real question, what is the largest job you could start at this time?""" return self.numberTestsRunningMax - self.numberTestsRunning - def getStatus(self, test): + def getStatus(self, test, allow_running_checks=True): """ Override this only if not using subprocess (unusual). Obtains the exit code of the test object process and then sets the status of the test object accordingly. Returns True if test done. + Args: + test: ATS test object whose child status should be checked. + allow_running_checks (bool): When ``False``, skip timeout and + running-error detection for children that have not yet exited. + When a test has completed you must set test.statusCode and call self.testEnded(test, status). You may add a message as a third arg, which will be shown in the test's final report. testEnded will call your bookkeeping method noteEnd. + + Returns: + bool: ``True`` when completion handling ran for ``test``, else + ``False`` while the child remains running. """ from ats import configuration - test.child.poll() - # print(f"This is the return code for the test:{test.child.returncode}") + if self._observeCompletedChild(test): + return True + + if not allow_running_checks: + return False + + return self._checkRunningHealth(test) + + def _observeCompletedChild(self, test): + """Finalize a child that has already exited if its return code is known. + + Args: + test: ATS test object whose child may already be complete. + + Returns: + bool: ``True`` when completion finalization ran. + """ + self._pollChild(test) if test.child.returncode is None: - overtime, fraction = self.checkForTimeOut(test) - #print "DEBUG getStatus 100" - #print overtime - #print fraction - #print "DEBUG getStatus 200" - if fraction > .9 or overtime != 0: - # If a process produces a lot of output, it may fill its output - # buffer and then block until something is read from it. - - # How should testStdout handle this? ??? - # - # 2017-08-15 SAD putting back in poll, to see if it fixes hang. - if configuration.SYS_TYPE.startswith('somesystemxxx'): - stdoutdata, stderrdata = test.child.communicate() - - # Now, poll it again. - test.child.poll() - - - if test.child.returncode is None: #still running, but too long? - overtime, fraction = self.checkForTimeOut(test) - #print "DEBUG getStatus 300" - #print overtime - #print fraction - #print "DEBUG getStatus 400" - if overtime != 0: - self.kill(test) - test.statusCode = 2 - test.setEndDateTime() - if overtime > 0: - status = TIMEDOUT - else: - status = HALTED #one minute mode - else: - #print "DEBUG getStatus 320" - # SAD - # Coding to detect SLURM deficiencies, and abort job. - # Implemented 2016-Aug-30 - slurm_error = False - f = open(test.errname, 'r', errors='replace') - lines = f.readlines() - f.close - for line in lines: - if slurm_error == False: - if "Slurmd could not set up environment for batch job" in line: - print("ATS Halting test %s. Detected slurm launch failure : %s " % (test.name, line)) - slurm_error = True - elif "srun: error: Unable to create job step" in line: - print("ATS Halting test %s. Detected slurm error : %s " % (test.name, line)) - slurm_error = True - elif "Error opening remote shared memory object in shm_open" in line: - print("ATS Halting test %s. Detected MPI shared memory failure : %s " % (test.name, line)) - slurm_error = True - elif "PSM could not set up shared memory segment" in line: - print("ATS Halting test %s. Detected MPI shared memory failure : %s " % (test.name, line)) - slurm_error = True - elif "Attempting to use an MPI routine before initializing MPICH" in line: - print("ATS Halting test %s. Detected MPI Error : %s " % (test.name, line)) - slurm_error = True - elif "Bus error)" in line: - print("ATS Halting test %s. Detected Bus Error (perhaps MPI related) : %s " % (test.name, line)) - slurm_error = True - - if slurm_error: - self.kill(test) - test.statusCode = 2 - test.setEndDateTime() - status = HALTED + return False + return self._finishCompletedTest(test) - else: - return False - else: - # print "DEBUG getStatus 400" + def _checkRunningHealth(self, test): + """Apply timeout and runtime-error checks to one still-running test. + + Args: + test: ATS test object whose child is still expected to be running. + + Returns: + bool: ``True`` when the health check finishes the test. + """ + from ats import configuration + + overtime, fraction = self.checkForTimeOut(test) + if fraction > .9 or overtime != 0: + # If a process produces a lot of output, it may fill its output + # buffer and then block until something is read from it. + if configuration.SYS_TYPE.startswith('somesystemxxx'): + stdoutdata, stderrdata = test.child.communicate() + + if self._observeCompletedChild(test): + return True + + overtime, fraction = self.checkForTimeOut(test) + if overtime != 0: + self.kill(test) + test.statusCode = 2 test.setEndDateTime() - test.statusCode = test.child.returncode - # If the user set ignoreReturnCode to True then set statusCode to 0. - ignoreReturnCode = test.options.get('ignoreReturnCode', False) - if ignoreReturnCode: - test.statusCode = 0 - if test.statusCode == 0: # process is done - status = PASSED - # This checks for flux timeouts since ATS' method for determining timeouts doesnt work with flux - elif "flux" in configuration.MACHINE_TYPE and test.statusCode == 142: # 142 is the return code for a timeout from flux + if overtime > 0: status = TIMEDOUT else: - # Coding to detect LSF deficiencies - # Implemented 2018-12-12 - lsf_error = False - f = open(test.errname, 'r', errors='replace') + status = HALTED + return self._completeTest(test, status) + + if self._detectRunningSlurmError(test): + self.kill(test) + test.statusCode = 2 + test.setEndDateTime() + return self._completeTest(test, HALTED) + + return False + + def _completionStatsEnabled(self): + """Return whether aggregated completion counters should be tracked. + + Returns: + bool: ``True`` when completion statistics are enabled. + """ + return bool(getattr(self, "completion_detection_stats", False)) + + def _completionSpansEnabled(self): + """Return whether internal completion spans should be emitted. + + Returns: + bool: ``True`` when completion span hooks are enabled. + """ + return bool(getattr(self, "completion_detection_spans", False)) + + def _incrementCompletionStat(self, name, amount=1): + """Increment one aggregated completion counter. + + Args: + name (str): Counter key to increment. + amount (int): Value added to the counter. + + Returns: + None: The in-memory stats dictionary is updated when enabled. + """ + if not self._completionStatsEnabled(): + return + with self._completionStatsLock: + self._completionStats[name] = self._completionStats.get(name, 0) + amount + + def _completionStatsSnapshot(self): + """Return a copy of the current completion statistics. + + Returns: + dict: Snapshot of aggregated completion counters. + """ + with self._completionStatsLock: + return dict(self._completionStats) + + def _addMachineHook(self, hook_attr, callback, description): + """Register a machine hook callback on one hook list. + + Args: + hook_attr (str): Attribute name holding the callback list. + callback (callable): Hook function to register. + description (str): Human-readable hook name used in validation + errors. + + Returns: + callable: The registered callback. + """ + if not callable(callback): + raise AtsError("%s hook must be callable" % description) + hooks = getattr(self, hook_attr, None) + if hooks is None: + hooks = [] + setattr(self, hook_attr, hooks) + hooks.append(callback) + return callback + + def _removeMachineHook(self, hook_attr, callback): + """Remove a callback from one machine hook list if present. + + Args: + hook_attr (str): Attribute name holding the callback list. + callback (callable): Hook function to remove. + + Returns: + None: Missing callbacks are ignored. + """ + hooks = getattr(self, hook_attr, None) + if hooks is None: + return + try: + hooks.remove(callback) + except ValueError: + pass + + def add_completion_span_hook(self, callback): + """Register a callback for internal completion-detection timing spans. + + Args: + callback (callable): Function called as + ``callback(name, start_us, end_us, metadata)``. + + Returns: + callable: The registered callback. + """ + return self._addMachineHook( + "_completion_span_hooks", + callback, + "completion span", + ) + + def remove_completion_span_hook(self, callback): + """Unregister a completion span callback. + + Args: + callback (callable): Previously registered completion span hook. + + Returns: + None: Missing callbacks are ignored. + """ + self._removeMachineHook("_completion_span_hooks", callback) + + def add_completion_queue_snapshot_hook(self, callback): + """Register a callback for completion queue depth snapshots. + + Args: + callback (callable): Function called as + ``callback(timestamp_us, metadata)``. + + Returns: + callable: The registered callback. + """ + return self._addMachineHook( + "_completion_queue_snapshot_hooks", + callback, + "completion queue snapshot", + ) + + def remove_completion_queue_snapshot_hook(self, callback): + """Unregister a completion queue snapshot callback. + + Args: + callback (callable): Previously registered queue snapshot hook. + + Returns: + None: Missing callbacks are ignored. + """ + self._removeMachineHook("_completion_queue_snapshot_hooks", callback) + + def _recordCompletionInternalSpan(self, name, start_us, end_us, metadata=None): + """Emit one internal completion-detection timing span. + + Args: + name (str): Span name. + start_us (int): Inclusive start timestamp in microseconds. + end_us (int): End timestamp in microseconds. + metadata (dict|None): Optional structured span metadata. + + Returns: + None: Registered hooks are called when enabled. + """ + if not self._completionSpansEnabled(): + return + for callback in list(getattr(self, "_completion_span_hooks", [])): + callback(name, start_us, end_us, metadata or {}) + + def _recordCompletionQueueSnapshot(self, depth, reason, timestamp_us=None, metadata=None): + """Emit one completion-queue depth snapshot and update queue stats. + + Args: + depth (int): Queue depth after the observed event. + reason (str): Short reason label for the snapshot. + timestamp_us (int|None): Event timestamp in microseconds. Uses the + current time when omitted. + metadata (dict|None): Optional extra snapshot metadata. + + Returns: + None: Registered hooks are called when present. + """ + if timestamp_us is None: + timestamp_us = time.time_ns() // 1000 + depth = max(0, int(depth)) + if self._completionStatsEnabled(): + with self._completionStatsLock: + self._completionStats["completion_queue_depth_latest"] = depth + peak = int(self._completionStats.get("completion_queue_depth_peak", 0)) + if depth > peak: + self._completionStats["completion_queue_depth_peak"] = depth + payload = { + "completion_queue_depth": depth, + "reason": reason, + } + if metadata: + payload.update(metadata) + for callback in list(getattr(self, "_completion_queue_snapshot_hooks", [])): + callback(timestamp_us, payload) + + def _pollChild(self, test): + """Poll one child process and return its current return code. + + Args: + test: ATS test object whose ``child`` process should be polled. + + Returns: + int|None: Child return code, or ``None`` while still running. + """ + if self._completionDetector.owns_child_reaping(): + return test.child.returncode + test.child.poll() + return test.child.returncode + + def _preserve_new_running_tests(self, remaining, seen_ids): + """Keep tests appended to ``self.running`` during a running-state scan. + + Args: + remaining (list): Running tests that should remain after the scan. + seen_ids (set): Object ids already considered in the current pass. + + Returns: + None: ``remaining`` is updated in place. + """ + remaining_ids = {id(test) for test in remaining} + for test in self.running: + test_id = id(test) + if test_id in seen_ids or test_id in remaining_ids: + continue + remaining.append(test) + remaining_ids.add(test_id) + + def scan_running_tests_for_health(self): + """Check still-running tests for timeout and runtime launch failures. + + Returns: + int: Number of running tests completed by health checks. + """ + # Work on a snapshot so completion callbacks can append to + # ``self.running`` without disturbing this scan. + ordered = list(self.running) + seen_ids = {id(test) for test in ordered} + remaining = [] + completed = 0 + for test in ordered: + if getattr(test.child, "returncode", None) is not None: + remaining.append(test) + continue + if self._checkRunningHealth(test): + completed += 1 + continue + remaining.append(test) + self._preserve_new_running_tests(remaining, seen_ids) + self.running = remaining + return completed + + def _finishCompletedTest(self, test): + """Finalize status selection for a child that has already exited. + + Args: + test: ATS test object whose child return code is available. + + Returns: + bool: ``True`` after the completion has been fully handled. + """ + from ats import configuration + + if getattr(test, "ats_returncode_observed_us", None) is None: + test.ats_returncode_observed_us = time.time_ns() // 1000 + test.setEndDateTime() + test.statusCode = test.child.returncode + # If the user set ignoreReturnCode to True then set statusCode to 0. + ignoreReturnCode = test.options.get('ignoreReturnCode', False) + if ignoreReturnCode: + test.statusCode = 0 + if test.statusCode == 0: + status = PASSED + # TODO: move this machine specific check to flux-specific machine files + # This checks for flux timeouts since ATS' method for determining timeouts doesnt work with flux + elif "flux" in configuration.MACHINE_TYPE and test.statusCode == 142: + # Flux reports scheduler-enforced timeouts as return code 142. + status = TIMEDOUT + else: + # Preserve ATS' historical LSF launch/runtime deficiency checks. + lsf_error = False + with open(test.errname, 'r', errors='replace') as f: lines = f.readlines() - f.close + for line in lines: + if lsf_error == False: + if "Terminated while pending" in line: + print("ATS ERROR: Detected LSF Job Start Error %s. Detected LSF launch failure : %s " % (test.name, line)) + lsf_error = True + elif "JSM daemon timed" in line: + print("ATS ERROR: Detected LSF Job Start Error %s. Detected LSF launch failure : %s " % (test.name, line)) + lsf_error = True + elif "Error initializing RM" in line: + print("ATS ERROR: Detected LSF Job Start Error %s. Detected LSF launch failure : %s " % (test.name, line)) + lsf_error = True + elif "Bus error)" in line: + print("ATS ERROR: Halting test %s. Detected Bus Error (perhaps MPI related) : %s " % (test.name, line)) + lsf_error = True + + if not lsf_error: + with open(test.outname, 'r', errors='replace') as f: + lines = f.readlines() for line in lines: if lsf_error == False: - if "Terminated while pending" in line: + if "ATS Error: Locate pipe file" in line: print("ATS ERROR: Detected LSF Job Start Error %s. Detected LSF launch failure : %s " % (test.name, line)) lsf_error = True - elif "JSM daemon timed" in line: - print("ATS ERROR: Detected LSF Job Start Error %s. Detected LSF launch failure : %s " % (test.name, line)) + elif "Could not read jskill" in line: + print("ATS ERROR: Detected LSF Job Scheduler Error %s. : %s " % (test.name, line)) lsf_error = True - #time.sleep(10) # See if sleeiping helps the JSM daemon recover - elif "Error initializing RM" in line: + elif "AST Error: initializing RM" in line: print("ATS ERROR: Detected LSF Job Start Error %s. Detected LSF launch failure : %s " % (test.name, line)) lsf_error = True - #time.sleep(10) # See if sleeiping helps the JSM daemon recover - elif "Bus error)" in line: - print("ATS ERROR: Halting test %s. Detected Bus Error (perhaps MPI related) : %s " % (test.name, line)) - lsf_error = True - if not lsf_error: - f = open(test.outname, 'r', errors='replace') - lines = f.readlines() - f.close - for line in lines: - if lsf_error == False: - if "ATS Error: Locate pipe file" in line: - print("ATS ERROR: Detected LSF Job Start Error %s. Detected LSF launch failure : %s " % (test.name, line)) - lsf_error = True - elif "Could not read jskill" in line: - print("ATS ERROR: Detected LSF Job Scheduler Error %s. : %s " % (test.name, line)) - lsf_error = True - elif "AST Error: initializing RM" in line: - print("ATS ERROR: Detected LSF Job Start Error %s. Detected LSF launch failure : %s " % (test.name, line)) - lsf_error = True - - - #sys.exit(-1) SAD ambyr - #print "DEBUG getStatus 420 statusCode is %d " % test.statusCode - if lsf_error: - print("ATS LSF Development: LSFE Detected statusCode is %d " % test.statusCode) - test.statusCode = 2 - test.setEndDateTime() - status = LSFERROR + if lsf_error: + print("ATS LSF Development: LSFE Detected statusCode is %d " % test.statusCode) + test.statusCode = 2 + test.setEndDateTime() + status = LSFERROR + else: + status = FAILED - else: - status = FAILED + return self._completeTest(test, status) + + def _completeTest(self, test, status): + """Run completion bookkeeping for a finished test. + Args: + test: ATS test object that has finished. + status: ATS status object chosen for the finished test. - # Send test's stdout/stderr to file and to terminal + Returns: + bool: Always ``True`` after completion handling runs. + """ if test.stdOutLocGet() == 'both': outhandle, errhandle = test.fileHandleGet() for line in test.child.stdout: print(line) print(line, file=outhandle) + self._completionDetector.unregister_finished_test(test) self.testEnded(test, status) + return True - #if hasattr(test, 'runningWithinSalloc'): - # if test.runningWithinSalloc == True: - # print "DEBUG Sleeping 1 sec after job end %s" % test.name - # time.sleep(1) + # TODO: move Slurm specific checks to slurm machine file + def _detectRunningSlurmError(self, test): + """Check a still-running test for known SLURM launch/runtime failures. - return True + Args: + test: ATS test object whose stderr should be inspected. + + Returns: + bool: ``True`` when a known SLURM-related fatal error was found. + """ + with open(test.errname, 'r', errors='replace') as f: + lines = f.readlines() + for line in lines: + if "Slurmd could not set up environment for batch job" in line: + print("ATS Halting test %s. Detected slurm launch failure : %s " % (test.name, line)) + return True + elif "srun: error: Unable to create job step" in line: + print("ATS Halting test %s. Detected slurm error : %s " % (test.name, line)) + return True + elif "Error opening remote shared memory object in shm_open" in line: + print("ATS Halting test %s. Detected MPI shared memory failure : %s " % (test.name, line)) + return True + elif "PSM could not set up shared memory segment" in line: + print("ATS Halting test %s. Detected MPI shared memory failure : %s " % (test.name, line)) + return True + elif "Attempting to use an MPI routine before initializing MPICH" in line: + print("ATS Halting test %s. Detected MPI Error : %s " % (test.name, line)) + return True + elif "Bus error)" in line: + print("ATS Halting test %s. Detected Bus Error (perhaps MPI related) : %s " % (test.name, line)) + return True + return False def testEnded(self, test, status): """Do book-keeping when a job has exited; @@ -660,8 +968,8 @@ def _launch(self, test): else: test.child = subprocess.Popen(test.commandList, cwd=test.directory, stdout = subprocess.PIPE, stderr=subprocess.STDOUT, env=E, stdin=testStdin) + self._completionDetector.register_launched_test(test) test.set(RUNNING, test.commandLine) - self.running.append(test) self.numberTestsRunning += 1 if MachineCore.debugClass or MachineCore.canRunNow_debugClass: @@ -797,12 +1105,17 @@ class Machine (MachineCore): You can call your class anything, just put the correct comment line at the top of your machine. See documentation for porting. """ - def __init__(self, name, npMaxH): + def __init__(self, name, npMaxH, completion_detection_mode=None): """Be sure to call this from child if overridden Initialize this machine. npMax supplied by __init__, hardware limit. If npMax is negative, may be overridden by command line. If positive, is hard upper limit. + +Args: + name (str): ATS machine type label. + npMaxH (int): Hardware processor-slot limit. + completion_detection_mode (str|None): Requested completion detector mode. """ # print "DEBUG Machine:MachineCore %s %d" % (name, npMaxH) @@ -816,6 +1129,9 @@ def __init__(self, name, npMaxH): self.hardLimit = (npMaxH > 0) self.naptime = 0.2 #number of seconds to sleep between checks on running tests. self.running = [] + super(Machine, self).__init__( + completion_detection_mode=completion_detection_mode, + ) self.runOrder = 0 from ats import schedulers self.scheduler = schedulers.StandardScheduler() diff --git a/ats/management.py b/ats/management.py index 0734845..a23b00c 100644 --- a/ats/management.py +++ b/ats/management.py @@ -304,8 +304,11 @@ def finalReport(self): successful_run = True if self.testlist: - log("=================================================\n" - "ATS RESULTS %s""" % datestamp(long_format=True), echo=True) + log( + "=================================================\n" + "ATS RESULTS %s" % datestamp(long_format=True), + echo=True, + ) log('-------------------------------------------------', echo = True) self.report() @@ -315,6 +318,8 @@ def finalReport(self): log("""ATS SUMMARY %s""" % datestamp(long_format=True), echo=True) successful_run = self.summary(log) self._summary2(log) + if hasattr(self, "machine"): + self.machine._completionDetector.logCompletionWarnings(log) return successful_run def finalBanner(self): @@ -734,14 +739,34 @@ def postprocess(self): log("-------------------------------", echo=True) return True - def init(self, clas = '', adder=None, examiner=None): - """This initialization is separate so that unit tests can be done on this module. - For this reason we delay any logging until main is called. - adder and examiner are called in configuration if given to allow user - a chance to add options and see results of option parsing. + def init(self, clas = '', adder=None, examiner=None, + completion_detection_mode=None): + """Initialize ATS configuration and machine state. + + This initialization is separate so that unit tests can exercise this + module without running ``main``. Logging is therefore delayed until + after initialization completes. + + Args: + clas (str): ATS command-line string to parse instead of + ``sys.argv[1:]``. + adder (callable|None): Optional callback that adds parser options + before ATS parses the command line. + examiner (callable|None): Optional callback that inspects parsed + options after initialization. + completion_detection_mode (str|None): Requested completion + detector mode to pass through ATS machine construction. + + Returns: + None: Manager state is updated in place. """ tempfile.tempdir = os.getcwd() - configuration.init(clas, adder, examiner) + configuration.init( + clas, + adder, + examiner, + completion_detection_mode=completion_detection_mode, + ) self.options = configuration.options self.inputFiles = configuration.inputFiles self.machine = configuration.machine diff --git a/docs/source/scheduler_extensions.rst b/docs/source/scheduler_extensions.rst index 5aa7d15..be1e585 100644 --- a/docs/source/scheduler_extensions.rst +++ b/docs/source/scheduler_extensions.rst @@ -49,6 +49,72 @@ A custom scheduler should preserve two invariants: readiness, but it should still call ``machine.canRunNow(test)`` or ``machine.startRun(test)`` before consuming resources. +Completion Detectors +==================== + +ATS machines delegate running-test completion policy to a completion detector. +The detector keeps strategy choice out of ``MachineCore.checkRunning()`` while +reusing the same machine-owned helpers for completion queues, polling, and +aggregated completion statistics. + +ATS ships three detector types: + +* ``ats.completion_waitpid_reaper.WaitpidReaperCompletionDetector`` owns child reaping + with a dedicated ``waitpid`` reaper and records completed tests into a queue. + It has the lowest steady-state overhead, but it still carries the known + ``waitpid(-1)`` race and is unsupported for ``FluxDirect``. +* ``ats.completion_per_test_watcher.PerTestWatcherCompletionDetector`` spawns one + watcher thread per running test. That avoids the ``waitpid(-1)`` race, but + it scales with the number of active children because each child keeps its own + waiting thread. +* ``ats.completion_poll.PollingCompletionDetector`` preserves the plain + sleep-then-poll behavior. It is the simplest comparison baseline, but + completion latency and polling work both scale with the scheduler interval. + +Completion counters and timing spans are opt-in. Machines only update the +aggregated counters when ``completion_detection_stats`` is enabled, and they +only emit internal span hooks when ``completion_detection_spans`` is enabled. + +The ATS initialization path accepts ``completion_detection_mode`` and passes it +through machine construction: + +:: + + import ats + + ats.manager.init( + clas="...", + completion_detection_mode="waitpid_reaper", + ) + +Machine constructors also accept the same argument directly and instantiate the +matching detector: + +:: + + from ats.machines import Machine + + machine = Machine( + "generic", + -1, + completion_detection_mode="waitpid_reaper", + ) + +Custom machine subclasses should pass the mode through to ``Machine`` so the +selection stays explicit at construction time: + +:: + + from ats import machines + + class MyMachine(machines.Machine): + def __init__(self, name, npMaxH, completion_detection_mode="waitpid_reaper"): + super(MyMachine, self).__init__( + name, + npMaxH, + completion_detection_mode=completion_detection_mode, + ) + ReadyWorkSet ============ diff --git a/test/test_completion_detector_examples.py b/test/test_completion_detector_examples.py new file mode 100644 index 0000000..d1269d1 --- /dev/null +++ b/test/test_completion_detector_examples.py @@ -0,0 +1,213 @@ +from collections import deque +import subprocess +import sys +import threading +import time +from types import SimpleNamespace +import unittest + +from ats import configuration +from ats.atsut import AtsError, PASSED +from ats.completion_waitpid_reaper import WaitpidReaperCompletionDetector +from ats.completion_detector import create_completion_detector +from ats.completion_per_test_watcher import PerTestWatcherCompletionDetector +from ats.completion_poll import PollingCompletionDetector +from ats.machines import Machine + +if not hasattr(configuration, "options"): + configuration.options = SimpleNamespace( + oneFailure=False, + verbose=False, + skip=False, + logUsage=False, + removeStartNote=False, + removeEndNote=False, + debug=False, + ) + + +class _DetectorMachineStub: + """Minimal machine stub for completion-detector focused unit tests.""" + + def __init__(self, naptime=0.01): + self.naptime = naptime + self.running = [] + self.completion_detection_mode = "waitpid_reaper" + self.completion_fast_path_drain_limit = 128 + self._completionEvent = threading.Event() + self._completionQueue = deque() + self._completionQueueIds = set() + self._completionQueueLock = threading.Lock() + self.stats = {} + self.get_status_calls = [] + self.health_scan_calls = 0 + + def _incrementCompletionStat(self, name, amount=1): + self.stats[name] = self.stats.get(name, 0) + amount + + def _recordCompletionInternalSpan(self, name, start_us, end_us, metadata=None): + pass + + def _recordCompletionQueueSnapshot(self, depth, reason, timestamp_us=None, metadata=None): + pass + + def getStatus(self, test, allow_running_checks=True): + self.get_status_calls.append((test, allow_running_checks)) + if test.child.returncode is None: + return False + test.status = PASSED + return True + + def scan_running_tests_for_health(self): + self.health_scan_calls += 1 + return 0 + + +class CompletionDetectorExamplesTest(unittest.TestCase): + """Keep the scheduler-extension completion-detector examples executable.""" + + def test_constructor_argument_selects_waitpid_reaper_completion_detector_class(self): + """Constructor selection should instantiate ``WaitpidReaperCompletionDetector``.""" + machine = Machine( + "example", + 1, + completion_detection_mode="waitpid_reaper", + ) + + self.assertEqual(machine.completion_detection_mode, "waitpid_reaper") + self.assertIsInstance( + machine._completionDetector, + WaitpidReaperCompletionDetector, + ) + + def test_constructor_argument_selects_polling_completion_detector_class(self): + """Constructor selection should instantiate ``PollingCompletionDetector``.""" + machine = Machine( + "example", + 1, + completion_detection_mode="poll", + ) + + self.assertEqual(machine.completion_detection_mode, "poll") + self.assertIsInstance( + machine._completionDetector, + PollingCompletionDetector, + ) + + def test_constructor_argument_selects_per_test_watcher_completion_detector_class(self): + """Constructor selection should instantiate ``PerTestWatcherCompletionDetector``.""" + machine = Machine( + "example", + 1, + completion_detection_mode="per_test_watcher", + ) + + self.assertEqual(machine.completion_detection_mode, "per_test_watcher") + self.assertIsInstance( + machine._completionDetector, + PerTestWatcherCompletionDetector, + ) + + def test_per_test_watcher_completion_detector_is_the_default_when_no_mode_is_requested(self): + """Default construction should use the detector without the waitpid reaper race.""" + machine = Machine("example", 1) + + self.assertEqual(machine.completion_detection_mode, "per_test_watcher") + self.assertIsInstance( + machine._completionDetector, + PerTestWatcherCompletionDetector, + ) + + def test_flux_direct_rejects_threaded_completion_modes(self): + """FluxDirect should reject the threaded detector modes.""" + + class FluxDirect: + __module__ = "ats.atsMachines.FutureMachines.flux_direct" + + with self.assertRaisesRegex(AtsError, "unsupported for FluxDirect"): + create_completion_detector(FluxDirect(), "waitpid_reaper") + + with self.assertRaisesRegex(AtsError, "unsupported for FluxDirect"): + create_completion_detector(FluxDirect(), "per_test_watcher") + + def test_waitpid_reaper_sets_child_returncode(self): + """``WaitpidReaperCompletionDetector`` should publish subprocess-style return codes.""" + machine = _DetectorMachineStub() + detector = WaitpidReaperCompletionDetector(machine) + child = subprocess.Popen( + [sys.executable, "-c", "import sys; sys.exit(7)"], + ) + test = SimpleNamespace( + child=child, + ats_completion_signal_us=None, + status=PASSED, + ) + + detector.register_launched_test(test) + self.assertTrue(machine._completionEvent.wait(5.0)) + + deadline = time.time() + 5.0 + while child.returncode is None and time.time() < deadline: + time.sleep(0.01) + + self.assertEqual(child.returncode, 7) + self.assertEqual(detector.drain_completion_queue(), [test]) + + def test_waitpid_reaper_drain_avoids_fallback_completion_rescan(self): + """``WaitpidReaperCompletionDetector`` should only finalize queued completions and scan health.""" + machine = _DetectorMachineStub() + detector = WaitpidReaperCompletionDetector(machine) + queued_test = SimpleNamespace( + child=SimpleNamespace(returncode=0), + ats_completion_signal_us=None, + status=PASSED, + ) + still_running = SimpleNamespace( + child=SimpleNamespace(returncode=None), + ats_completion_signal_us=None, + status=PASSED, + ) + machine.running = [queued_test, still_running] + + detector.record_completion_signal(queued_test) + detector.check_running() + + self.assertEqual(machine.get_status_calls, [(queued_test, False)]) + self.assertEqual(machine.health_scan_calls, 1) + self.assertEqual(machine.running, [still_running]) + + def test_waitpid_reaper_records_unexpected_reaped_children(self): + """``WaitpidReaperCompletionDetector`` should record when waitpid reaps a non-test child.""" + machine = _DetectorMachineStub() + detector = WaitpidReaperCompletionDetector(machine) + + detector._handle_reaped_pid(4242, 0) + + self.assertEqual(machine.stats["completion_queue_reaper_unknown_pid"], 1) + snapshot = detector._unexpectedReapsSnapshot() + self.assertEqual(snapshot["count"], 1) + self.assertEqual(snapshot["samples"][0]["pid"], 4242) + self.assertEqual(snapshot["samples"][0]["outcome"], "exit 0") + + def test_waitpid_reaper_logs_unexpected_completion_reap_warning(self): + """``WaitpidReaperCompletionDetector`` should summarize unexpected reaper events at end of run.""" + machine = _DetectorMachineStub() + detector = WaitpidReaperCompletionDetector(machine) + messages = [] + + def collect(message, **_kwargs): + messages.append(message) + + detector._recordUnexpectedCompletionReap(111, 0) + detector._recordUnexpectedCompletionReap(222, 9) + detector.logCompletionWarnings(collect) + + self.assertEqual(len(messages), 3) + self.assertIn("waitpid_reaper reaped 2 child process(es)", messages[0]) + self.assertIn("pid=111", messages[1]) + self.assertIn("outcome=exit 0", messages[1]) + self.assertIn("pid=222", messages[2]) + self.assertIn("outcome=signal 9", messages[2]) + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_streaming_core_helpers.py b/test/test_streaming_core_helpers.py new file mode 100644 index 0000000..f388bf1 --- /dev/null +++ b/test/test_streaming_core_helpers.py @@ -0,0 +1,117 @@ +import types +import unittest + +from ats.atsut import CREATED, PASSED +from ats.management import AtsManager +from ats.schedulers import StandardScheduler + + +class _Group(list): + def __init__(self, number): + """Create a sortable fake ATS group. + + Args: + number (int): Stable group number used for scheduler ordering. + """ + super().__init__() + self.number = number + self.totalPriority = 0 + + def __lt__(self, other): + """Compare fake groups by group number. + + Args: + other (_Group): Group to compare against. + + Returns: + bool: ``True`` when this group should sort before ``other``. + """ + return self.number < other.number + + +class StreamingCoreHelperTest(unittest.TestCase): + def test_streaming_definition_accepts_wrapped_group(self): + """Verify streamed definitions normalize wrapper and list inputs.""" + manager = AtsManager() + tests = [types.SimpleNamespace(serialNumber=1, group=object())] + testcase = types.SimpleNamespace(atsGroup=tests) + + self.assertEqual(manager._streamingTestsFromDefinition(testcase), tests) + self.assertEqual(manager._streamingTestsFromDefinition(tests), tests) + self.assertEqual(manager._streamingTestsFromDefinition("not a test"), []) + + def test_streaming_finalize_waits_adds_missing_parent_once(self): + """Verify streamed wait-finalization adds each live parent once.""" + manager = AtsManager() + child = types.SimpleNamespace(serialNumber=2, waitUntil=[]) + parent = types.SimpleNamespace( + serialNumber=1, + status=CREATED, + dependents=[child], + ) + + manager._streamingFinalizeWaits([child], [parent]) + manager._streamingFinalizeWaits([child], [parent]) + + self.assertEqual(child.waitUntil, [parent]) + + done_parent = types.SimpleNamespace( + serialNumber=3, + status=PASSED, + dependents=[child], + ) + manager._streamingFinalizeWaits([child], [done_parent]) + + self.assertEqual(child.waitUntil, [parent]) + + def test_streaming_distinct_names_match_ats_suffix_style(self): + """Verify streamed duplicate names use the same suffixes as ATS collect.""" + manager = AtsManager() + tests = [ + types.SimpleNamespace(name="sample"), + types.SimpleNamespace(name="SAMPLE"), + types.SimpleNamespace(name="sample"), + ] + + manager._streamingEnsureDistinctNames(tests, {}) + + self.assertEqual([test.name for test in tests], ["sample", "SAMPLE#2", "sample#3"]) + + def test_standard_scheduler_accepts_incremental_interactive_tests(self): + """Verify the default scheduler can load tests after initial load.""" + scheduler = StandardScheduler() + scheduler.groups = [] + scheduled = [] + + def record_schedule(*args): + """Record one scheduler log message. + + Args: + *args: Positional values passed by ``StandardScheduler``. + + Returns: + None. + """ + scheduled.append(args) + + scheduler.schedule = record_schedule + group = _Group(7) + test = types.SimpleNamespace( + group=group, + name="streamed", + priority=3, + serialNumber=11, + totalPriority=5, + waitUntil=[], + ) + group.append(test) + + self.assertTrue(scheduler.addInteractiveTests([test])) + + self.assertEqual(scheduler.groups, [group]) + self.assertEqual(group.totalPriority, 5) + self.assertEqual(len(scheduled), 1) + + +if __name__ == "__main__": + unittest.main()