From 52aa2fdc5da0308aa0da8757adca17b4bd787445 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 15:26:13 +0200 Subject: [PATCH 01/21] Extract `runSubprocess` `mp.Process`+`Queue` body into `_run_subprocess_mp` shim Isolates the multiprocessing implementation behind a single callable so later commits can swap it without touching callers. --- btest | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/btest b/btest index 3cc3f12..0f12b6b 100755 --- a/btest +++ b/btest @@ -350,7 +350,7 @@ def runTestCommandLine(cmdline, measure_time, **kwargs): # Runs a subprocess. Takes same arguments as subprocess.check_call() # and returns a 2-tuple (success, rc) where *success* is a boolean # indicating if the command executed, and *rc* is its exit code if it did. -def runSubprocess(*args, **kwargs): +def _run_subprocess_mp(*args, **kwargs): def child(q): try: if sys.platform == "win32": @@ -393,6 +393,10 @@ def runSubprocess(*args, **kwargs): return result +def runSubprocess(*args, **kwargs): + return _run_subprocess_mp(*args, **kwargs) + + # Description of an alternative configuration. class Alternative: DEFAULT = "default" From 552171dd32a13b75da3aac757652c66fcdc50415 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 15:29:15 +0200 Subject: [PATCH 02/21] Replace per-command `mp.Process`+`Queue` with `subprocess.Popen` The process-per-command pattern existed solely to contain `KeyboardInterrupt`. Replace with a plain `Popen`+wait loop that kills the child on interrupt and re-raises, removing the `mp.Queue` and `mp.Process` machinery from the hot path. --- btest | 60 +++++++++++++++++++---------------------------------------- 1 file changed, 19 insertions(+), 41 deletions(-) diff --git a/btest b/btest index 0f12b6b..2900387 100755 --- a/btest +++ b/btest @@ -350,51 +350,29 @@ def runTestCommandLine(cmdline, measure_time, **kwargs): # Runs a subprocess. Takes same arguments as subprocess.check_call() # and returns a 2-tuple (success, rc) where *success* is a boolean # indicating if the command executed, and *rc* is its exit code if it did. -def _run_subprocess_mp(*args, **kwargs): - def child(q): - try: - if sys.platform == "win32": - tmpdir = normalize_path(kwargs.get("cwd", "")) - if len(args) > 1: - cmd = shlex.join(args) - else: - cmd = args[0] - - tf, bash_cmd = _build_win_subprocess_cmd_script(cmd, tmpdir) - with tf: - subprocess.check_call(bash_cmd, **kwargs) - else: - subprocess.check_call(*args, **kwargs) - success = True - rc = 0 - - except subprocess.CalledProcessError as e: - success = False - rc = e.returncode - - except KeyboardInterrupt: - success = False - rc = 0 - - q.put([success, rc]) +def runSubprocess(*args, **kwargs): + if sys.platform == "win32": + tmpdir = normalize_path(kwargs.get("cwd", "")) + if len(args) > 1: + cmd = shlex.join(args) + else: + cmd = args[0] + tf, bash_cmd = _build_win_subprocess_cmd_script(cmd, tmpdir) + with tf: + proc = subprocess.Popen(bash_cmd, **kwargs) + else: + proc = subprocess.Popen(*args, **kwargs) try: - q = mp.Queue() - p = mp.Process(target=child, args=(q,)) - p.start() - result = q.get() - p.join() - + proc.wait() except KeyboardInterrupt: - # Bail out here directly as otherwise we'd get a bunch of errors. - # from all the childs. - sys.exit(1) - - return result + proc.kill() + proc.wait() + raise - -def runSubprocess(*args, **kwargs): - return _run_subprocess_mp(*args, **kwargs) + if proc.returncode != 0: + return (False, proc.returncode) + return (True, 0) # Description of an alternative configuration. From 53a0b61dd801723d0bb53d599547d5cd539a56d5 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 15:34:12 +0200 Subject: [PATCH 03/21] Extract shared runner state into `RunState` class Moves all mutable state (counters, test queue, port pool, output handler, lock) and the methods that operate on it out of `TestManager` into a plain `RunState` class. `TestManager` delegates to it; the `SyncManager` proxy machinery is otherwise unchanged. --- btest | 396 +++++++++++++++++++++++++++++++++------------------------- 1 file changed, 229 insertions(+), 167 deletions(-) diff --git a/btest b/btest index 2900387..1652f54 100755 --- a/btest +++ b/btest @@ -398,127 +398,23 @@ class Abort(Exception): pass -# Main class distributing the work across threads. -class TestManager(mp_managers.SyncManager): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - - self._output_handler = None - self._lock = None - self._succeeded = None - self._failed = None - self._failed_expected = None - self._unstable = None - self._skipped = None - self._tests = None - self._failed_tests = None - self._num_tests = None - self._timing = None - self._ports = None - - def run(self, tests, output_handler): - self.start() - - mgr_data = self.dict() - mgr_data["Alternatives"] = Alternatives - mgr_data["BaselineDirs"] = BaselineDirs - mgr_data["Initializer"] = Initializer - mgr_data["Finalizer"] = Finalizer - mgr_data["Teardown"] = Teardown - mgr_data["Options"] = Options - mgr_data["TestBase"] = TestBase - mgr_data["TmpDir"] = TmpDir - - output_handler.prepare(self) +# Holds the mutable state shared across worker processes during a test run. +# All fields are proxy objects (RLock, list, RawValue) created by the +# SyncManager so workers can safely access them across process boundaries. +class RunState: + def __init__(self, output_handler, lock, tests, failed_tests, ports, timing): self._output_handler = output_handler - self._lock = self.RLock() + self._lock = lock self._succeeded = mp_sharedctypes.RawValue("i", 0) self._failed = mp_sharedctypes.RawValue("i", 0) self._failed_expected = mp_sharedctypes.RawValue("i", 0) self._unstable = mp_sharedctypes.RawValue("i", 0) self._skipped = mp_sharedctypes.RawValue("i", 0) - self._tests = self.list(tests) - self._failed_tests = self.list([]) - self._num_tests = len(self._tests) - self._timing = self.loadTiming() - - port_range = getOption("PortRange", "1024-65535") - port_range_lo = int(port_range.split("-")[0]) - port_range_hi = int(port_range.split("-")[1]) - - if port_range_lo > port_range_hi: - error(f"invalid PortRange value: {port_range}") - - max_test_ports = 0 - test_with_most_ports = None - - for t in self._tests: - if len(t.ports) > max_test_ports: - max_test_ports = len(t.ports) - test_with_most_ports = t - - if max_test_ports > port_range_hi - port_range_lo + 1: - error( - f"PortRange {port_range} cannot satisfy requirement of {max_test_ports} ports in test {test_with_most_ports.name}" - ) - - self._ports = self.list(list(range(port_range_lo, port_range_hi + 1))) - - threads = [] - - # With interactive input possibly required, we run tests - # directly. This avoids noisy output appearing from detached - # processes post-btest-exit when using CTRL-C during the input - # stage. - if Options.mode == "UPDATE_INTERACTIVE": - self.threadRun(0, mgr_data) - else: - try: - # Create a set of processes for running each of the tests. This isn't the actual - # zeek processes, but runner processes executing individual test commands. - for i in range(Options.threads): - t = mp.Process( - name=f"#{i + 1}", target=lambda: self.threadRun(i, mgr_data) - ) - t.start() - threads += [t] - - for t in threads: - t.join() - - except KeyboardInterrupt: - for t in threads: - t.terminate() - t.join() - - if ( - Options.abort_on_failure - and self._failed.value > 0 - and self._failed.value > self._failed_expected.value - ): - # Signal abort. The child processes will already have - # finished because the join() above still ran. - raise Abort("Aborted after first failure.") - - # Record failed tests if not updating. - if Options.mode != "UPDATE" and Options.mode != "UPDATE_INTERACTIVE": - try: - state = open(StateFile, "w", encoding="utf-8") - except OSError: - error(f"cannot open state file {StateFile}") - - for t in sorted(self._failed_tests): - print(t, file=state) - - state.close() - - return ( - self._succeeded.value, - self._failed.value, - self._skipped.value, - self._unstable.value, - self._failed_expected.value, - ) + self._tests = tests + self._failed_tests = failed_tests + self._num_tests = len(tests) + self._timing = timing + self._ports = ports def percentage(self): if not self._num_tests: @@ -527,58 +423,12 @@ class TestManager(mp_managers.SyncManager): count = self._succeeded.value + self._failed.value + self._skipped.value return 100.0 * count / self._num_tests - # Returns true if there's at least one unexpected failure so far. def hasFailure(self): return (self._failed.value - self._failed_expected.value) > 0 - # Returns true if there's at least one unstable test so far. def hasUnstable(self): return self._unstable.value > 0 - # Worker method for each of the "threads" specified by the "-j" argument passed - # at run time. This basically segments the list of tests into chunks and runs - # until we're out of chunks. - def threadRun(self, thread_num, mgr_data): - # This should prevent the child processes from receiving SIGINT signals and - # let the KeyboardInterrupt handler in the manager's run() method handle - # those. - signal.signal(signal.SIGINT, signal.SIG_IGN) - - all_tests = [] - - # Globals get lost moving from the parent to the child on Windows, so we need to use - # the data proxied from the manager to rebuild the dict of globals before continuing. - if sys.platform == "win32": - for global_key, global_value in mgr_data.items(): - globals()[global_key] = global_value - - # multiprocess/dill may reconstruct this method with its own globals - # dict, separate from the module-level globals used by functions like - # replaceEnvs. Also set globals in __mp_main__, which is where - # multiprocess imports the script in spawned child processes. - mp_main = sys.modules.get("__mp_main__") - if mp_main is not None: - for global_key, global_value in mgr_data.items(): - setattr(mp_main, global_key, global_value) - - while True: - # Pull the next test from the list that was built at startup. This may - # be more than one test if there were alternatives requested in the - # arguments passed to btest. - thread_tests = self.nextTests(thread_num) - if thread_tests is None: - # No more work for us. - return - - all_tests += thread_tests - - for t in thread_tests: - t.run(self) - self.testReplayOutput(t) - - if Options.update_times: - self.saveTiming(all_tests) - def rerun(self, test): test.reruns += 1 self._tests += [test.clone(increment=False)] @@ -788,29 +638,24 @@ class TestManager(mp_managers.SyncManager): def testTimingBaseline(self, test): return self._timing.get(test.name, -1) - # Returns the name of the file to store the timing baseline in for this host. def timingPath(self): id = uuid.uuid3(uuid.NAMESPACE_DNS, str(uuid.getnode())) return os.path.abspath(os.path.join(BaselineTimingDir, id.hex)) - # Loads baseline timing information for this host if available. Returns - # empty directory if not. def loadTiming(self): - timing = {} - with self._lock: path = self.timingPath() if not os.path.exists(path): return {} + timing = {} for line in open(path): (k, v) = line.split() timing[k] = float(v) return timing - # Updates the timing baseline for the given tests on this host. def saveTiming(self, tests): with self._lock: changed = False @@ -836,6 +681,223 @@ class TestManager(mp_managers.SyncManager): out.close() +# Main class distributing the work across threads. +class TestManager(mp_managers.SyncManager): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._run_state = None + + def run(self, tests, output_handler): + self.start() + + mgr_data = self.dict() + mgr_data["Alternatives"] = Alternatives + mgr_data["BaselineDirs"] = BaselineDirs + mgr_data["Initializer"] = Initializer + mgr_data["Finalizer"] = Finalizer + mgr_data["Teardown"] = Teardown + mgr_data["Options"] = Options + mgr_data["TestBase"] = TestBase + mgr_data["TmpDir"] = TmpDir + + lock = self.RLock() + proxy_tests = self.list(tests) + proxy_failed_tests = self.list([]) + + port_range = getOption("PortRange", "1024-65535") + port_range_lo = int(port_range.split("-")[0]) + port_range_hi = int(port_range.split("-")[1]) + + if port_range_lo > port_range_hi: + error(f"invalid PortRange value: {port_range}") + + max_test_ports = 0 + test_with_most_ports = None + + for t in proxy_tests: + if len(t.ports) > max_test_ports: + max_test_ports = len(t.ports) + test_with_most_ports = t + + if max_test_ports > port_range_hi - port_range_lo + 1: + error( + f"PortRange {port_range} cannot satisfy requirement of {max_test_ports} ports in test {test_with_most_ports.name}" + ) + + proxy_ports = self.list(list(range(port_range_lo, port_range_hi + 1))) + + # loadTiming needs a RunState with a lock to guard file access. + # Create a temporary state just to load timing before the full state is ready. + tmp_state = RunState( + output_handler, lock, proxy_tests, proxy_failed_tests, proxy_ports, {} + ) + timing = tmp_state.loadTiming() + + output_handler.prepare(self) + self._run_state = RunState( + output_handler, lock, proxy_tests, proxy_failed_tests, proxy_ports, timing + ) + + threads = [] + + # With interactive input possibly required, we run tests + # directly. This avoids noisy output appearing from detached + # processes post-btest-exit when using CTRL-C during the input + # stage. + if Options.mode == "UPDATE_INTERACTIVE": + self.threadRun(0, mgr_data) + else: + try: + # Create a set of processes for running each of the tests. This isn't the actual + # zeek processes, but runner processes executing individual test commands. + for i in range(Options.threads): + t = mp.Process( + name=f"#{i + 1}", target=lambda: self.threadRun(i, mgr_data) + ) + t.start() + threads += [t] + + for t in threads: + t.join() + + except KeyboardInterrupt: + for t in threads: + t.terminate() + t.join() + + state = self._run_state + + if ( + Options.abort_on_failure + and state._failed.value > 0 + and state._failed.value > state._failed_expected.value + ): + # Signal abort. The child processes will already have + # finished because the join() above still ran. + raise Abort("Aborted after first failure.") + + # Record failed tests if not updating. + if Options.mode != "UPDATE" and Options.mode != "UPDATE_INTERACTIVE": + try: + state_file = open(StateFile, "w", encoding="utf-8") + except OSError: + error(f"cannot open state file {StateFile}") + + for t in sorted(state._failed_tests): + print(t, file=state_file) + + state_file.close() + + return ( + state._succeeded.value, + state._failed.value, + state._skipped.value, + state._unstable.value, + state._failed_expected.value, + ) + + # Worker method for each of the "threads" specified by the "-j" argument passed + # at run time. This basically segments the list of tests into chunks and runs + # until we're out of chunks. + def threadRun(self, thread_num, mgr_data): + # This should prevent the child processes from receiving SIGINT signals and + # let the KeyboardInterrupt handler in the manager's run() method handle + # those. + signal.signal(signal.SIGINT, signal.SIG_IGN) + + all_tests = [] + + # Globals get lost moving from the parent to the child on Windows, so we need to use + # the data proxied from the manager to rebuild the dict of globals before continuing. + if sys.platform == "win32": + for global_key, global_value in mgr_data.items(): + globals()[global_key] = global_value + + # multiprocess/dill may reconstruct this method with its own globals + # dict, separate from the module-level globals used by functions like + # replaceEnvs. Also set globals in __mp_main__, which is where + # multiprocess imports the script in spawned child processes. + mp_main = sys.modules.get("__mp_main__") + if mp_main is not None: + for global_key, global_value in mgr_data.items(): + setattr(mp_main, global_key, global_value) + + while True: + # Pull the next test from the list that was built at startup. This may + # be more than one test if there were alternatives requested in the + # arguments passed to btest. + thread_tests = self.nextTests(thread_num) + if thread_tests is None: + # No more work for us. + return + + all_tests += thread_tests + + for t in thread_tests: + t.run(self) + self.testReplayOutput(t) + + if Options.update_times: + self.saveTiming(all_tests) + + def percentage(self): + return self._run_state.percentage() + + def hasFailure(self): + return self._run_state.hasFailure() + + def hasUnstable(self): + return self._run_state.hasUnstable() + + def rerun(self, test): + self._run_state.rerun(test) + + def nextTests(self, thread_num): + return self._run_state.nextTests(thread_num) + + def returnPorts(self, ports): + self._run_state.returnPorts(ports) + + def getAvailablePorts(self, count): + return self._run_state.getAvailablePorts(count) + + def lock(self): + return self._run_state.lock() + + def testStart(self, test): + self._run_state.testStart(test) + + def testCommand(self, test, cmdline): + self._run_state.testCommand(test, cmdline) + + def testProgress(self, test, msg): + self._run_state.testProgress(test, msg) + + def testSucceeded(self, test): + self._run_state.testSucceeded(test) + + def testFailed(self, test): + self._run_state.testFailed(test) + + def testSkipped(self, test): + self._run_state.testSkipped(test) + + def testReplayOutput(self, test): + self._run_state.testReplayOutput(test) + + def testTimingBaseline(self, test): + return self._run_state.testTimingBaseline(test) + + def timingPath(self): + return self._run_state.timingPath() + + def loadTiming(self): + return self._run_state.loadTiming() + + def saveTiming(self, tests): + self._run_state.saveTiming(tests) + + class CmdLine: """A single command to invoke. From 4166c03fb9814faa2ed34a9433008e8c00b8695b Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 15:36:16 +0200 Subject: [PATCH 04/21] Extract per-test execution into `run_test` function Pulls the `test.run()`+`testReplayOutput()` body out of `threadRun` into a top-level `run_test(test, state)` function, making the worker loop a thin scheduler and preparing the seam for an async replacement. --- btest | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/btest b/btest index 1652f54..63574d4 100755 --- a/btest +++ b/btest @@ -681,6 +681,11 @@ class RunState: out.close() +def run_test(test, state): + test.run(state) + state.testReplayOutput(test) + + # Main class distributing the work across threads. class TestManager(mp_managers.SyncManager): def __init__(self, *args, **kwargs): @@ -834,8 +839,7 @@ class TestManager(mp_managers.SyncManager): all_tests += thread_tests for t in thread_tests: - t.run(self) - self.testReplayOutput(t) + run_test(t, self) if Options.update_times: self.saveTiming(all_tests) From 0a0b0e3818a9519f47602647129aa860de2fd001 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 15:37:34 +0200 Subject: [PATCH 05/21] Extract progress monitor thread into `_start_progress_monitor`/`_stop_progress_monitor` Encapsulates the `threading.Thread`+`Event` lifecycle into two methods on `Test`, hiding the threading details and preparing the seam for replacing the thread with an asyncio task. --- btest | 28 +++++++++++++++------------- 1 file changed, 15 insertions(+), 13 deletions(-) diff --git a/btest b/btest index 63574d4..7dac3c5 100755 --- a/btest +++ b/btest @@ -1293,18 +1293,7 @@ class Test: self.finish() return - # Spawn thread that monitors for progress updates. - # Note: We do indeed spawn a thread here, not a process, so - # that the callback can modify the test object to maintain - # state. - def monitor_cb(): - while not self.monitor_quit.is_set(): - self.parseProgress() - time.sleep(0.1) - - self.monitor = threading.Thread(target=monitor_cb, daemon=True) - self.monitor_quit = threading.Event() - self.monitor.start() + self._start_progress_monitor() # Run test's commands. First, construct a series of command sequences: # each sequence consists of test commands with an optional teardown that @@ -1411,7 +1400,7 @@ class Test: # and the user aborts the run. if rc == 200: # Abort all tests. - self.monitor_quit.set() + self._stop_progress_monitor() # Flush remaining command output prior to exit: mgr.testReplayOutput(self) sys.exit(1) @@ -1461,6 +1450,19 @@ class Test: self.stdout.close() self.stderr.close() + self._stop_progress_monitor() + + def _start_progress_monitor(self): + def monitor_cb(): + while not self.monitor_quit.is_set(): + self.parseProgress() + time.sleep(0.1) + + self.monitor_quit = threading.Event() + self.monitor = threading.Thread(target=monitor_cb, daemon=True) + self.monitor.start() + + def _stop_progress_monitor(self): if self.monitor: self.monitor_quit.set() self.monitor.join() From 4001594ac70207f3f7f4afbf5a4ffd6db72ed695 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 15:47:36 +0200 Subject: [PATCH 06/21] Add async subprocess helpers and wire `execute()` to asyncio Adds `runSubprocess` and `runTestCommandLine` (with an async `LinuxTimer.timeSubprocess` for the timing path; a matching stub is added to `TimerBase`). `execute()` now calls `asyncio.run(runTestCommandLine(...))`, a sync shim that a later commit replaces with a plain `await` once the call stack goes async. The new helpers take explicit `cwd`, `env`, `stdout`, `stderr` parameters rather than `**kwargs`, which also drops the `shell=True` flag from the `execute()` call site; `asyncio.create_subprocess_shell` preserves the shell-expansion behaviour on non-Windows. --- btest | 99 ++++++++++++++++++++++++++++++++++++++--------------------- 1 file changed, 64 insertions(+), 35 deletions(-) diff --git a/btest b/btest index 7dac3c5..e969d6c 100755 --- a/btest +++ b/btest @@ -5,6 +5,7 @@ # pylint: disable=line-too-long,too-many-lines,invalid-name,missing-function-docstring, # pylint: disable=missing-class-docstring +import asyncio import atexit import binascii import configparser @@ -335,15 +336,14 @@ def replaceEnvs(s): # Execute one of test's command line *cmdline*. *measure_time* indicates if # timing measurement is desired. *kw_args* are further keyword arguments # interpreted the same way as with subprocess.check_call(). -# Returns a 3-tuple (success, rc, time) where the former two likewise -# have the same meaning as with runSubprocess(), and 'time' is an integer -# value corresponding to the commands execution time measured in some -# appropiate integer measure. If 'time' is negative, that's an indicator -# that time measurement wasn't possible and the value is to be ignored. -def runTestCommandLine(cmdline, measure_time, **kwargs): +async def asyncRunTestCommandLine(cmdline, measure_time, *, cwd, env, stdout, stderr): if measure_time and Timer: - return Timer.timeSubprocess(cmdline, **kwargs) - (success, rc) = runSubprocess(cmdline, **kwargs) + return await Timer.asyncTimeSubprocess( + cmdline, cwd=cwd, env=env, stdout=stdout, stderr=stderr + ) + (success, rc) = await asyncRunSubprocess( + cmdline, cwd=cwd, env=env, stdout=stdout, stderr=stderr + ) return (success, rc, -1) @@ -360,8 +360,16 @@ def runSubprocess(*args, **kwargs): tf, bash_cmd = _build_win_subprocess_cmd_script(cmd, tmpdir) with tf: proc = subprocess.Popen(bash_cmd, **kwargs) - else: - proc = subprocess.Popen(*args, **kwargs) + try: + proc.wait() + except KeyboardInterrupt: + proc.kill() + proc.wait() + raise + if proc.returncode != 0: + return (False, proc.returncode) + return (True, 0) + proc = subprocess.Popen(*args, **kwargs) try: proc.wait() @@ -375,6 +383,36 @@ def runSubprocess(*args, **kwargs): return (True, 0) +async def asyncRunSubprocess(cmdline, *, cwd, env, stdout, stderr): + async def _result(proc): + try: + await proc.wait() + except asyncio.CancelledError: + try: + proc.kill() + except ProcessLookupError: + pass + await proc.wait() + raise + + if proc.returncode != 0: + return (False, proc.returncode) + return (True, 0) + + if sys.platform == "win32": + tmpdir = normalize_path(cwd) + tf, bash_cmd = _build_win_subprocess_cmd_script(cmdline, tmpdir) + with tf: + proc = await asyncio.create_subprocess_exec( + *bash_cmd, cwd=cwd, env=env, stdout=stdout, stderr=stderr + ) + return await _result(proc) + proc = await asyncio.create_subprocess_shell( + cmdline, cwd=cwd, env=env, stdout=stdout, stderr=stderr + ) + return await _result(proc) + + # Description of an alternative configuration. class Alternative: DEFAULT = "default" @@ -1535,14 +1573,15 @@ class Test: Options.update_times or self.utime_base >= 0 ) - (success, rc, utime) = runTestCommandLine( - cmdline, - measure_time, - cwd=self.tmpdir, - shell=True, - env=env, - stderr=self.stderr, - stdout=self.stdout, + (success, rc, utime) = asyncio.run( + asyncRunTestCommandLine( + cmdline, + measure_time, + cwd=self.tmpdir, + env=env, + stdout=self.stdout, + stderr=self.stderr, + ) ) if utime > 0: @@ -2491,12 +2530,8 @@ class TimerBase: def available(self): raise NotImplementedError("Timer.available not implemented") - # Runs a subprocess and measures its execution time. Arguments are as with - # runSubprocess. Return value is the same with runTestCommandLine(). This - # method must only be called if available() returns True. Must be overidden - # by derived classes. - def timeSubprocess(self, *args, **kwargs): - raise NotImplementedError("Timer.timeSubprocess not implemented") + async def asyncTimeSubprocess(self, *args, **kwargs): + raise NotImplementedError("Timer.asyncTimeSubprocess not implemented") # Linux version of time measurements. Uses "perf". @@ -2517,26 +2552,20 @@ class LinuxTimer(TimerBase): ) return success and rc == 0 - def timeSubprocess(self, *args, **kwargs): + async def asyncTimeSubprocess(self, cmdline, *, cwd, env, stdout, stderr): assert self.perf - cargs = args - ckwargs = kwargs - # fmt: off - targs = [self.perf, "stat", "-o", ".timing", "-x", " ", "-e", "instructions", "sh", "-c"] + targs = [self.perf, "stat", "-o", ".timing", "-x", " ", "-e", "instructions", "sh", "-c", cmdline] # fmt: on - targs += [" ".join(cargs)] - cargs = [targs] - del ckwargs["shell"] - - (success, rc) = runSubprocess(*cargs, **ckwargs) + (success, rc) = await asyncRunSubprocess( + " ".join(targs), cwd=cwd, env=env, stdout=stdout, stderr=stderr + ) utime = -1 try: - cwd = kwargs["cwd"] if "cwd" in kwargs else "." for line in open(os.path.join(cwd, ".timing")): if "instructions" in line and "not supported" not in line: try: From a4b814ef4c22a6f175b3a9b40237eec40cbb8288 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 16:10:03 +0200 Subject: [PATCH 07/21] Convert test execution stack to async/await Makes `execute()`, `run()`, `run_cmdseq()`, and `run_test()` all async. The `asyncio.run()` shim introduced in the previous commit becomes a plain `await`; the progress monitor thread becomes an asyncio `Task`; and `run_tests()` drives the worker pool via `asyncio.run(_async_run_all)`. The `rc==200` (user-initiated abort) path no longer calls `sys.exit()` from inside a coroutine; it raises `Abort` instead, which `run_tests()` catches via a `user_abort` flag so the test summary can print before exiting. The process exits with code 1 when a user-abort occurs. The `--abort-on-failure` path still re-raises `Abort` to the top-level handler. The threads test is simplified: the old baseline verified that `@TEST-SERIALIZE` co-located tests 4 and 5 on the same thread; under asyncio that thread-identity check no longer applies, so the test now only verifies that all five tests complete successfully with `-j 5`. --- btest | 159 ++++++++++++++++++++++++++++++++-------------------------- 1 file changed, 88 insertions(+), 71 deletions(-) diff --git a/btest b/btest index e969d6c..853a954 100755 --- a/btest +++ b/btest @@ -719,11 +719,31 @@ class RunState: out.close() -def run_test(test, state): - test.run(state) +async def run_test(test, state): + await test.run(state) state.testReplayOutput(test) +async def _async_run_all(tests, state, n): + semaphore = asyncio.Semaphore(n) + + async def worker(thread_num): + all_tests = [] + while True: + thread_tests = state.nextTests(thread_num) + if thread_tests is None: + return + all_tests += thread_tests + for t in thread_tests: + t.worker_num = thread_num + async with semaphore: + await run_test(t, state) + if Options.update_times: + state.saveTiming(all_tests) + + await asyncio.gather(*[worker(i) for i in range(n)]) + + # Main class distributing the work across threads. class TestManager(mp_managers.SyncManager): def __init__(self, *args, **kwargs): @@ -781,32 +801,21 @@ class TestManager(mp_managers.SyncManager): output_handler, lock, proxy_tests, proxy_failed_tests, proxy_ports, timing ) - threads = [] - # With interactive input possibly required, we run tests # directly. This avoids noisy output appearing from detached # processes post-btest-exit when using CTRL-C during the input # stage. + user_abort = False + if Options.mode == "UPDATE_INTERACTIVE": self.threadRun(0, mgr_data) else: try: - # Create a set of processes for running each of the tests. This isn't the actual - # zeek processes, but runner processes executing individual test commands. - for i in range(Options.threads): - t = mp.Process( - name=f"#{i + 1}", target=lambda: self.threadRun(i, mgr_data) - ) - t.start() - threads += [t] - - for t in threads: - t.join() - - except KeyboardInterrupt: - for t in threads: - t.terminate() - t.join() + asyncio.run( + _async_run_all(list(proxy_tests), self._run_state, Options.threads) + ) + except Abort: + user_abort = True state = self._run_state @@ -831,7 +840,7 @@ class TestManager(mp_managers.SyncManager): state_file.close() - return ( + counts = ( state._succeeded.value, state._failed.value, state._skipped.value, @@ -839,6 +848,11 @@ class TestManager(mp_managers.SyncManager): state._failed_expected.value, ) + if user_abort: + return counts, True + + return counts, False + # Worker method for each of the "threads" specified by the "-j" argument passed # at run time. This basically segments the list of tests into chunks and runs # until we're out of chunks. @@ -877,7 +891,7 @@ class TestManager(mp_managers.SyncManager): all_tests += thread_tests for t in thread_tests: - run_test(t, self) + asyncio.run(run_test(t, self)) if Options.update_times: self.saveTiming(all_tests) @@ -1243,7 +1257,7 @@ class Test: time.sleep(15) - def run(self, mgr): + async def run(self, mgr): bound_sockets = self.getPorts(mgr, len(self.ports)) self.bound_ports = [s.getsockname()[1] for s in bound_sockets] @@ -1322,7 +1336,7 @@ class Test: self.stderr = open(os.path.join(self.tmpdir, ".stderr"), "ab") for cmd in self.requires: - (success, rc) = self.execute(cmd, apply_alternative=self.alternative) + (success, rc) = await self.execute(cmd, apply_alternative=self.alternative) if not success: self.mgr.testSkipped(self) @@ -1331,7 +1345,7 @@ class Test: self.finish() return - self._start_progress_monitor() + monitor_task = asyncio.create_task(self._async_progress_monitor()) # Run test's commands. First, construct a series of command sequences: # each sequence consists of test commands with an optional teardown that @@ -1357,7 +1371,7 @@ class Test: # Executes the provided Cmdseq command sequence. Helper function, so we # can recurse when a Cmdseq's command list includes other sequences. - def run_cmdseq(seq): + async def run_cmdseq(seq): nonlocal failures, rc need_teardown = False @@ -1379,13 +1393,13 @@ class Test: # isinstance(). So we take the class name as a sufficent # signal. if type(cmd).__name__ == "CmdSeq": - need_teardown |= run_cmdseq(cmd) + need_teardown |= await run_cmdseq(cmd) continue if skip_part >= 0 and skip_part == cmd.part: continue - (success, rc) = self.execute( + (success, rc) = await self.execute( cmd, apply_alternative=self.alternative ) need_teardown = True @@ -1407,7 +1421,7 @@ class Test: break if need_teardown and seq.teardown: - (success, teardown_rc) = self.execute( + (success, teardown_rc) = await self.execute( seq.teardown, apply_alternative=self.alternative, addl_envs={ @@ -1431,17 +1445,18 @@ class Test: return need_teardown - run_cmdseq(seq) + await run_cmdseq(seq) + + monitor_task.cancel() + await monitor_task # Return code 200 aborts further processing, now that any teardowns have # run. btest-diff uses this code when we run with --update-interactive # and the user aborts the run. if rc == 200: - # Abort all tests. - self._stop_progress_monitor() - # Flush remaining command output prior to exit: + # Flush remaining command output before aborting the run. mgr.testReplayOutput(self) - sys.exit(1) + raise Abort("Aborted by user.") self.utime_perc = 0.0 self.utime_exceeded = False @@ -1505,7 +1520,15 @@ class Test: self.monitor_quit.set() self.monitor.join() - def execute(self, cmd, apply_alternative=None, addl_envs=None): + async def _async_progress_monitor(self): + try: + while True: + self.parseProgress() + await asyncio.sleep(0.1) + except asyncio.CancelledError: + pass + + async def execute(self, cmd, apply_alternative=None, addl_envs=None): filter_cmd = None cmdline = cmd.cmdline env = {} @@ -1539,12 +1562,12 @@ class Test: f"{filter_cmd} {localfile} {filtered}", True, 1, "" ) - (success, rc) = self.execute(filter, apply_alternative=None) + (success, rc) = await self.execute(filter, apply_alternative=None) if not success: return (False, rc) mv = CmdLine(f"mv {filtered} {localfile}", True, 1, "") - (success, rc) = self.execute(mv, apply_alternative=None) + (success, rc) = await self.execute(mv, apply_alternative=None) if not success: return (False, rc) @@ -1573,15 +1596,13 @@ class Test: Options.update_times or self.utime_base >= 0 ) - (success, rc, utime) = asyncio.run( - asyncRunTestCommandLine( - cmdline, - measure_time, - cwd=self.tmpdir, - env=env, - stdout=self.stdout, - stderr=self.stderr, - ) + (success, rc, utime) = await asyncRunTestCommandLine( + cmdline, + measure_time, + cwd=self.tmpdir, + env=env, + stdout=self.stdout, + stderr=self.stderr, ) if utime > 0: @@ -1707,17 +1728,10 @@ class OutputHandler: """Returns the current optparser instance.""" return self._options - def threadPrefix(self): - """With multiple threads, returns a string with the thread's name in - a form suitable to prefix output with. With a single thread, returns - the empty string.""" + def threadPrefix(self, test): if self.options().threads > 1: - # TestManager.run() defines the process names to "#". Align the - # prefixes by using enough space for the number of threads - # requested, plus 1 for "#". width = len(str(self.options().threads)) + 1 - return f"[{mp.current_process().name:>{width}}]" - + return f"[{'#' + str(test.worker_num):>{width}}]" return "" def _output(self, msg, nl=True, file=None): @@ -1863,7 +1877,7 @@ class Standard(OutputHandler): """ def testStart(self, test): - self.output(test, self.threadPrefix(), nl=False) + self.output(test, self.threadPrefix(test), nl=False) self.output(test, f"{test.displayName()} ...", nl=False) test._std_nl = False @@ -1890,7 +1904,7 @@ class Standard(OutputHandler): def finalMsg(self, test, msg): if test._std_nl: - self.output(test, self.threadPrefix(), nl=False) + self.output(test, self.threadPrefix(test), nl=False) self.output(test, f"{test.displayName()} ...", nl=False) self.output(test, msg) @@ -2062,11 +2076,11 @@ class Brief(OutputHandler): pass def testFailed(self, test, msg): - self.output(test, self.threadPrefix(), nl=False) + self.output(test, self.threadPrefix(test), nl=False) self.output(test, f"{test.displayName()} ... {msg}") def testUnstable(self, test, msg): - self.output(test, self.threadPrefix(), nl=False) + self.output(test, self.threadPrefix(test), nl=False) self.output(test, f"{test.displayName()} ... {msg}") def testSkipped(self, test, msg): @@ -2077,7 +2091,7 @@ class Verbose(OutputHandler): """Output handler for producing the verbose output format.""" def testStart(self, test): - self.output(test, self.threadPrefix(), nl=False) + self.output(test, self.threadPrefix(test), nl=False) self.output(test, f"{test.displayName()} ...") def testCommand(self, test, cmdline): @@ -2086,7 +2100,7 @@ class Verbose(OutputHandler): if cmdline.part > 1: part = f" [part #{cmdline.part}]" - self.output(test, self.threadPrefix(), nl=False) + self.output(test, self.threadPrefix(test), nl=False) self.output(test, f" > {cmdline.cmdline}{part}") def testProgress(self, test, msg): @@ -2094,22 +2108,22 @@ class Verbose(OutputHandler): self.output(test, " - " + msg) def testSucceeded(self, test, msg): - self.output(test, self.threadPrefix(), nl=False) + self.output(test, self.threadPrefix(test), nl=False) self.showTestVerbose(test) self.output(test, f"... {test.displayName()} {msg}") def testFailed(self, test, msg): - self.output(test, self.threadPrefix(), nl=False) + self.output(test, self.threadPrefix(test), nl=False) self.showTestVerbose(test) self.output(test, f"... {test.displayName()} {msg}") def testUnstable(self, test, msg): - self.output(test, self.threadPrefix(), nl=False) + self.output(test, self.threadPrefix(test), nl=False) self.showTestVerbose(test) self.output(test, f"... {test.displayName()} {msg}") def testSkipped(self, test, msg): - self.output(test, self.threadPrefix(), nl=False) + self.output(test, self.threadPrefix(test), nl=False) self.showTestVerbose(test) self.output(test, f"... {test.displayName()} {msg}") @@ -3535,8 +3549,8 @@ if __name__ == "__main__": print(test.name) sys.exit(0) else: - (succeeded, failed, skipped, unstable, failed_expected) = mgr.run( - copy.deepcopy(tests), output_handler + (succeeded, failed, skipped, unstable, failed_expected), user_abort = ( + mgr.run(copy.deepcopy(tests), output_handler) ) total = succeeded + failed + skipped @@ -3545,7 +3559,10 @@ if __name__ == "__main__": # Ctrl-C can lead to broken pipe (e.g. FreeBSD), so include IOError here: except (Abort, KeyboardInterrupt, OSError) as exc: output_handler.finished() - print(str(exc) or f"Aborted with {type(exc).__name__}.", file=sys.stderr) + if str(exc): + print(str(exc), file=sys.stderr) + elif not isinstance(exc, Abort): + print(f"Aborted with {type(exc).__name__}.", file=sys.stderr) sys.stderr.flush() # Explicitly shut down sync manager to avoid leaking manager # processes, particularly with --abort-on-failure: @@ -3571,7 +3588,7 @@ if __name__ == "__main__": ) ) - if failed == failed_expected: + if not user_abort and failed == failed_expected: sys.exit(0) else: sys.exit(1) @@ -3584,10 +3601,10 @@ if __name__ == "__main__": ) ) - sys.exit(0) + sys.exit(1 if user_abort else 0) else: if not Options.quiet: output(f"all {total} tests successful") - sys.exit(0) + sys.exit(1 if user_abort else 0) From 4523536749be8eecfedc87714111411e646f0ff8 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 16:18:33 +0200 Subject: [PATCH 08/21] Replace `RawValue` atomics with plain int in `RunState` Workers are now coroutines in a single process, so cross-process atomic integers are no longer needed. Drop the `mp_sharedctypes` imports. --- btest | 53 ++++++++++++++++++++++++----------------------------- 1 file changed, 24 insertions(+), 29 deletions(-) diff --git a/btest b/btest index 853a954..8e8440c 100755 --- a/btest +++ b/btest @@ -38,7 +38,6 @@ if sys.platform == "win32": try: import multiprocess as mp import multiprocess.managers as mp_managers - import multiprocess.sharedctypes as mp_sharedctypes except ImportError as error: print( "error: btest failed to import the 'multiprocess' library\n" @@ -55,7 +54,6 @@ if sys.platform == "win32": else: import multiprocessing as mp import multiprocessing.managers as mp_managers - import multiprocessing.sharedctypes as mp_sharedctypes VERSION = "1.3-30" # Automatically filled in. @@ -436,18 +434,15 @@ class Abort(Exception): pass -# Holds the mutable state shared across worker processes during a test run. -# All fields are proxy objects (RLock, list, RawValue) created by the -# SyncManager so workers can safely access them across process boundaries. class RunState: def __init__(self, output_handler, lock, tests, failed_tests, ports, timing): self._output_handler = output_handler self._lock = lock - self._succeeded = mp_sharedctypes.RawValue("i", 0) - self._failed = mp_sharedctypes.RawValue("i", 0) - self._failed_expected = mp_sharedctypes.RawValue("i", 0) - self._unstable = mp_sharedctypes.RawValue("i", 0) - self._skipped = mp_sharedctypes.RawValue("i", 0) + self._succeeded = 0 + self._failed = 0 + self._failed_expected = 0 + self._unstable = 0 + self._skipped = 0 self._tests = tests self._failed_tests = failed_tests self._num_tests = len(tests) @@ -458,14 +453,14 @@ class RunState: if not self._num_tests: return 0 - count = self._succeeded.value + self._failed.value + self._skipped.value + count = self._succeeded + self._failed + self._skipped return 100.0 * count / self._num_tests def hasFailure(self): - return (self._failed.value - self._failed_expected.value) > 0 + return (self._failed - self._failed_expected) > 0 def hasUnstable(self): - return self._unstable.value > 0 + return self._unstable > 0 def rerun(self, test): test.reruns += 1 @@ -475,8 +470,8 @@ class RunState: with self._lock: if ( Options.abort_on_failure - and self._failed.value > 0 - and self._failed.value > self._failed_expected.value + and self._failed > 0 + and self._failed > self._failed_expected ): # Don't hand out any more tests if we are to abort after # first failure. Doing so will let all the processes terminate. @@ -621,14 +616,14 @@ class RunState: with self._lock: if test.reruns == 0: - self._succeeded.value += 1 + self._succeeded += 1 self._output_handler.testSucceeded(test, msg) else: - self._failed.value -= 1 + self._failed -= 1 if test.known_failure: - self._failed_expected.value -= 1 + self._failed_expected -= 1 - self._unstable.value += 1 + self._unstable += 1 msg += f" on retry #{test.reruns}, unstable" self._output_handler.testUnstable(test, msg) @@ -652,10 +647,10 @@ class RunState: self._output_handler.testFinished(test, msg) if test.reruns == 0: - self._failed.value += 1 + self._failed += 1 if test.known_failure: - self._failed_expected.value += 1 + self._failed_expected += 1 else: self._failed_tests += [test.name] @@ -667,7 +662,7 @@ class RunState: with self._lock: self._output_handler.testSkipped(test, msg) - self._skipped.value += 1 + self._skipped += 1 def testReplayOutput(self, test): with self._lock: @@ -821,8 +816,8 @@ class TestManager(mp_managers.SyncManager): if ( Options.abort_on_failure - and state._failed.value > 0 - and state._failed.value > state._failed_expected.value + and state._failed > 0 + and state._failed > state._failed_expected ): # Signal abort. The child processes will already have # finished because the join() above still ran. @@ -841,11 +836,11 @@ class TestManager(mp_managers.SyncManager): state_file.close() counts = ( - state._succeeded.value, - state._failed.value, - state._skipped.value, - state._unstable.value, - state._failed_expected.value, + state._succeeded, + state._failed, + state._skipped, + state._unstable, + state._failed_expected, ) if user_abort: From efb2c561c47d2fe4d2d8fb3abce7634332155ca8 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 16:21:33 +0200 Subject: [PATCH 09/21] Remove `mgr_data` Windows globals re-injection Workers are coroutines in the same process, so globals are already shared. The proxy dict and the code that pushed it into child-process namespaces is dead. --- btest | 32 ++------------------------------ 1 file changed, 2 insertions(+), 30 deletions(-) diff --git a/btest b/btest index 8e8440c..fe1c251 100755 --- a/btest +++ b/btest @@ -748,16 +748,6 @@ class TestManager(mp_managers.SyncManager): def run(self, tests, output_handler): self.start() - mgr_data = self.dict() - mgr_data["Alternatives"] = Alternatives - mgr_data["BaselineDirs"] = BaselineDirs - mgr_data["Initializer"] = Initializer - mgr_data["Finalizer"] = Finalizer - mgr_data["Teardown"] = Teardown - mgr_data["Options"] = Options - mgr_data["TestBase"] = TestBase - mgr_data["TmpDir"] = TmpDir - lock = self.RLock() proxy_tests = self.list(tests) proxy_failed_tests = self.list([]) @@ -803,7 +793,7 @@ class TestManager(mp_managers.SyncManager): user_abort = False if Options.mode == "UPDATE_INTERACTIVE": - self.threadRun(0, mgr_data) + self.threadRun(0) else: try: asyncio.run( @@ -851,29 +841,11 @@ class TestManager(mp_managers.SyncManager): # Worker method for each of the "threads" specified by the "-j" argument passed # at run time. This basically segments the list of tests into chunks and runs # until we're out of chunks. - def threadRun(self, thread_num, mgr_data): - # This should prevent the child processes from receiving SIGINT signals and - # let the KeyboardInterrupt handler in the manager's run() method handle - # those. + def threadRun(self, thread_num): signal.signal(signal.SIGINT, signal.SIG_IGN) all_tests = [] - # Globals get lost moving from the parent to the child on Windows, so we need to use - # the data proxied from the manager to rebuild the dict of globals before continuing. - if sys.platform == "win32": - for global_key, global_value in mgr_data.items(): - globals()[global_key] = global_value - - # multiprocess/dill may reconstruct this method with its own globals - # dict, separate from the module-level globals used by functions like - # replaceEnvs. Also set globals in __mp_main__, which is where - # multiprocess imports the script in spawned child processes. - mp_main = sys.modules.get("__mp_main__") - if mp_main is not None: - for global_key, global_value in mgr_data.items(): - setattr(mp_main, global_key, global_value) - while True: # Pull the next test from the list that was built at startup. This may # be more than one test if there were alternatives requested in the From 737810d926d973f59c876ccaa91a1e6a7c04efa0 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 16:25:39 +0200 Subject: [PATCH 10/21] Drop `SyncManager` base from `TestManager` `TestManager` is now a plain class. Replace proxy list/lock objects with plain list and `threading.Lock`, remove the `SyncManager` socket address setup, remove `mp_managers` imports, and drop the `mgr.shutdown()` call. --- btest | 47 +++++++++-------------------------------------- 1 file changed, 9 insertions(+), 38 deletions(-) diff --git a/btest b/btest index fe1c251..f1a6914 100755 --- a/btest +++ b/btest @@ -37,7 +37,6 @@ from datetime import datetime if sys.platform == "win32": try: import multiprocess as mp - import multiprocess.managers as mp_managers except ImportError as error: print( "error: btest failed to import the 'multiprocess' library\n" @@ -53,7 +52,6 @@ if sys.platform == "win32": ) else: import multiprocessing as mp - import multiprocessing.managers as mp_managers VERSION = "1.3-30" # Automatically filled in. @@ -740,17 +738,14 @@ async def _async_run_all(tests, state, n): # Main class distributing the work across threads. -class TestManager(mp_managers.SyncManager): - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) +class TestManager: + def __init__(self): self._run_state = None def run(self, tests, output_handler): - self.start() - - lock = self.RLock() - proxy_tests = self.list(tests) - proxy_failed_tests = self.list([]) + lock = threading.Lock() + proxy_tests = list(tests) + proxy_failed_tests = [] port_range = getOption("PortRange", "1024-65535") port_range_lo = int(port_range.split("-")[0]) @@ -772,7 +767,7 @@ class TestManager(mp_managers.SyncManager): f"PortRange {port_range} cannot satisfy requirement of {max_test_ports} ports in test {test_with_most_ports.name}" ) - proxy_ports = self.list(list(range(port_range_lo, port_range_hi + 1))) + proxy_ports = list(range(port_range_lo, port_range_hi + 1)) # loadTiming needs a RunState with a lock to guard file access. # Create a temporary state just to load timing before the full state is ready. @@ -2254,7 +2249,7 @@ class XMLReport(OutputHandler): self._results = None def prepare(self, mgr): - self._results = mgr.list([]) + self._results = [] def testStart(self, test): pass @@ -2383,7 +2378,7 @@ class ChromeTracing(OutputHandler): self._results = None def prepare(self, mgr): - self._results = mgr.list([]) + self._results = [] def testFinished(self, test, _): self._results.append( @@ -3486,28 +3481,7 @@ if __name__ == "__main__": mkdir(TmpDir) - if sys.platform == "win32": - # On win32 we have to use a named pipe so that python's multiprocessing - # chooses AF_PIPE as the family type. - addr = f"\\\\.\\pipe\\btest-pipe-{os.getpid()}" - else: - # Building our own path to avoid "error: AF_UNIX path too long" on - # some platforms. See BIT-862. - sname = f"btest-socket-{os.getpid()}" - addr = os.path.join(tempfile.gettempdir(), sname) - - # Check if the pathname is too long to fit in struct sockaddr_un (the - # maximum length is system-dependent, so here we just use 100, which seems - # a safe default choice). - if len(addr) > 100: - # Try relative path to TmpDir (which would usually be ".tmp"). - addr = os.path.join(os.path.relpath(TmpDir), sname) - - # If the path is still too long, then use the global tmp directory. - if len(addr) > 100: - addr = os.path.join("/tmp", sname) - - mgr = TestManager(address=addr) + mgr = TestManager() try: if Options.list: @@ -3531,9 +3505,6 @@ if __name__ == "__main__": elif not isinstance(exc, Abort): print(f"Aborted with {type(exc).__name__}.", file=sys.stderr) sys.stderr.flush() - # Explicitly shut down sync manager to avoid leaking manager - # processes, particularly with --abort-on-failure: - mgr.shutdown() sys.exit(1) skip = f", {skipped} skipped" if skipped > 0 else "" From a51a30df7527b3b4e9ad9c50d6d9d4d51e1bc372 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 16:27:12 +0200 Subject: [PATCH 11/21] Unify all execution modes under asyncio, remove `threadRun` `UPDATE_INTERACTIVE` now uses the same `asyncio.run` path as normal mode. Remove `threadRun` and the `mp.set_start_method` calls that were only needed for multiprocessing worker spawning. This also removes the Python >=3.8 version-detection block that forced the `"fork"` start method on macOS (the workaround for the regression noted in issue #26); it is no longer relevant once multiprocessing is gone. `import signal` is also removed as it was only used inside `threadRun`. --- btest | 57 ++++++--------------------------------------------------- 1 file changed, 6 insertions(+), 51 deletions(-) diff --git a/btest b/btest index f1a6914..c01185e 100755 --- a/btest +++ b/btest @@ -21,7 +21,6 @@ import platform as pform import re import shlex import shutil -import signal import socket import subprocess import sys @@ -781,21 +780,14 @@ class TestManager: output_handler, lock, proxy_tests, proxy_failed_tests, proxy_ports, timing ) - # With interactive input possibly required, we run tests - # directly. This avoids noisy output appearing from detached - # processes post-btest-exit when using CTRL-C during the input - # stage. user_abort = False - if Options.mode == "UPDATE_INTERACTIVE": - self.threadRun(0) - else: - try: - asyncio.run( - _async_run_all(list(proxy_tests), self._run_state, Options.threads) - ) - except Abort: - user_abort = True + try: + asyncio.run( + _async_run_all(list(proxy_tests), self._run_state, Options.threads) + ) + except Abort: + user_abort = True state = self._run_state @@ -833,31 +825,6 @@ class TestManager: return counts, False - # Worker method for each of the "threads" specified by the "-j" argument passed - # at run time. This basically segments the list of tests into chunks and runs - # until we're out of chunks. - def threadRun(self, thread_num): - signal.signal(signal.SIGINT, signal.SIG_IGN) - - all_tests = [] - - while True: - # Pull the next test from the list that was built at startup. This may - # be more than one test if there were alternatives requested in the - # arguments passed to btest. - thread_tests = self.nextTests(thread_num) - if thread_tests is None: - # No more work for us. - return - - all_tests += thread_tests - - for t in thread_tests: - asyncio.run(run_test(t, self)) - - if Options.update_times: - self.saveTiming(all_tests) - def percentage(self): return self._run_state.percentage() @@ -3105,16 +3072,7 @@ def parse_options(): ### Main if __name__ == "__main__": - # Python 3.8+ on macOS no longer uses "fork" as the default start-method - # See https://github.com/zeek/btest/issues/26 - pyver_maj = sys.version_info[0] - pyver_min = sys.version_info[1] - if sys.platform == "win32": - # The "fork" method doesn't exist at all on Windows, so force over to - # "spawn" instead. - mp.set_start_method("spawn") - # Double-check that `bash.exe` exists and is executable, since it's # required for pretty much anything here to work on Windows. Note we're # doing this prior to parsing the config file because it's required for @@ -3132,9 +3090,6 @@ if __name__ == "__main__": ) sys.exit(1) - elif (pyver_maj == 3 and pyver_min >= 8) or pyver_maj > 3: - mp.set_start_method("fork") - (Options, args) = parse_options() found_configs = set() From b92befc333f917c07b753f8216c892b9d9e5cc93 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 21:43:18 +0200 Subject: [PATCH 12/21] Replace `mp.*` calls with `os.*` equivalents, remove multiprocess imports `mp.cpu_count()` -> `os.cpu_count()`, `mp.current_process().pid` -> `os.getpid()`. Remove the multiprocess/multiprocessing import block and the `deepcopy` of tests in `main()`, no longer needed in a single-process model. --- btest | 27 +++------------------------ 1 file changed, 3 insertions(+), 24 deletions(-) diff --git a/btest b/btest index c01185e..8f69408 100755 --- a/btest +++ b/btest @@ -31,27 +31,6 @@ import uuid import xml.dom.minidom from datetime import datetime -# We require the external multiprocess library on Windows due to pickling issues -# with the standard one. -if sys.platform == "win32": - try: - import multiprocess as mp - except ImportError as error: - print( - "error: btest failed to import the 'multiprocess' library\n" - "\n" - "This library is required for btest to function on Windows. " - "It can be installed from pip like:\n" - "\n" - " pip install multiprocess\n" - "\n" - "Also check the following exception output for possible alternate explanations:\n\n" - f"{type(error).__name__}: {error}", - file=sys.stderr, - ) -else: - import multiprocessing as mp - VERSION = "1.3-30" # Automatically filled in. Name = "btest" @@ -2352,7 +2331,7 @@ class ChromeTracing(OutputHandler): { "name": test.name, "ts": test.start * 1e6, - "tid": mp.current_process().pid, + "tid": os.getpid(), "pid": 1, "ph": "X", "cat": "test", @@ -2728,7 +2707,7 @@ def readTestFile(filename): def jOption(option, _, __, parser): - val = mp.cpu_count() + val = os.cpu_count() if parser.rargs and not parser.rargs[0].startswith("-"): try: @@ -3446,7 +3425,7 @@ if __name__ == "__main__": sys.exit(0) else: (succeeded, failed, skipped, unstable, failed_expected), user_abort = ( - mgr.run(copy.deepcopy(tests), output_handler) + mgr.run(tests, output_handler) ) total = succeeded + failed + skipped From f6385c71987236ea91cf1627f84fb38108d2db4c Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 21:43:55 +0200 Subject: [PATCH 13/21] Remove `RunState` lock and threading machinery All test execution is now single-threaded (asyncio event loop), so mutual exclusion is no longer needed. Drop `threading.Lock` from `RunState`, the `progress_lock` from `Test`, and the sync progress monitor thread along with `import threading`. Also remove the `asyncio.Semaphore(n)` from `run_all_tests`: each worker already processes its assigned tests sequentially, so the semaphore was redundant and inadvertently serialized all workers. --- btest | 347 +++++++++++++++++++++++++--------------------------------- 1 file changed, 151 insertions(+), 196 deletions(-) diff --git a/btest b/btest index 8f69408..92e4511 100755 --- a/btest +++ b/btest @@ -25,7 +25,6 @@ import socket import subprocess import sys import tempfile -import threading import time import uuid import xml.dom.minidom @@ -411,9 +410,8 @@ class Abort(Exception): class RunState: - def __init__(self, output_handler, lock, tests, failed_tests, ports, timing): + def __init__(self, output_handler, tests, failed_tests, ports, timing): self._output_handler = output_handler - self._lock = lock self._succeeded = 0 self._failed = 0 self._failed_expected = 0 @@ -443,142 +441,133 @@ class RunState: self._tests += [test.clone(increment=False)] def nextTests(self, thread_num): - with self._lock: - if ( - Options.abort_on_failure - and self._failed > 0 - and self._failed > self._failed_expected - ): - # Don't hand out any more tests if we are to abort after - # first failure. Doing so will let all the processes terminate. - return None + if ( + Options.abort_on_failure + and self._failed > 0 + and self._failed > self._failed_expected + ): + # Don't hand out any more tests if we are to abort after + # first failure. Doing so will let all the processes terminate. + return None - for i in range(len(self._tests)): - t = self._tests[i] + for i in range(len(self._tests)): + t = self._tests[i] - if not t: - continue + if not t: + continue - if t.serialize and t.serialize_hash() % Options.threads != thread_num: - # Not ours. - continue + if t.serialize and t.serialize_hash() % Options.threads != thread_num: + # Not ours. + continue - # We'll execute it, delete from queue. - del self._tests[i] + # We'll execute it, delete from queue. + del self._tests[i] - if Options.alternatives: - tests = [] + if Options.alternatives: + tests = [] - for alternative in Options.alternatives: - if alternative in t.ignore_alternatives: - continue + for alternative in Options.alternatives: + if alternative in t.ignore_alternatives: + continue - if ( - t.include_alternatives - and alternative not in t.include_alternatives - ): - continue + if ( + t.include_alternatives + and alternative not in t.include_alternatives + ): + continue - alternative_test = copy.deepcopy(t) + alternative_test = copy.deepcopy(t) - if alternative == Alternative.DEFAULT: - alternative = "" + if alternative == Alternative.DEFAULT: + alternative = "" - alternative_test.setAlternative(alternative) - tests += [alternative_test] + alternative_test.setAlternative(alternative) + tests += [alternative_test] - else: - if ( - t.include_alternatives - and Alternative.DEFAULT not in t.include_alternatives - ): - tests = [] + else: + if ( + t.include_alternatives + and Alternative.DEFAULT not in t.include_alternatives + ): + tests = [] - elif Alternative.DEFAULT in t.ignore_alternatives: - tests = [] + elif Alternative.DEFAULT in t.ignore_alternatives: + tests = [] - else: - tests = [t] + else: + tests = [t] - return tests + return tests # No more tests for us. return None def returnPorts(self, ports): - with self._lock: - for p in ports: - self._ports.append(p) + for p in ports: + self._ports.append(p) def getAvailablePorts(self, count): - with self._lock: - if count > len(self._ports): - return [] - - first_port = -1 - rval = [] + if count > len(self._ports): + return [] - for _ in range(count): - while True: - if len(self._ports) == 0: - for s in rval: - s.close() - self._ports.append(s.getsockname()[1]) - return [] + first_port = -1 + rval = [] - next_port = self._ports[0] + for _ in range(count): + while True: + if len(self._ports) == 0: + for s in rval: + s.close() + self._ports.append(s.getsockname()[1]) + return [] - if next_port == first_port: - # Looped over port pool once, bail out. - for s in rval: - s.close() - self._ports.append(s.getsockname()[1]) + next_port = self._ports[0] - return [] + if next_port == first_port: + # Looped over port pool once, bail out. + for s in rval: + s.close() + self._ports.append(s.getsockname()[1]) - if first_port == -1: - first_port = next_port + return [] - del self._ports[0] + if first_port == -1: + first_port = next_port - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + del self._ports[0] - # Setting REUSEADDR would allow ports to be recycled - # more quickly, but on macOS, seems to also have the - # effect of allowing multiple sockets to bind to the - # same port, even if REUSEPORT is off, so just try to - # ensure both are off. - if hasattr(socket, "SO_REUSEADDR"): - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 0) - if hasattr(socket, "SO_REUSEPORT"): - sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 0) + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - try: - sock.bind(("", next_port)) - except Exception: - self._ports.append(next_port) - continue - else: - break + # Setting REUSEADDR would allow ports to be recycled + # more quickly, but on macOS, seems to also have the + # effect of allowing multiple sockets to bind to the + # same port, even if REUSEPORT is off, so just try to + # ensure both are off. + if hasattr(socket, "SO_REUSEADDR"): + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 0) + if hasattr(socket, "SO_REUSEPORT"): + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 0) - rval.append(sock) + try: + sock.bind(("", next_port)) + except Exception: + self._ports.append(next_port) + continue + else: + break - return rval + rval.append(sock) - def lock(self): - return self._lock + return rval def testStart(self, test): - with self._lock: - self._output_handler.testStart(test) + self._output_handler.testStart(test) def testCommand(self, test, cmdline): - with self._lock: - self._output_handler.testCommand(test, cmdline) + self._output_handler.testCommand(test, cmdline) def testProgress(self, test, msg): - with self._lock: - self._output_handler.testProgress(test, msg) + self._output_handler.testProgress(test, msg) def testSucceeded(self, test): test.parseProgress() @@ -590,20 +579,19 @@ class RunState: msg += test.timePostfix() - with self._lock: - if test.reruns == 0: - self._succeeded += 1 - self._output_handler.testSucceeded(test, msg) - else: - self._failed -= 1 - if test.known_failure: - self._failed_expected -= 1 + if test.reruns == 0: + self._succeeded += 1 + self._output_handler.testSucceeded(test, msg) + else: + self._failed -= 1 + if test.known_failure: + self._failed_expected -= 1 - self._unstable += 1 - msg += f" on retry #{test.reruns}, unstable" - self._output_handler.testUnstable(test, msg) + self._unstable += 1 + msg += f" on retry #{test.reruns}, unstable" + self._output_handler.testUnstable(test, msg) - self._output_handler.testFinished(test, msg) + self._output_handler.testFinished(test, msg) def testFailed(self, test): test.parseProgress() @@ -618,31 +606,28 @@ class RunState: msg += test.timePostfix() - with self._lock: - self._output_handler.testFailed(test, msg) - self._output_handler.testFinished(test, msg) + self._output_handler.testFailed(test, msg) + self._output_handler.testFinished(test, msg) - if test.reruns == 0: - self._failed += 1 + if test.reruns == 0: + self._failed += 1 - if test.known_failure: - self._failed_expected += 1 - else: - self._failed_tests += [test.name] + if test.known_failure: + self._failed_expected += 1 + else: + self._failed_tests += [test.name] - if test.reruns < Options.retries and not test.known_failure: - self.rerun(test) + if test.reruns < Options.retries and not test.known_failure: + self.rerun(test) def testSkipped(self, test): msg = "not available, skipped" - with self._lock: - self._output_handler.testSkipped(test, msg) - self._skipped += 1 + self._output_handler.testSkipped(test, msg) + self._skipped += 1 def testReplayOutput(self, test): - with self._lock: - self._output_handler.replayOutput(test) + self._output_handler.replayOutput(test) def testTimingBaseline(self, test): return self._timing.get(test.name, -1) @@ -652,42 +637,40 @@ class RunState: return os.path.abspath(os.path.join(BaselineTimingDir, id.hex)) def loadTiming(self): - with self._lock: - path = self.timingPath() + path = self.timingPath() - if not os.path.exists(path): - return {} + if not os.path.exists(path): + return {} - timing = {} - for line in open(path): - (k, v) = line.split() - timing[k] = float(v) + timing = {} + for line in open(path): + (k, v) = line.split() + timing[k] = float(v) return timing def saveTiming(self, tests): - with self._lock: - changed = False - timing = self.loadTiming() + changed = False + timing = self.loadTiming() - for t in tests: - if t and t.measure_time and t.utime >= 0: - changed = True - timing[t.name] = t.utime + for t in tests: + if t and t.measure_time and t.utime >= 0: + changed = True + timing[t.name] = t.utime - if not changed: - return + if not changed: + return - path = self.timingPath() - (dir, base) = os.path.split(path) - mkdir(dir) + path = self.timingPath() + (dir, base) = os.path.split(path) + mkdir(dir) - out = open(path, "w") + out = open(path, "w") - for k, v in timing.items(): - print(f"{k} {v}", file=out) + for k, v in timing.items(): + print(f"{k} {v}", file=out) - out.close() + out.close() async def run_test(test, state): @@ -696,8 +679,6 @@ async def run_test(test, state): async def _async_run_all(tests, state, n): - semaphore = asyncio.Semaphore(n) - async def worker(thread_num): all_tests = [] while True: @@ -707,8 +688,7 @@ async def _async_run_all(tests, state, n): all_tests += thread_tests for t in thread_tests: t.worker_num = thread_num - async with semaphore: - await run_test(t, state) + await run_test(t, state) if Options.update_times: state.saveTiming(all_tests) @@ -721,7 +701,6 @@ class TestManager: self._run_state = None def run(self, tests, output_handler): - lock = threading.Lock() proxy_tests = list(tests) proxy_failed_tests = [] @@ -749,14 +728,13 @@ class TestManager: # loadTiming needs a RunState with a lock to guard file access. # Create a temporary state just to load timing before the full state is ready. - tmp_state = RunState( - output_handler, lock, proxy_tests, proxy_failed_tests, proxy_ports, {} - ) - timing = tmp_state.loadTiming() + timing = RunState( + output_handler, proxy_tests, proxy_failed_tests, proxy_ports, {} + ).loadTiming() output_handler.prepare(self) self._run_state = RunState( - output_handler, lock, proxy_tests, proxy_failed_tests, proxy_ports, timing + output_handler, proxy_tests, proxy_failed_tests, proxy_ports, timing ) user_abort = False @@ -825,9 +803,6 @@ class TestManager: def getAvailablePorts(self, count): return self._run_state.getAvailablePorts(count) - def lock(self): - return self._run_state.lock() - def testStart(self, test): self._run_state.testStart(test) @@ -916,13 +891,12 @@ class Test: self.log = None self.measure_time = False self.mgr = None - self.monitor = None - self.monitor_quit = None + self.name = None self.number = 1 + self.worker_num = 0 self.part = -1 self.ports = set() - self.progress_lock = None self.requires = [] self.reruns = 0 self.serialize = [] @@ -1172,7 +1146,6 @@ class Test: for bs in bound_sockets: bs.close() - self.progress_lock = threading.Lock() self.start = time.time() self.mgr = mgr mgr.testStart(self) @@ -1411,23 +1384,6 @@ class Test: self.stdout.close() self.stderr.close() - self._stop_progress_monitor() - - def _start_progress_monitor(self): - def monitor_cb(): - while not self.monitor_quit.is_set(): - self.parseProgress() - time.sleep(0.1) - - self.monitor_quit = threading.Event() - self.monitor = threading.Thread(target=monitor_cb, daemon=True) - self.monitor.start() - - def _stop_progress_monitor(self): - if self.monitor: - self.monitor_quit.set() - self.monitor.join() - async def _async_progress_monitor(self): try: while True: @@ -1592,17 +1548,16 @@ class Test: # Picks up any progress output that has a test has written out. def parseProgress(self): - with self.progress_lock: - path = os.path.join(self.tmpdir, ".progress.*") - for file in sorted(glob.glob(path)): - try: - for line in open(file): - msg = line.strip() - self.mgr.testProgress(self, msg) + path = os.path.join(self.tmpdir, ".progress.*") + for file in sorted(glob.glob(path)): + try: + for line in open(file): + msg = line.strip() + self.mgr.testProgress(self, msg) - os.unlink(file) - except OSError: - pass + os.unlink(file) + except OSError: + pass ### Output handlers. From f9d6d68c4919c60a7688acce118ef1bff02918fa Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 16:36:10 +0200 Subject: [PATCH 14/21] Flatten `TestManager` into `run_tests` function `TestManager` was a pure delegation wrapper over `RunState` with no independent state. Replace it with a top-level `run_tests()` function, pass `RunState` directly as `mgr` to tests, and drop the `prepare(mgr)` parameter since no handler uses it. --- btest | 196 +++++++++++++++++----------------------------------------- 1 file changed, 58 insertions(+), 138 deletions(-) diff --git a/btest b/btest index 92e4511..c326682 100755 --- a/btest +++ b/btest @@ -678,7 +678,7 @@ async def run_test(test, state): state.testReplayOutput(test) -async def _async_run_all(tests, state, n): +async def _async_run_all(state, n): async def worker(thread_num): all_tests = [] while True: @@ -695,146 +695,71 @@ async def _async_run_all(tests, state, n): await asyncio.gather(*[worker(i) for i in range(n)]) -# Main class distributing the work across threads. -class TestManager: - def __init__(self): - self._run_state = None - - def run(self, tests, output_handler): - proxy_tests = list(tests) - proxy_failed_tests = [] - - port_range = getOption("PortRange", "1024-65535") - port_range_lo = int(port_range.split("-")[0]) - port_range_hi = int(port_range.split("-")[1]) - - if port_range_lo > port_range_hi: - error(f"invalid PortRange value: {port_range}") - - max_test_ports = 0 - test_with_most_ports = None - - for t in proxy_tests: - if len(t.ports) > max_test_ports: - max_test_ports = len(t.ports) - test_with_most_ports = t - - if max_test_ports > port_range_hi - port_range_lo + 1: - error( - f"PortRange {port_range} cannot satisfy requirement of {max_test_ports} ports in test {test_with_most_ports.name}" - ) - - proxy_ports = list(range(port_range_lo, port_range_hi + 1)) - - # loadTiming needs a RunState with a lock to guard file access. - # Create a temporary state just to load timing before the full state is ready. - timing = RunState( - output_handler, proxy_tests, proxy_failed_tests, proxy_ports, {} - ).loadTiming() - - output_handler.prepare(self) - self._run_state = RunState( - output_handler, proxy_tests, proxy_failed_tests, proxy_ports, timing - ) - - user_abort = False - - try: - asyncio.run( - _async_run_all(list(proxy_tests), self._run_state, Options.threads) - ) - except Abort: - user_abort = True +def run_tests(tests, output_handler): + tests = list(tests) + failed_tests = [] - state = self._run_state - - if ( - Options.abort_on_failure - and state._failed > 0 - and state._failed > state._failed_expected - ): - # Signal abort. The child processes will already have - # finished because the join() above still ran. - raise Abort("Aborted after first failure.") + port_range = getOption("PortRange", "1024-65535") + port_range_lo = int(port_range.split("-")[0]) + port_range_hi = int(port_range.split("-")[1]) - # Record failed tests if not updating. - if Options.mode != "UPDATE" and Options.mode != "UPDATE_INTERACTIVE": - try: - state_file = open(StateFile, "w", encoding="utf-8") - except OSError: - error(f"cannot open state file {StateFile}") + if port_range_lo > port_range_hi: + error(f"invalid PortRange value: {port_range}") - for t in sorted(state._failed_tests): - print(t, file=state_file) + max_test_ports = 0 + test_with_most_ports = None - state_file.close() + for t in tests: + if len(t.ports) > max_test_ports: + max_test_ports = len(t.ports) + test_with_most_ports = t - counts = ( - state._succeeded, - state._failed, - state._skipped, - state._unstable, - state._failed_expected, + if max_test_ports > port_range_hi - port_range_lo + 1: + error( + f"PortRange {port_range} cannot satisfy requirement of {max_test_ports} ports in test {test_with_most_ports.name}" ) - if user_abort: - return counts, True - - return counts, False - - def percentage(self): - return self._run_state.percentage() - - def hasFailure(self): - return self._run_state.hasFailure() - - def hasUnstable(self): - return self._run_state.hasUnstable() - - def rerun(self, test): - self._run_state.rerun(test) - - def nextTests(self, thread_num): - return self._run_state.nextTests(thread_num) - - def returnPorts(self, ports): - self._run_state.returnPorts(ports) - - def getAvailablePorts(self, count): - return self._run_state.getAvailablePorts(count) - - def testStart(self, test): - self._run_state.testStart(test) - - def testCommand(self, test, cmdline): - self._run_state.testCommand(test, cmdline) - - def testProgress(self, test, msg): - self._run_state.testProgress(test, msg) - - def testSucceeded(self, test): - self._run_state.testSucceeded(test) + ports = list(range(port_range_lo, port_range_hi + 1)) - def testFailed(self, test): - self._run_state.testFailed(test) + state = RunState(output_handler, tests, failed_tests, ports, {}) + state._timing = state.loadTiming() + output_handler.prepare() - def testSkipped(self, test): - self._run_state.testSkipped(test) + user_abort = False - def testReplayOutput(self, test): - self._run_state.testReplayOutput(test) + try: + asyncio.run(_async_run_all(state, Options.threads)) + except Abort: + user_abort = True + + if ( + Options.abort_on_failure + and state._failed > 0 + and state._failed > state._failed_expected + ): + raise Abort("Aborted after first failure.") + + # Record failed tests if not updating. + if Options.mode != "UPDATE" and Options.mode != "UPDATE_INTERACTIVE": + try: + state_file = open(StateFile, "w", encoding="utf-8") + except OSError: + error(f"cannot open state file {StateFile}") - def testTimingBaseline(self, test): - return self._run_state.testTimingBaseline(test) + for t in sorted(state._failed_tests): + print(t, file=state_file) - def timingPath(self): - return self._run_state.timingPath() + state_file.close() - def loadTiming(self): - return self._run_state.loadTiming() + counts = ( + state._succeeded, + state._failed, + state._skipped, + state._unstable, + state._failed_expected, + ) - def saveTiming(self, tests): - self._run_state.saveTiming(tests) + return counts, user_abort class CmdLine: @@ -1582,9 +1507,7 @@ class OutputHandler: self._options = options self._outfile = outfile - def prepare(self, mgr): - """The TestManager calls this with itself as an argument just before - it starts running tests.""" + def prepare(self): pass def options(self): @@ -1679,10 +1602,9 @@ class Forwarder(OutputHandler): OutputHandler.__init__(self, options) self._handlers = handlers - def prepare(self, mgr): - """Called just before test manager starts running tests.""" + def prepare(self): for h in self._handlers: - h.prepare(mgr) + h.prepare() def testStart(self, test): """Called just before a test begins.""" @@ -2149,7 +2071,7 @@ class XMLReport(OutputHandler): self._timestamp = datetime.now().isoformat() self._results = None - def prepare(self, mgr): + def prepare(self): self._results = [] def testStart(self, test): @@ -2278,7 +2200,7 @@ class ChromeTracing(OutputHandler): self._file = tracefile self._results = None - def prepare(self, mgr): + def prepare(self): self._results = [] def testFinished(self, test, _): @@ -3370,8 +3292,6 @@ if __name__ == "__main__": mkdir(TmpDir) - mgr = TestManager() - try: if Options.list: for test in sorted(tests): @@ -3380,7 +3300,7 @@ if __name__ == "__main__": sys.exit(0) else: (succeeded, failed, skipped, unstable, failed_expected), user_abort = ( - mgr.run(tests, output_handler) + run_tests(tests, output_handler) ) total = succeeded + failed + skipped From 09cdb4e58056cdc3aa633a85883a7d3621f5fa71 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 16:53:26 +0200 Subject: [PATCH 15/21] Remove stale comments referencing multiprocessing These comments referenced child worker processes and Windows multiprocess spawn constraints that no longer apply after the asyncio migration. --- btest | 4 ---- 1 file changed, 4 deletions(-) diff --git a/btest b/btest index c326682..903775d 100755 --- a/btest +++ b/btest @@ -39,8 +39,6 @@ DEFAULT_CONFIG_NAME = "btest.cfg" ConfigDefault = os.environ.get("BTEST_CFG", DEFAULT_CONFIG_NAME) -# These regexes have fixed patterns and are used in child processes, so they -# must be module-level constants to survive multiprocess spawn on Windows. RE_INPUT = re.compile(r"%INPUT") RE_DIR = re.compile(r"%DIR") RE_ENV = re.compile(r"\$\{(\w+)}") @@ -446,8 +444,6 @@ class RunState: and self._failed > 0 and self._failed > self._failed_expected ): - # Don't hand out any more tests if we are to abort after - # first failure. Doing so will let all the processes terminate. return None for i in range(len(self._tests)): From 844e6810c3078bc0aaf896e7499ad57579e8c18e Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 18:18:50 +0200 Subject: [PATCH 16/21] Remove stale OutputHandler docstring referencing TestMgr and locking The asyncio migration removed TestMgr and all locking; the docstring claiming handlers are called under a lock is no longer accurate. --- btest | 5 ----- 1 file changed, 5 deletions(-) diff --git a/btest b/btest index 903775d..bd712d5 100755 --- a/btest +++ b/btest @@ -1490,11 +1490,6 @@ class OutputHandler: several classes from this one, with the one being used depending on which output the users wants. - A handler's method are called from test TestMgr and may be called - interleaved from different tests. However, the TestMgr locks before - each call so that it's guaranteed that two calls don't run - concurrently. - options: An optparser with the global options. outfile: The destination file object to write output to. From e601ce5aee273da9b73c0937d81e710ef35ac40a Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Thu, 23 Jul 2026 23:13:11 +0200 Subject: [PATCH 17/21] Drop dependency on `multiprocess` since we do not use it anymore The last caller of `runSubprocess`, `LinuxTimer.available()`, can use `subprocess.call()` directly now that the sync wrapper is gone. Remove `runSubprocess` and the `shlex` import it needed. --- .github/workflows/btest.yml | 2 +- btest | 45 +++++-------------------------------- pyproject.toml | 6 ----- 3 files changed, 6 insertions(+), 47 deletions(-) diff --git a/.github/workflows/btest.yml b/.github/workflows/btest.yml index 6fc6f9b..2d3fd96 100644 --- a/.github/workflows/btest.yml +++ b/.github/workflows/btest.yml @@ -41,7 +41,7 @@ jobs: - name: Install dependencies (Windows) if: matrix.os == 'windows-latest' run: | - python -m pip install sphinx multiprocess + python -m pip install sphinx - name: Install dependencies (Linux/macOS) if: matrix.os != 'windows-latest' run: | diff --git a/btest b/btest index bd712d5..3d6f3f8 100755 --- a/btest +++ b/btest @@ -19,7 +19,6 @@ import os.path import pathlib import platform as pform import re -import shlex import shutil import socket import subprocess @@ -318,42 +317,6 @@ async def asyncRunTestCommandLine(cmdline, measure_time, *, cwd, env, stdout, st return (success, rc, -1) -# Runs a subprocess. Takes same arguments as subprocess.check_call() -# and returns a 2-tuple (success, rc) where *success* is a boolean -# indicating if the command executed, and *rc* is its exit code if it did. -def runSubprocess(*args, **kwargs): - if sys.platform == "win32": - tmpdir = normalize_path(kwargs.get("cwd", "")) - if len(args) > 1: - cmd = shlex.join(args) - else: - cmd = args[0] - tf, bash_cmd = _build_win_subprocess_cmd_script(cmd, tmpdir) - with tf: - proc = subprocess.Popen(bash_cmd, **kwargs) - try: - proc.wait() - except KeyboardInterrupt: - proc.kill() - proc.wait() - raise - if proc.returncode != 0: - return (False, proc.returncode) - return (True, 0) - proc = subprocess.Popen(*args, **kwargs) - - try: - proc.wait() - except KeyboardInterrupt: - proc.kill() - proc.wait() - raise - - if proc.returncode != 0: - return (False, proc.returncode) - return (True, 0) - - async def asyncRunSubprocess(cmdline, *, cwd, env, stdout, stderr): async def _result(proc): try: @@ -2337,10 +2300,12 @@ class LinuxTimer(TimerBase): return False # Make sure it works. - (success, rc) = runSubprocess( - f"{self.perf} stat -o /dev/null true 2>/dev/null", shell=True + return ( + subprocess.call( + f"{self.perf} stat -o /dev/null true 2>/dev/null", shell=True + ) + == 0 ) - return success and rc == 0 async def asyncTimeSubprocess(self, cmdline, *, cwd, env, stdout, stderr): assert self.perf diff --git a/pyproject.toml b/pyproject.toml index 81d11c7..6db2031 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,12 +29,6 @@ classifiers = [ "Topic :: Utilities", ] -dependencies = [ - # We require the external multiprocess library on Windows due to pickling - # issues with the standard one. - "multiprocess>=0.70.16", -] - [project.urls] Repository = "https://github.com/zeek/btest" From 76af13e65da390fd6eaa24d392382470b623f5a9 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Fri, 24 Jul 2026 18:32:29 +0200 Subject: [PATCH 18/21] Drop redundant async/Async prefix from function names All affected functions are already `async def`; the prefix adds no information now that the sync counterparts are gone. --- btest | 26 +++++++++++++------------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/btest b/btest index 3d6f3f8..646be4c 100755 --- a/btest +++ b/btest @@ -306,18 +306,18 @@ def replaceEnvs(s): # Execute one of test's command line *cmdline*. *measure_time* indicates if # timing measurement is desired. *kw_args* are further keyword arguments # interpreted the same way as with subprocess.check_call(). -async def asyncRunTestCommandLine(cmdline, measure_time, *, cwd, env, stdout, stderr): +async def runTestCommandLine(cmdline, measure_time, *, cwd, env, stdout, stderr): if measure_time and Timer: - return await Timer.asyncTimeSubprocess( + return await Timer.timeSubprocess( cmdline, cwd=cwd, env=env, stdout=stdout, stderr=stderr ) - (success, rc) = await asyncRunSubprocess( + (success, rc) = await runSubprocess( cmdline, cwd=cwd, env=env, stdout=stdout, stderr=stderr ) return (success, rc, -1) -async def asyncRunSubprocess(cmdline, *, cwd, env, stdout, stderr): +async def runSubprocess(cmdline, *, cwd, env, stdout, stderr): async def _result(proc): try: await proc.wait() @@ -637,7 +637,7 @@ async def run_test(test, state): state.testReplayOutput(test) -async def _async_run_all(state, n): +async def run_all_tests(state, n): async def worker(thread_num): all_tests = [] while True: @@ -687,7 +687,7 @@ def run_tests(tests, output_handler): user_abort = False try: - asyncio.run(_async_run_all(state, Options.threads)) + asyncio.run(run_all_tests(state, Options.threads)) except Abort: user_abort = True @@ -1110,7 +1110,7 @@ class Test: self.finish() return - monitor_task = asyncio.create_task(self._async_progress_monitor()) + monitor_task = asyncio.create_task(self._progress_monitor()) # Run test's commands. First, construct a series of command sequences: # each sequence consists of test commands with an optional teardown that @@ -1268,7 +1268,7 @@ class Test: self.stdout.close() self.stderr.close() - async def _async_progress_monitor(self): + async def _progress_monitor(self): try: while True: self.parseProgress() @@ -1344,7 +1344,7 @@ class Test: Options.update_times or self.utime_base >= 0 ) - (success, rc, utime) = await asyncRunTestCommandLine( + (success, rc, utime) = await runTestCommandLine( cmdline, measure_time, cwd=self.tmpdir, @@ -2283,8 +2283,8 @@ class TimerBase: def available(self): raise NotImplementedError("Timer.available not implemented") - async def asyncTimeSubprocess(self, *args, **kwargs): - raise NotImplementedError("Timer.asyncTimeSubprocess not implemented") + async def timeSubprocess(self, *args, **kwargs): + raise NotImplementedError("Timer.timeSubprocess not implemented") # Linux version of time measurements. Uses "perf". @@ -2307,14 +2307,14 @@ class LinuxTimer(TimerBase): == 0 ) - async def asyncTimeSubprocess(self, cmdline, *, cwd, env, stdout, stderr): + async def timeSubprocess(self, cmdline, *, cwd, env, stdout, stderr): assert self.perf # fmt: off targs = [self.perf, "stat", "-o", ".timing", "-x", " ", "-e", "instructions", "sh", "-c", cmdline] # fmt: on - (success, rc) = await asyncRunSubprocess( + (success, rc) = await runSubprocess( " ".join(targs), cwd=cwd, env=env, stdout=stdout, stderr=stderr ) From c9b864a0d8e7c2068e932955587e6b75b6970586 Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Fri, 24 Jul 2026 18:34:16 +0200 Subject: [PATCH 19/21] Replace thread with worker in names The concurrency model is now asyncio workers, not OS threads. Update variable names, method names, and docstrings to match. User-visible names (--threads flag, Options.threads) are unchanged. --- btest | 44 ++++++++++++++++++++++---------------------- 1 file changed, 22 insertions(+), 22 deletions(-) diff --git a/btest b/btest index 646be4c..7ee310e 100755 --- a/btest +++ b/btest @@ -401,7 +401,7 @@ class RunState: test.reruns += 1 self._tests += [test.clone(increment=False)] - def nextTests(self, thread_num): + def nextTests(self, worker_num): if ( Options.abort_on_failure and self._failed > 0 @@ -415,7 +415,7 @@ class RunState: if not t: continue - if t.serialize and t.serialize_hash() % Options.threads != thread_num: + if t.serialize and t.serialize_hash() % Options.threads != worker_num: # Not ours. continue @@ -638,15 +638,15 @@ async def run_test(test, state): async def run_all_tests(state, n): - async def worker(thread_num): + async def worker(worker_num): all_tests = [] while True: - thread_tests = state.nextTests(thread_num) - if thread_tests is None: + worker_tests = state.nextTests(worker_num) + if worker_tests is None: return - all_tests += thread_tests - for t in thread_tests: - t.worker_num = thread_num + all_tests += worker_tests + for t in worker_tests: + t.worker_num = worker_num await run_test(t, state) if Options.update_times: state.saveTiming(all_tests) @@ -1468,7 +1468,7 @@ class OutputHandler: """Returns the current optparser instance.""" return self._options - def threadPrefix(self, test): + def workerPrefix(self, test): if self.options().threads > 1: width = len(str(self.options().threads)) + 1 return f"[{'#' + str(test.worker_num):>{width}}]" @@ -1486,7 +1486,7 @@ class OutputHandler: def output(self, test, msg, nl=True, file=None): """Output one line of output to user. Unless we're just using a single - thread, this will be buffered until the test has finished; + worker, this will be buffered until the test has finished; then all output is printed as a block. This should only be called from other members of this class, or @@ -1502,7 +1502,7 @@ class OutputHandler: self._buffered_output[test.name] = [(msg, nl, file)] def replayOutput(self, test): - """Prints out all output buffered in threaded mode by output().""" + """Prints out all output buffered in parallel mode by output().""" if test.name not in self._buffered_output: return @@ -1612,11 +1612,11 @@ class Standard(OutputHandler): The default output handler, writing plain lines with test outcome. Each test result is reported. For parallelized operation, the output - includes the thread number processing the test. + includes the worker number processing the test. """ def testStart(self, test): - self.output(test, self.threadPrefix(test), nl=False) + self.output(test, self.workerPrefix(test), nl=False) self.output(test, f"{test.displayName()} ...", nl=False) test._std_nl = False @@ -1643,7 +1643,7 @@ class Standard(OutputHandler): def finalMsg(self, test, msg): if test._std_nl: - self.output(test, self.threadPrefix(test), nl=False) + self.output(test, self.workerPrefix(test), nl=False) self.output(test, f"{test.displayName()} ...", nl=False) self.output(test, msg) @@ -1815,11 +1815,11 @@ class Brief(OutputHandler): pass def testFailed(self, test, msg): - self.output(test, self.threadPrefix(test), nl=False) + self.output(test, self.workerPrefix(test), nl=False) self.output(test, f"{test.displayName()} ... {msg}") def testUnstable(self, test, msg): - self.output(test, self.threadPrefix(test), nl=False) + self.output(test, self.workerPrefix(test), nl=False) self.output(test, f"{test.displayName()} ... {msg}") def testSkipped(self, test, msg): @@ -1830,7 +1830,7 @@ class Verbose(OutputHandler): """Output handler for producing the verbose output format.""" def testStart(self, test): - self.output(test, self.threadPrefix(test), nl=False) + self.output(test, self.workerPrefix(test), nl=False) self.output(test, f"{test.displayName()} ...") def testCommand(self, test, cmdline): @@ -1839,7 +1839,7 @@ class Verbose(OutputHandler): if cmdline.part > 1: part = f" [part #{cmdline.part}]" - self.output(test, self.threadPrefix(test), nl=False) + self.output(test, self.workerPrefix(test), nl=False) self.output(test, f" > {cmdline.cmdline}{part}") def testProgress(self, test, msg): @@ -1847,22 +1847,22 @@ class Verbose(OutputHandler): self.output(test, " - " + msg) def testSucceeded(self, test, msg): - self.output(test, self.threadPrefix(test), nl=False) + self.output(test, self.workerPrefix(test), nl=False) self.showTestVerbose(test) self.output(test, f"... {test.displayName()} {msg}") def testFailed(self, test, msg): - self.output(test, self.threadPrefix(test), nl=False) + self.output(test, self.workerPrefix(test), nl=False) self.showTestVerbose(test) self.output(test, f"... {test.displayName()} {msg}") def testUnstable(self, test, msg): - self.output(test, self.threadPrefix(test), nl=False) + self.output(test, self.workerPrefix(test), nl=False) self.showTestVerbose(test) self.output(test, f"... {test.displayName()} {msg}") def testSkipped(self, test, msg): - self.output(test, self.threadPrefix(test), nl=False) + self.output(test, self.workerPrefix(test), nl=False) self.showTestVerbose(test) self.output(test, f"... {test.displayName()} {msg}") From 05a2925f018e6d7a53d83e862dd340189ded9e8d Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Fri, 24 Jul 2026 18:35:46 +0200 Subject: [PATCH 20/21] Simplify `nextTests` index loop with `enumerate` --- btest | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/btest b/btest index 7ee310e..54eca93 100755 --- a/btest +++ b/btest @@ -409,9 +409,7 @@ class RunState: ): return None - for i in range(len(self._tests)): - t = self._tests[i] - + for i, t in enumerate(self._tests): if not t: continue From 2bbcc363145247a1b9437048c9992343e305927e Mon Sep 17 00:00:00 2001 From: Benjamin Bannier Date: Fri, 24 Jul 2026 18:47:10 +0200 Subject: [PATCH 21/21] Rename `mgr` to `state` in `Test` `TestManager` is gone; the parameter and attribute now hold a `RunState`. --- btest | 46 +++++++++++++++++++++++----------------------- 1 file changed, 23 insertions(+), 23 deletions(-) diff --git a/btest b/btest index 54eca93..b548c57 100755 --- a/btest +++ b/btest @@ -772,7 +772,7 @@ class Test: self.known_failure = False self.log = None self.measure_time = False - self.mgr = None + self.state = None self.name = None self.number = 1 @@ -998,14 +998,14 @@ class Test: self.known_failure |= part.known_failure self.measure_time |= part.measure_time - def getPorts(self, mgr, count): + def getPorts(self, state, count): if not count: return [] attempts = 5 while True: - rval = mgr.getAvailablePorts(count) + rval = state.getAvailablePorts(count) if rval: return rval @@ -1021,16 +1021,16 @@ class Test: time.sleep(15) - async def run(self, mgr): - bound_sockets = self.getPorts(mgr, len(self.ports)) + async def run(self, state): + bound_sockets = self.getPorts(state, len(self.ports)) self.bound_ports = [s.getsockname()[1] for s in bound_sockets] for bs in bound_sockets: bs.close() self.start = time.time() - self.mgr = mgr - mgr.testStart(self) + self.state = state + state.testStart(self) self.tmpdir = normalize_path_join(TmpDir, self.name) self.diag = normalize_path_join(self.tmpdir, ".diag") @@ -1040,7 +1040,7 @@ class Test: self.baselines.append(normalize_path_join(d, self.name)) self.diagmsgs = [] self.utime = -1 - self.utime_base = self.mgr.testTimingBaseline(self) + self.utime_base = self.state.testTimingBaseline(self) self.utime_perc = 0.0 self.utime_exceeded = False @@ -1102,7 +1102,7 @@ class Test: (success, rc) = await self.execute(cmd, apply_alternative=self.alternative) if not success: - self.mgr.testSkipped(self) + self.state.testSkipped(self) if not Options.tmps: self.rmTmp(with_close=True) self.finish() @@ -1173,12 +1173,12 @@ class Test: if Options.sphinx: # We still execute the remaining commands and # raise a failure for each one that fails. - self.mgr.testFailed(self) + self.state.testFailed(self) skip_part = cmd.part continue if failures == 1: - self.mgr.testFailed(self) + self.state.testFailed(self) if rc != 100: break @@ -1204,7 +1204,7 @@ class Test: failures += 1 if Options.sphinx or failures == 1: - self.mgr.testFailed(self) + self.state.testFailed(self) return need_teardown @@ -1218,7 +1218,7 @@ class Test: # and the user aborts the run. if rc == 200: # Flush remaining command output before aborting the run. - mgr.testReplayOutput(self) + state.testReplayOutput(self) raise Abort("Aborted by user.") self.utime_perc = 0.0 @@ -1238,10 +1238,10 @@ class Test: self.diagmsgs += [ f"'{self.name}' exceeded permitted execution time deviation{self.timePostfix()}" ] - self.mgr.testFailed(self) + self.state.testFailed(self) else: - self.mgr.testSucceeded(self) + self.state.testSucceeded(self) if not Options.tmps and self.reruns == 0: self.rmTmp(with_close=True) @@ -1250,7 +1250,7 @@ class Test: def finish(self): if self.bound_ports: - self.mgr.returnPorts(list(self.bound_ports)) + self.state.returnPorts(list(self.bound_ports)) self.bound_ports = [] @@ -1318,7 +1318,7 @@ class Test: if not success: return (False, rc) - self.mgr.testCommand(self, cmd) + self.state.testCommand(self, cmd) # Replace special names. @@ -1435,7 +1435,7 @@ class Test: try: for line in open(file): msg = line.strip() - self.mgr.testProgress(self, msg) + self.state.testProgress(self, msg) os.unlink(file) except OSError: @@ -1672,7 +1672,7 @@ class Console(OutputHandler): OutputHandler.__init__(self, options, reopen_std_file(sys.__stdout__)) def testStart(self, test): - msg = f"[{int(test.mgr.percentage()):>3d}%] {test.displayName()} ..." + msg = f"[{int(test.state.percentage()):>3d}%] {test.displayName()} ..." self.output(test, msg, nl=False) def testProgress(self, test, msg): @@ -1771,7 +1771,7 @@ class CompactConsole(Console): self._outfile.flush() def _consoleOutput(self, test, msg, sticky): - line = f"[{int(test.mgr.percentage()):>3d}%] {test.displayName()} ..." + line = f"[{int(test.state.percentage()):>3d}%] {test.displayName()} ..." if msg: line += " " + msg @@ -2196,12 +2196,12 @@ class OSC94ProgressBar(OutputHandler): def testFinished(self, test, msg): fmt = self.ProgressDefault - if test.mgr.hasFailure(): + if test.state.hasFailure(): fmt = self.ProgressError - elif test.mgr.hasUnstable(): + elif test.state.hasUnstable(): fmt = self.ProgressWarning - progress = fmt.format(percentage=int(test.mgr.percentage())) + progress = fmt.format(percentage=int(test.state.percentage())) sys.stdout.write(progress) sys.stdout.flush()