From aa22198d4af4d7fb5f69b47ebed67a8c66db89cf Mon Sep 17 00:00:00 2001 From: "Peter B. Robinson" Date: Tue, 19 May 2026 10:43:00 -0700 Subject: [PATCH 1/6] add test registration hooks to support streaming test discovery --- ats/cwd.py | 19 +++++ ats/machines.py | 29 +++---- ats/management.py | 121 ++++++++++++++++----------- docs/source/scheduler_extensions.rst | 35 ++++++++ 4 files changed, 137 insertions(+), 67 deletions(-) create mode 100644 ats/cwd.py diff --git a/ats/cwd.py b/ats/cwd.py new file mode 100644 index 0000000..030defc --- /dev/null +++ b/ats/cwd.py @@ -0,0 +1,19 @@ +"""Thread-safe helpers for process current-working-directory changes.""" +from contextlib import contextmanager +import os +import threading + + +_cwd_lock = threading.RLock() + + +@contextmanager +def chdir(path): + """Temporarily change process cwd while holding the global cwd lock.""" + with _cwd_lock: + here = os.getcwd() + os.chdir(path) + try: + yield + finally: + os.chdir(here) diff --git a/ats/machines.py b/ats/machines.py index 2915f49..6a42374 100644 --- a/ats/machines.py +++ b/ats/machines.py @@ -3,6 +3,7 @@ import subprocess, sys, os, time, shlex from ats.atsut import RUNNING, TIMEDOUT, PASSED, FAILED, LSFERROR, \ SKIPPED, HALTED, AtsError +from ats.cwd import chdir from ats.log import log, terminal from shutil import copytree, ignore_patterns @@ -291,14 +292,12 @@ def testEnded(self, test, status): verbose = configuration.options.debug if not (globalPostrunScript == "unset"): - here = os.getcwd() - os.chdir(test.directory) - if os.path.exists(globalPostrunScript): - self._executePreOrPostRunScript(globalPostrunScript, test, verbose, globalPostrunScript_outname) - else: - log("ATS ERROR: globalPostrunScript %s not found" % (globalPostrunScript), echo=True) - sys.exit(-1) - os.chdir(here) + with chdir(test.directory): + if os.path.exists(globalPostrunScript): + self._executePreOrPostRunScript(globalPostrunScript, test, verbose, globalPostrunScript_outname) + else: + log("ATS ERROR: globalPostrunScript %s not found" % (globalPostrunScript), echo=True) + sys.exit(-1) self.numberTestsRunning -= 1 if MachineCore.debugClass or MachineCore.canRunNow_debugClass: @@ -557,14 +556,12 @@ def _launch(self, test): if not (globalPrerunScript == "unset"): - here = os.getcwd() - os.chdir(test.directory) - if os.path.exists(globalPrerunScript): - self._executePreOrPostRunScript(globalPrerunScript, test, verbose, globalPrerunScript_outname) - else: - log("ATS ERROR: globalPrerunScript %s not found" % (globalPrerunScript), echo=True) - sys.exit(-1) - os.chdir(here) + with chdir(test.directory): + if os.path.exists(globalPrerunScript): + self._executePreOrPostRunScript(globalPrerunScript, test, verbose, globalPrerunScript_outname) + else: + log("ATS ERROR: globalPrerunScript %s not found" % (globalPrerunScript), echo=True) + sys.exit(-1) try: Eadd = test.options.get('env', None) diff --git a/ats/management.py b/ats/management.py index 0734845..71e11d6 100644 --- a/ats/management.py +++ b/ats/management.py @@ -7,6 +7,7 @@ from ats.tests import AtsTest from ats.log import log, terminal from ats.parser import AtsCodeParser, AtsFileParser +from ats.cwd import chdir def standardIntrospection(line): "Standard magic detector for input." @@ -38,6 +39,7 @@ class AtsManager(object): * collectTimeEnded -- when test collection was done * onCollected -- just after test collection ends * on Prioritized -- just after test totalPriority has been assigned + * testDefinedRoutines -- list of routines for streaming discovery callbacks * onExitRoutines -- list of routines for onExit to call * onResultsRoutines -- list of routines for onResults to call * continuationFileName -- "continue.ats" if written @@ -58,6 +60,7 @@ def restart(self): self.badlist = [] self.onCollectedRoutines = [] self.onPrioritizedRoutines = [] + self.testDefinedRoutines = [] self.onExitRoutines = [] self.beforeRunRoutines = [] self.onResultsRoutines = [] @@ -186,7 +189,6 @@ def source(self, *paths, **vocabulary): def _source(self, path, introspector, vocabulary): "Process source file. Returns true if successful" - here = os.getcwd() t = abspath(path) directory, filename = os.path.split(t) name, e = os.path.splitext(filename) @@ -229,58 +231,56 @@ def _source(self, path, introspector, vocabulary): if magic is not None: atstext.append(magic) f.close() - if atstext: - log('-> Executing statements in', t1, echo=False) - log.indent() - code = '\n'.join(atstext) - if debug(): - for line in atstext: - log(line, echo=False) - os.chdir(directory) - try: - exec(code, testenv) - # parser = AtsCodeParser(code) - # for code_segment in parser.get_code_iterator(): - # exec(code_segment, testenv) - if debug(): - log('Finished ', t1, datestamp()) - except KeyboardInterrupt: - raise - except Exception as details: - self.badlist.append(t1) - log('ATS ERROR while processing statements in', t1, ':', echo=True) - log(details, echo=True) - log.dedent() - else: - log('-> Sourcing', t1, echo=False) - log.indent() - os.chdir(directory) - try: - exec(compile(open(t1, "rb").read(), t1, 'exec'), testenv) - # parser = AtsFileParser(t1) - # for code_segment in parser.get_code_iterator(): - # exec(code_segment, testenv) + with chdir(directory): + if atstext: + log('-> Executing statements in', t1, echo=False) + log.indent() + code = '\n'.join(atstext) if debug(): - log('Finished ', t1, datestamp()) - - result = 1 - except KeyboardInterrupt: - raise - except Exception as details: - self.badlist.append(t1) - log('ATS ERROR in input file', t1, ':', echo=True) - exc_type, exc_value, exc_traceback = sys.exc_info() - log(traceback.print_exception(exc_type, exc_value, exc_traceback), echo=True) - log('------------------------------------------', echo=True) + for line in atstext: + log(line, echo=False) + try: + exec(code, testenv) + # parser = AtsCodeParser(code) + # for code_segment in parser.get_code_iterator(): + # exec(code_segment, testenv) + if debug(): + log('Finished ', t1, datestamp()) + except KeyboardInterrupt: + raise + except Exception as details: + self.badlist.append(t1) + log('ATS ERROR while processing statements in', t1, ':', echo=True) + log(details, echo=True) + log.dedent() + else: + log('-> Sourcing', t1, echo=False) + log.indent() + try: + exec(compile(open(t1, "rb").read(), t1, 'exec'), testenv) + # parser = AtsFileParser(t1) + # for code_segment in parser.get_code_iterator(): + # exec(code_segment, testenv) + if debug(): + log('Finished ', t1, datestamp()) - log.dedent() - AtsTest.endGroup() - unstick() - stick(**savestuck) - untack() - tack(**savetacked) - AtsTest.waitEndSource() - os.chdir(here) + result = 1 + except KeyboardInterrupt: + raise + except Exception as details: + self.badlist.append(t1) + log('ATS ERROR in input file', t1, ':', echo=True) + exc_type, exc_value, exc_traceback = sys.exc_info() + log(traceback.print_exception(exc_type, exc_value, exc_traceback), echo=True) + log('------------------------------------------', echo=True) + + log.dedent() + AtsTest.endGroup() + unstick() + stick(**savestuck) + untack() + tack(**savetacked) + AtsTest.waitEndSource() def onCollected(self, routine): "Call routine after collection with argument manager." @@ -290,6 +290,25 @@ def onPrioritized(self, routine): "Call routine after collection with argument manager." self.onPrioritizedRoutines.append(routine) + def add_test_defined_hook(self, routine): + """Call routine when streaming discovery publishes a completed definition.""" + if not callable(routine): + raise AtsError("test-defined hook must be callable") + self.testDefinedRoutines.append(routine) + return routine + + def remove_test_defined_hook(self, routine): + """Remove a previously registered streaming discovery hook.""" + try: + self.testDefinedRoutines.remove(routine) + except ValueError: + pass + + def test_defined(self, test_definition): + """Publish a completed test or test group to streaming discovery hooks.""" + for routine in list(self.testDefinedRoutines): + routine(test_definition) + def onExit(self, routine): "Call postprocessing routine before exiting with argument manager." self.onExitRoutines.append(routine) diff --git a/docs/source/scheduler_extensions.rst b/docs/source/scheduler_extensions.rst index 5aa7d15..dfcfd5d 100644 --- a/docs/source/scheduler_extensions.rst +++ b/docs/source/scheduler_extensions.rst @@ -49,6 +49,25 @@ A custom scheduler should preserve two invariants: readiness, but it should still call ``machine.canRunNow(test)`` or ``machine.startRun(test)`` before consuming resources. +Lifecycle Hooks +=============== + +``manager.add_test_defined_hook(callback)`` + Called when a completed test definition or group is published during + streaming discovery. The callback receives the object passed to + ``manager.test_defined(value)``. + +``manager.remove_test_defined_hook(callback)`` + Removes a previously registered test-defined callback. Removing a callback + that is no longer registered is a no-op. + +``manager.test_defined(value)`` + Publishes a completed definition to all currently registered callbacks. + +Drivers that install hooks around one run should unregister them during +cleanup. Hook bodies should stay short and hand work to the driver's main +scheduling thread when scheduler or machine state needs to change. + ReadyWorkSet ============ @@ -279,6 +298,22 @@ periodic reports, and a cheap "work remains" check. The example shows the division of labor: the ready set stores candidates; the scheduler owns dependency and block policy; the machine owns resource admission. +Beginning Tutorial: Streaming Discovery +======================================= + +Streaming discovery overlaps expensive input parsing with test execution. The +safe pattern is single-producer discovery plus main-thread scheduling: + +1. The driver registers ``manager.add_test_defined_hook``. +2. A discovery thread calls ``manager.collectTests()``. +3. Test definitions are pushed into a thread-safe queue by the hook. +4. The main thread drains the queue, normalizes ATS dependencies and names, and + hands completed interactive tests to the scheduler. +5. Only the main thread calls scheduler or machine methods. + +This pattern lets an allocation start useful work earlier while preserving ATS +machine and scheduler state on one thread. + Design Checklist ================ From cdd16cccfa3c6e72c4aaf42b8f2f3a6586fe36f2 Mon Sep 17 00:00:00 2001 From: "Peter B. Robinson" Date: Tue, 19 May 2026 14:15:17 -0700 Subject: [PATCH 2/6] bring the threaded test discovery support from aleats into ats --- ats/management.py | 304 +++++++++++++++++++++++---- ats/schedulers.py | 20 ++ docs/source/scheduler_extensions.rst | 51 ++++- 3 files changed, 322 insertions(+), 53 deletions(-) diff --git a/ats/management.py b/ats/management.py index 71e11d6..36b42a7 100644 --- a/ats/management.py +++ b/ats/management.py @@ -1,4 +1,4 @@ -import os, re, sys, time, tempfile, traceback, socket +import os, re, sys, time, tempfile, traceback, socket, queue, threading from ats import configuration, version from ats.atsut import INVALID, PASSED, FAILED, SKIPPED, BATCHED, LSFERROR, \ RUNNING, FILTERED, CREATED, TIMEDOUT, HALTED, EXPECTED,\ @@ -819,10 +819,9 @@ def firstBanner(self): log('Default time limit for each test=', Duration(configuration.timelimit)) - def core(self): - "This is the 'guts' of ATS." - - if configuration.SYS_TYPE == "toss_3_x86_64": + def _checkCoreMachinePolicy(self): + """Validate machine policy that applies before collection or execution.""" + if configuration.SYS_TYPE == "toss_4_x86_64": if configuration.options.bypassSerialMachineCheck == False: log("**********************************************************************************", echo=True) log("*** This is a serial machine --- Do not use ATS on more than 1 node here! ***", echo=True) @@ -831,6 +830,197 @@ def core(self): log("**********************************************************************************", echo=True) sys.exit(-1) + def _runOnCollectedRoutines(self, keyboard_message): + """Run registered onCollected callbacks and return whether they succeeded.""" + try: + for f in self.onCollectedRoutines: + log("Calling onCollected routine", f.__name__, echo=self.verbose) + f(self) + except KeyboardInterrupt: + log(keyboard_message, echo=True) + return False + except Exception: + log("Error in user-specified onCollected routine.", echo=True) + log(traceback.format_exc(), echo=True) + return False + return True + + def _dispatchBatchTests(self, batchTests): + """Load batch tests through the batch machine if one is active.""" + if not self.batchmachine or not batchTests: + return True + if configuration.options.skip: + log("Skipping execution due to --skip") + return True + try: + log("Sending %d tests to %s." % (len(batchTests), self.batchmachine.name), echo=True) + self.batchmachine.load(batchTests) + except AtsError: + log(traceback.format_exc(), echo=True) + log("ATS ERROR.", echo=True) + return False + except KeyboardInterrupt: + log("Keyboard interrupt while dispatching batch, terminating.", echo=True) + return False + return True + + def _killRunningTests(self): + """Give running tests a moment to exit, then kill any survivors.""" + time.sleep(3) + for test in self.testlist: + if test.status is RUNNING: + self.machine.kill(test) + + def _finishCoreRun(self, interactiveTests, dieDieDie, found_tests=True, batchTests=None): + """Shared shutdown, no-test handling, and continuation-file generation.""" + if dieDieDie: + self._killRunningTests() + + self.machine.quit() #machine shutdown / cleanup + + if not found_tests and not (batchTests or []): + log("No tests found.", echo=True) + return False + + self.continuationFile(interactiveTests) + return not dieDieDie + + def core(self, stream=False): + "This is the 'guts' of ATS." + + self._checkCoreMachinePolicy() + + if stream: + discovery_queue = queue.Queue() + discovery_done = threading.Event() + discovery_errors = [] + all_interactive_tests = [] + all_batch_tests = [] + name_counts = {} + on_collected_ran = False + + def testDefined(test_definition): + """Called by test construction on the discovery thread.""" + discovery_queue.put(test_definition) + + def discoverTests(): + """Collect tests and report any failure back to the scheduler thread.""" + try: + self.collectTests() + except BaseException: + discovery_errors.append(traceback.format_exc()) + finally: + self.collectTimeEnded = datestamp(long_format=True) + discovery_done.set() + + self.add_test_defined_hook(testDefined) + discovery_thread = threading.Thread(target=discoverTests, name="ats-test-discovery") + discovery_thread.daemon = True + discovery_thread.start() + + scheduler = self.machine.scheduler + scheduler.prioritize([]) + scheduler.load([]) + self.preprocess() + + log("Beginning test executions") + timeStatusReport = time.time() + if configuration.options.continueFreq is not None: + timeContinuation = time.time() + continuationStep = int(configuration.options.continueFreq * 60) + else: + timeContinuation = None + continuationStep = None + + found_tests = False + dieDieDie = False + + try: + while True: + block_for_discovery = ( + not discovery_done.is_set() + and self.machine.numberTestsRunning == 0 + and not getattr(scheduler, "groups", []) + ) + events = self._streamingDrainDiscoveryQueue( + discovery_queue, + block=block_for_discovery, + timeout=getattr(self.machine, "naptime", 0.2), + ) + for test_definition in events: + tests = self._streamingTestsFromDefinition(test_definition) + if not tests: + continue + found_tests = True + self._streamingFinalizeWaits(tests, self.testlist) + self._streamingEnsureDistinctNames(tests, name_counts) + + interactiveTests = [t for t in tests if t.status is CREATED] + batchTests = [t for t in tests if t.status is BATCHED] + if interactiveTests: + all_interactive_tests.extend(interactiveTests) + scheduler.addInteractiveTests(interactiveTests) + if batchTests: + all_batch_tests.extend(batchTests) + if not self._dispatchBatchTests(batchTests): + dieDieDie = True + + if discovery_done.is_set() and not on_collected_ran: + on_collected_ran = True + if not self._runOnCollectedRoutines( + "Keyboard interrupt while collecting tests, terminating." + ): + dieDieDie = True + + if dieDieDie or discovery_errors: + break + + timeNow = time.time() + timePassed = timeNow - timeStatusReport + if timePassed >= configuration.options.reportFreq * 60: + terminal("ATS REPORT AT ELAPSED TIME", wallTime()) + timeStatusReport = timeNow + self.summary(terminal) + scheduler.periodicReport() + + if configuration.options.continueFreq is not None: + timeNow = time.time() + if (timeNow - timeContinuation) >= continuationStep: + self.continuationFile(all_interactive_tests, True) + timeContinuation = timeNow + + unfinished = False + if getattr(scheduler, "groups", []): + unfinished = scheduler.step() + elif self.machine.numberTestsRunning > 0: + self.machine.checkRunning() + unfinished = self.machine.numberTestsRunning > 0 + + if discovery_done.is_set() and discovery_queue.empty() and not unfinished: + break + except AtsError: + log(traceback.format_exc(), echo=True) + log("ATS ERROR. Removing running jobs....", echo=True) + dieDieDie = True + except KeyboardInterrupt: + log("Keyboard interrupt. Removing running jobs....", echo=True) + dieDieDie = True + finally: + discovery_thread.join() + self.remove_test_defined_hook(testDefined) + + if discovery_errors: + for error in discovery_errors: + log(error, echo=True) + dieDieDie = True + + return self._finishCoreRun( + all_interactive_tests, + dieDieDie, + found_tests=found_tests, + batchTests=all_batch_tests, + ) + # Phase 1 -- collect the tests errorOccurred = False try: # surround with keyboard interrupt, AtsError handlers @@ -850,20 +1040,9 @@ def core(self): if errorOccurred: return False - try: - for f in self.onCollectedRoutines: - log("Calling onCollected routine", f.__name__, - echo=self.verbose) - f(self) - except KeyboardInterrupt: - log("Keyboard interrupt while collecting tests, terminating.", - echo=True) - errorOccurred = True - except Exception: - log("Error in user-specified onCollected routine.", echo=True) - log(traceback.format_exc(), echo=True) - errorOccured = True - if errorOccurred: + if not self._runOnCollectedRoutines( + "Keyboard interrupt while collecting tests, terminating." + ): return False # divide into interactive and batch tests @@ -879,21 +1058,8 @@ def core(self): # Phase 2 -- dispatch the batch tests - if self.batchmachine and batchTests: - if configuration.options.skip: - log("Skipping execution due to --skip") - else: - try: - log("Sending %d tests to %s." % (len(batchTests), self.batchmachine.name), - echo = True) - self.batchmachine.load(batchTests) - except AtsError: - log(traceback.format_exc(), echo=True) - log("ATS ERROR.", echo=True) - return False - except KeyboardInterrupt: - log("Keyboard interrupt while dispatching batch, terminating.", echo=True) - return False + if not self._dispatchBatchTests(batchTests): + return False # Phase 3 -- run the interactive tests @@ -914,7 +1080,7 @@ def core(self): except Exception: log("ATS ERROR in prioritizing tests.", echo=True) log(traceback.format_exc(), echo=True) - errorOccured = True + errorOccurred = True if errorOccurred: return False @@ -929,14 +1095,6 @@ def core(self): dieDieDie = True log("Keyboard interrupt. Removing running jobs....", echo=True) - if dieDieDie: - time.sleep(3) - for test in self.testlist: - if (test.status is RUNNING): - self.machine.kill(test) - - self.machine.quit() #machine shutdown / cleanup - # Phase 4 -- Continuation file # for t in interactiveTests: # if t.status not in (PASSED, EXPECTED, FILTERED): @@ -945,10 +1103,64 @@ def core(self): # self.continuationFileName = '' # return - self.continuationFile(interactiveTests) - - return True + return self._finishCoreRun(interactiveTests, dieDieDie) + + def _streamingTestsFromDefinition(self, test_definition): + """Return ATS tests represented by a streamed test-defined event.""" + if test_definition is None: + return [] + test_group = getattr(test_definition, "atsGroup", None) + if test_group is not None: + return list(test_group) + if isinstance(test_definition, AtsTest): + return [test_definition] + if hasattr(test_definition, "serialNumber") and hasattr(test_definition, "group"): + return [test_definition] + if isinstance(test_definition, str): + return [] + try: + return list(test_definition) + except TypeError: + pass + return [] + + def _streamingFinalizeWaits(self, tests, parent_candidates=None): + """Apply ATS parent wait edges before streamed tests reach the scheduler.""" + streamed_serials = {test.serialNumber for test in tests} + parents = list(parent_candidates) if parent_candidates is not None else list(tests) + for parent in parents: + if parent.status not in (CREATED, RUNNING): + continue + for dependent in getattr(parent, "dependents", []): + if dependent.serialNumber not in streamed_serials or parent in dependent.waitUntil: + continue + dependent.waitUntil = dependent.waitUntil + [parent] + + def _streamingEnsureDistinctNames(self, tests, name_counts): + """Apply ATS-style duplicate-name suffixes before streamed tests can run.""" + for current in tests: + name = current.name.lower() + count = name_counts.get(name, 0) + 1 + name_counts[name] = count + if count > 1: + current.name += ("#%d" % count) + + def _streamingDrainDiscoveryQueue(self, discovery_queue, block=False, timeout=0.0): + """Return all currently available testcase-discovery events.""" + events = [] + try: + if block: + events.append(discovery_queue.get(timeout=timeout)) + else: + events.append(discovery_queue.get_nowait()) + except queue.Empty: + return events + while True: + try: + events.append(discovery_queue.get_nowait()) + except queue.Empty: + return events def continuationFile(self, interactiveTests, force = False): diff --git a/ats/schedulers.py b/ats/schedulers.py index 73207e6..12a4b09 100644 --- a/ats/schedulers.py +++ b/ats/schedulers.py @@ -60,6 +60,26 @@ def load(self, interactiveTests): self.schedule(msg) return len(self.groups) > 0 + def addInteractiveTests(self, interactiveTests): + """Add newly-discovered interactive tests after the scheduler is loaded.""" + interactiveTests = list(interactiveTests) + if not interactiveTests: + return False + self.calculatePriority(interactiveTests) + for t in interactiveTests: + if t.group not in self.groups: + self.groups.append(t.group) + t.group.totalPriority = t.totalPriority + else: + t.group.totalPriority = max(t.group.totalPriority, t.totalPriority) + self.groups.sort() + + for t in interactiveTests: + msg = "%8d %8d %6d %6d %s" % \ + (t.totalPriority, t.priority, t.serialNumber, t.group.number, t.name) + self.schedule(msg) + return True + def testlist(self): """Return the list of tests in groups that have not yet completed.""" return chain(*self.groups) diff --git a/docs/source/scheduler_extensions.rst b/docs/source/scheduler_extensions.rst index dfcfd5d..fce8402 100644 --- a/docs/source/scheduler_extensions.rst +++ b/docs/source/scheduler_extensions.rst @@ -64,6 +64,11 @@ Lifecycle Hooks ``manager.test_defined(value)`` Publishes a completed definition to all currently registered callbacks. +``manager.core(stream=True)`` + Runs ATS with streaming discovery. This mode collects tests on one worker + thread while the main thread schedules completed definitions published + through ``manager.test_defined``. + Drivers that install hooks around one run should unregister them during cleanup. Hook bodies should stay short and hand work to the driver's main scheduling thread when scheduler or machine state needs to change. @@ -107,7 +112,8 @@ Beginning Tutorial: A Cached Ready Scheduler A scheduler can use ``ReadyWorkSet`` to avoid rescanning every created test on every pass. This example is intentionally small, but it includes the parts needed for a copied scheduler to stay correct: initial indexing, direct -dependent updates, directory-block updates, and launch-failure recovery. +dependent updates, directory-block updates, launch-failure recovery, and +incremental loading from streaming discovery. :: @@ -174,6 +180,19 @@ dependent updates, directory-block updates, and launch-failure recovery. # ready set never decides what CREATED, waits, or blocks mean. self.ready.enqueue_if_ready(test, self.is_ready) + def load(self, interactive_tests): + # ``manager.core()`` calls this once after full collection. The + # streaming path calls it once with an empty list before discovery + # starts, then feeds real work through ``addInteractiveTests``. + self.add_tests(interactive_tests) + return bool(interactive_tests) + + def addInteractiveTests(self, interactive_tests): + # ``manager.core(stream=True)`` calls this on the main thread when a + # completed definition is published with ``manager.test_defined``. + self.add_tests(interactive_tests) + return bool(interactive_tests) + def is_ready(self, test): # "Ready" here means structurally ready for scheduler consideration. # It does not mean resources are currently available; the machine @@ -293,10 +312,27 @@ dependent updates, directory-block updates, and launch-failure recovery. self.ready.enqueue_if_ready(test, self.is_ready) return +Streaming use has two driver-side requirements. The driver installs the +scheduler before entering ATS core, and test-definition code publishes only +completed groups: + +:: + + ats.manager.machine.scheduler = ReadyScheduler() + ats.manager.core(stream=True) + + # In the code that finishes defining one complete group: + ats.manager.test_defined(group) + +``core(stream=True)`` receives the published group, performs the same wait-edge +and duplicate-name normalization that normally happens after collection, and +then calls ``scheduler.addInteractiveTests()`` on the main thread. + Production schedulers also need logging, retry behavior, group-output handling, periodic reports, and a cheap "work remains" check. The example shows the division of labor: the ready set stores candidates; the scheduler owns -dependency and block policy; the machine owns resource admission. +dependency and block policy; ATS core owns streaming discovery; the machine owns +resource admission. Beginning Tutorial: Streaming Discovery ======================================= @@ -304,12 +340,13 @@ Beginning Tutorial: Streaming Discovery Streaming discovery overlaps expensive input parsing with test execution. The safe pattern is single-producer discovery plus main-thread scheduling: -1. The driver registers ``manager.add_test_defined_hook``. -2. A discovery thread calls ``manager.collectTests()``. -3. Test definitions are pushed into a thread-safe queue by the hook. -4. The main thread drains the queue, normalizes ATS dependencies and names, and +1. The driver calls ``manager.core(stream=True)``. +2. ``core(stream=True)`` registers ``manager.add_test_defined_hook``. +3. A discovery thread calls ``manager.collectTests()``. +4. Test definitions are pushed into a thread-safe queue by the hook. +5. The main thread drains the queue, normalizes ATS dependencies and names, and hands completed interactive tests to the scheduler. -5. Only the main thread calls scheduler or machine methods. +6. Only the main thread calls scheduler or machine methods. This pattern lets an allocation start useful work earlier while preserving ATS machine and scheduler state on one thread. From 051d7ffa64aff6c3af95b6d049160d88e17ee28a Mon Sep 17 00:00:00 2001 From: "Peter B. Robinson" Date: Tue, 19 May 2026 15:08:40 -0700 Subject: [PATCH 3/6] add documentation --- ats/cwd.py | 11 ++- ats/management.py | 234 +++++++++++++++++++++++++++++++++++++++++++--- ats/schedulers.py | 12 ++- 3 files changed, 240 insertions(+), 17 deletions(-) diff --git a/ats/cwd.py b/ats/cwd.py index 030defc..08d7a87 100644 --- a/ats/cwd.py +++ b/ats/cwd.py @@ -9,7 +9,16 @@ @contextmanager def chdir(path): - """Temporarily change process cwd while holding the global cwd lock.""" + """Temporarily change process cwd while holding the global cwd lock. + + Args: + path (str): Directory to make the process current-working-directory + while the context is active. + + Yields: + None. The previous current-working-directory is restored when the + context exits. + """ with _cwd_lock: here = os.getcwd() os.chdir(path) diff --git a/ats/management.py b/ats/management.py index 36b42a7..3a787b9 100644 --- a/ats/management.py +++ b/ats/management.py @@ -291,21 +291,48 @@ def onPrioritized(self, routine): self.onPrioritizedRoutines.append(routine) def add_test_defined_hook(self, routine): - """Call routine when streaming discovery publishes a completed definition.""" + """Register a callback for completed streaming-discovery definitions. + + Args: + routine (callable): Function called with the value supplied to + ``test_defined``. The callback should avoid touching scheduler + or machine state from a discovery thread. + + Returns: + callable: The registered callback, matching ATS's decorator-friendly + hook style. + """ if not callable(routine): raise AtsError("test-defined hook must be callable") self.testDefinedRoutines.append(routine) return routine def remove_test_defined_hook(self, routine): - """Remove a previously registered streaming discovery hook.""" + """Remove a previously registered streaming-discovery callback. + + Args: + routine (callable): Callback previously passed to + ``add_test_defined_hook``. + + Returns: + None. + """ try: self.testDefinedRoutines.remove(routine) except ValueError: pass def test_defined(self, test_definition): - """Publish a completed test or test group to streaming discovery hooks.""" + """Publish one completed test definition to streaming-discovery hooks. + + Args: + test_definition (object): Completed ATS test, iterable of ATS tests, + or application-defined wrapper object that a registered hook can + interpret. + + Returns: + None. + """ for routine in list(self.testDefinedRoutines): routine(test_definition) @@ -820,7 +847,12 @@ def firstBanner(self): Duration(configuration.timelimit)) def _checkCoreMachinePolicy(self): - """Validate machine policy that applies before collection or execution.""" + """Validate machine policy that applies before collection or execution. + + Returns: + None. Exits the process with an ATS error banner when the current + machine/options combination is not allowed. + """ if configuration.SYS_TYPE == "toss_4_x86_64": if configuration.options.bypassSerialMachineCheck == False: log("**********************************************************************************", echo=True) @@ -831,7 +863,16 @@ def _checkCoreMachinePolicy(self): sys.exit(-1) def _runOnCollectedRoutines(self, keyboard_message): - """Run registered onCollected callbacks and return whether they succeeded.""" + """Run registered ``onCollected`` callbacks. + + Args: + keyboard_message (str): Message to log when callback execution is + interrupted with ``KeyboardInterrupt``. + + Returns: + bool: ``True`` when every callback completed; ``False`` when a + callback raised an exception or was interrupted. + """ try: for f in self.onCollectedRoutines: log("Calling onCollected routine", f.__name__, echo=self.verbose) @@ -846,7 +887,16 @@ def _runOnCollectedRoutines(self, keyboard_message): return True def _dispatchBatchTests(self, batchTests): - """Load batch tests through the batch machine if one is active.""" + """Load batch tests through the batch machine if one is active. + + Args: + batchTests (iterable): ATS tests whose status is ``BATCHED``. + + Returns: + bool: ``True`` when there is no batch work, batch dispatch is + skipped by options, or batch loading succeeds; ``False`` when ATS + should stop because dispatch failed or was interrupted. + """ if not self.batchmachine or not batchTests: return True if configuration.options.skip: @@ -865,14 +915,36 @@ def _dispatchBatchTests(self, batchTests): return True def _killRunningTests(self): - """Give running tests a moment to exit, then kill any survivors.""" + """Give running tests a moment to exit, then kill any survivors. + + Returns: + None. + """ time.sleep(3) for test in self.testlist: if test.status is RUNNING: self.machine.kill(test) def _finishCoreRun(self, interactiveTests, dieDieDie, found_tests=True, batchTests=None): - """Shared shutdown, no-test handling, and continuation-file generation.""" + """Shared shutdown, no-test handling, and continuation-file generation. + + Args: + interactiveTests (iterable): Interactive tests known to this core + invocation. In streaming mode this is the accumulated list of + streamed interactive tests. + dieDieDie (bool): Whether ATS should kill running tests and return + failure because the run was interrupted or hit a fatal error. + found_tests (bool): Whether discovery found at least one runnable + or batchable test definition. + batchTests (iterable or None): Batch tests known to this core + invocation. Used only to distinguish "no tests found" from a + batch-only run. + + Returns: + bool: ``True`` when the core invocation completed without fatal + interruption; ``False`` when no tests were found or cleanup followed + a fatal interruption. + """ if dieDieDie: self._killRunningTests() @@ -886,25 +958,64 @@ def _finishCoreRun(self, interactiveTests, dieDieDie, found_tests=True, batchTes return not dieDieDie def core(self, stream=False): - "This is the 'guts' of ATS." + """Run the core ATS collect/schedule/execute lifecycle. + + Args: + stream (bool): When ``False``, preserve the classic ATS behavior: + collect every input file before sorting and running tests. When + ``True``, collect tests on a discovery thread and schedule + completed definitions on the main thread as they are published + through ``test_defined`` hooks. + + Returns: + bool: ``True`` when ATS completed the run successfully; ``False`` + when collection, callback handling, dispatch, execution, or final + cleanup failed. + """ self._checkCoreMachinePolicy() if stream: + # Streaming core is built around a single producer/single consumer + # handoff. The discovery thread is the only producer and never + # touches scheduler or machine state directly. discovery_queue = queue.Queue() discovery_done = threading.Event() discovery_errors = [] + + # These accumulated lists replace the all-at-once + # ``sortTests()`` results from classic core. They are also the + # authoritative inputs for continuation-file generation. all_interactive_tests = [] all_batch_tests = [] + + # Classic ``collectTests()`` applies duplicate-name suffixes after + # every file is sourced. Streaming has to apply the same rule + # incrementally before a test can be handed to the scheduler. name_counts = {} on_collected_ran = False def testDefined(test_definition): - """Called by test construction on the discovery thread.""" + """Queue one completed definition from the discovery thread. + + Args: + test_definition (object): An ATS test, an iterable of ATS + tests, or an application wrapper with an ``atsGroup`` + attribute. + + Returns: + None. + """ discovery_queue.put(test_definition) def discoverTests(): - """Collect tests and report any failure back to the scheduler thread.""" + """Collect tests and report any failure back to the scheduler thread. + + Returns: + None. Discovery errors are captured in + ``discovery_errors`` so the main thread can log and clean up + from the normal ATS execution path. + """ try: self.collectTests() except BaseException: @@ -913,18 +1024,31 @@ def discoverTests(): self.collectTimeEnded = datestamp(long_format=True) discovery_done.set() + # Register the handoff hook before starting discovery so that + # definitions published by the first sourced file cannot be missed. self.add_test_defined_hook(testDefined) discovery_thread = threading.Thread(target=discoverTests, name="ats-test-discovery") discovery_thread.daemon = True discovery_thread.start() + # Initialize scheduler logging and state before the first streamed + # event arrives. Schedulers that cache state should treat this as + # an empty initial load and expect real tests through + # ``addInteractiveTests``. scheduler = self.machine.scheduler scheduler.prioritize([]) scheduler.load([]) + + # Preprocess hooks may clean logs or prepare run state. Run them + # on the main thread before any streamed test can launch. self.preprocess() log("Beginning test executions") timeStatusReport = time.time() + + # Continuation timing mirrors ``run()``. Streaming uses the + # accumulated interactive list because the final suite is not known + # when the loop begins. if configuration.options.continueFreq is not None: timeContinuation = time.time() continuationStep = int(configuration.options.continueFreq * 60) @@ -937,6 +1061,9 @@ def discoverTests(): try: while True: + # When no work is known and no test is running, block + # briefly for discovery. Otherwise drain without blocking + # so the scheduler keeps making progress. block_for_discovery = ( not discovery_done.is_set() and self.machine.numberTestsRunning == 0 @@ -948,10 +1075,18 @@ def discoverTests(): timeout=getattr(self.machine, "naptime", 0.2), ) for test_definition in events: + # Application drivers may publish a wrapper object + # instead of raw ATS tests. Normalize that surface once + # here so the rest of the streaming loop remains ATS + # test oriented. tests = self._streamingTestsFromDefinition(test_definition) if not tests: continue found_tests = True + + # Classic core finalizes wait edges and duplicate names + # after full collection. Streaming does the same work + # just-in-time for this completed definition. self._streamingFinalizeWaits(tests, self.testlist) self._streamingEnsureDistinctNames(tests, name_counts) @@ -959,12 +1094,23 @@ def discoverTests(): batchTests = [t for t in tests if t.status is BATCHED] if interactiveTests: all_interactive_tests.extend(interactiveTests) + + # This is the only scheduler entry point used by + # streaming discovery for newly-created interactive + # tests. It runs on the main thread. scheduler.addInteractiveTests(interactiveTests) if batchTests: all_batch_tests.extend(batchTests) + + # Batch loading is also kept on the main thread for + # consistency with classic core and batch-machine + # implementations. if not self._dispatchBatchTests(batchTests): dieDieDie = True + # ``onCollected`` routines are a post-discovery phase, so + # they cannot run until the discovery thread has completed + # all input files. They still run before final shutdown. if discovery_done.is_set() and not on_collected_ran: on_collected_ran = True if not self._runOnCollectedRoutines( @@ -975,6 +1121,9 @@ def discoverTests(): if dieDieDie or discovery_errors: break + # Periodic reporting is handled here instead of delegating + # to ``run()`` because streaming does not have a fixed + # interactive test list at loop entry. timeNow = time.time() timePassed = timeNow - timeStatusReport if timePassed >= configuration.options.reportFreq * 60: @@ -983,12 +1132,18 @@ def discoverTests(): self.summary(terminal) scheduler.periodicReport() + # Update the continuation file periodically with the tests + # that have been discovered so far. if configuration.options.continueFreq is not None: timeNow = time.time() if (timeNow - timeContinuation) >= continuationStep: self.continuationFile(all_interactive_tests, True) timeContinuation = timeNow + # Advance scheduling only from the main thread. If no + # scheduler groups are known yet but machine work is still + # running, poll completion directly so the loop can drain + # in-flight tests. unfinished = False if getattr(scheduler, "groups", []): unfinished = scheduler.step() @@ -996,6 +1151,9 @@ def discoverTests(): self.machine.checkRunning() unfinished = self.machine.numberTestsRunning > 0 + # Stop after discovery is complete, the handoff queue is + # empty, and neither scheduler nor machine has unfinished + # work. if discovery_done.is_set() and discovery_queue.empty() and not unfinished: break except AtsError: @@ -1006,6 +1164,9 @@ def discoverTests(): log("Keyboard interrupt. Removing running jobs....", echo=True) dieDieDie = True finally: + # Always wait for discovery and unregister the hook before + # leaving streaming core; otherwise later ATS runs in the same + # process could receive stale callbacks. discovery_thread.join() self.remove_test_defined_hook(testDefined) @@ -1106,7 +1267,18 @@ def discoverTests(): return self._finishCoreRun(interactiveTests, dieDieDie) def _streamingTestsFromDefinition(self, test_definition): - """Return ATS tests represented by a streamed test-defined event.""" + """Return ATS tests represented by a streamed test-defined event. + + Args: + test_definition (object): An object published through + ``test_defined``. Supported values are an ATS test, an + iterable of ATS tests, or an application wrapper with an + ``atsGroup`` attribute. + + Returns: + list: ATS tests extracted from ``test_definition``. Unknown objects + and strings return an empty list. + """ if test_definition is None: return [] test_group = getattr(test_definition, "atsGroup", None) @@ -1125,7 +1297,18 @@ def _streamingTestsFromDefinition(self, test_definition): return [] def _streamingFinalizeWaits(self, tests, parent_candidates=None): - """Apply ATS parent wait edges before streamed tests reach the scheduler.""" + """Apply ATS parent wait edges before streamed tests reach the scheduler. + + Args: + tests (iterable): Newly streamed ATS tests that may be waiting on + already-discovered parent tests. + parent_candidates (iterable or None): Tests whose dependents should + be checked for wait edges. When ``None``, only ``tests`` are + considered. + + Returns: + None. + """ streamed_serials = {test.serialNumber for test in tests} parents = list(parent_candidates) if parent_candidates is not None else list(tests) for parent in parents: @@ -1137,7 +1320,17 @@ def _streamingFinalizeWaits(self, tests, parent_candidates=None): dependent.waitUntil = dependent.waitUntil + [parent] def _streamingEnsureDistinctNames(self, tests, name_counts): - """Apply ATS-style duplicate-name suffixes before streamed tests can run.""" + """Apply ATS-style duplicate-name suffixes before streamed tests can run. + + Args: + tests (iterable): Newly streamed ATS tests whose names should be + checked against names already published in this streaming run. + name_counts (dict): Mutable mapping from lower-case test name to + the number of times that name has appeared. + + Returns: + None. Test names and ``name_counts`` are updated in place. + """ for current in tests: name = current.name.lower() count = name_counts.get(name, 0) + 1 @@ -1146,7 +1339,18 @@ def _streamingEnsureDistinctNames(self, tests, name_counts): current.name += ("#%d" % count) def _streamingDrainDiscoveryQueue(self, discovery_queue, block=False, timeout=0.0): - """Return all currently available testcase-discovery events.""" + """Return all currently available testcase-discovery events. + + Args: + discovery_queue (queue.Queue): Queue receiving completed + definitions from the discovery thread. + block (bool): Whether to wait for one event before draining the + queue. + timeout (float): Maximum seconds to wait when ``block`` is true. + + Returns: + list: All events available after the optional initial blocking wait. + """ events = [] try: if block: diff --git a/ats/schedulers.py b/ats/schedulers.py index 12a4b09..1f5ffe4 100644 --- a/ats/schedulers.py +++ b/ats/schedulers.py @@ -61,7 +61,17 @@ def load(self, interactiveTests): return len(self.groups) > 0 def addInteractiveTests(self, interactiveTests): - """Add newly-discovered interactive tests after the scheduler is loaded.""" + """Add newly-discovered interactive tests after the scheduler is loaded. + + Args: + interactiveTests (iterable): ATS tests discovered after the initial + ``load`` call. ``AtsManager.core(stream=True)`` calls this on + the main thread as completed definitions are published. + + Returns: + bool: ``True`` when at least one test was added; ``False`` when the + input iterable was empty. + """ interactiveTests = list(interactiveTests) if not interactiveTests: return False From f12a5d8317a0bb1f58da66031c7cafe26674efe3 Mon Sep 17 00:00:00 2001 From: "Peter B. Robinson" Date: Wed, 20 May 2026 10:00:18 -0700 Subject: [PATCH 4/6] reword test_defined description --- ats/management.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ats/management.py b/ats/management.py index 3a787b9..5c632e8 100644 --- a/ats/management.py +++ b/ats/management.py @@ -323,7 +323,8 @@ def remove_test_defined_hook(self, routine): pass def test_defined(self, test_definition): - """Publish one completed test definition to streaming-discovery hooks. + """Using hook(s) previously registered through add_test_defined_hook, notifies + the driver that a test definition has been added. Args: test_definition (object): Completed ATS test, iterable of ATS tests, From cf118e682f99e5dbca4e05ce4381296771594ebe Mon Sep 17 00:00:00 2001 From: "Peter B. Robinson" Date: Wed, 20 May 2026 10:05:52 -0700 Subject: [PATCH 5/6] revert sys_type change unrelated to this branch --- ats/management.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ats/management.py b/ats/management.py index 5c632e8..a04132c 100644 --- a/ats/management.py +++ b/ats/management.py @@ -854,7 +854,7 @@ def _checkCoreMachinePolicy(self): None. Exits the process with an ATS error banner when the current machine/options combination is not allowed. """ - if configuration.SYS_TYPE == "toss_4_x86_64": + if configuration.SYS_TYPE == "toss_3_x86_64": if configuration.options.bypassSerialMachineCheck == False: log("**********************************************************************************", echo=True) log("*** This is a serial machine --- Do not use ATS on more than 1 node here! ***", echo=True) From b10d111550acd64022d5d243e7277f88abd69108 Mon Sep 17 00:00:00 2001 From: "Peter B. Robinson" Date: Wed, 20 May 2026 10:31:31 -0700 Subject: [PATCH 6/6] update documentation to be a bit less confusing. --- ats/management.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/ats/management.py b/ats/management.py index a04132c..24f7cc0 100644 --- a/ats/management.py +++ b/ats/management.py @@ -1277,8 +1277,9 @@ def _streamingTestsFromDefinition(self, test_definition): ``atsGroup`` attribute. Returns: - list: ATS tests extracted from ``test_definition``. Unknown objects - and strings return an empty list. + list: ATS tests extracted from ``test_definition``. Unknown + objects return an empty list; strings are explicitly ignored so + they are not treated as iterables of tests. """ if test_definition is None: return []