gh-156082: Do not run tests using the same machine-wide resource at once

Doc/library/test.rst

@@ -836,6 +836,28 @@ The :mod:`!test.support` module defines the following functions:
    the trace function.
 
 
+.. decorator:: exclusive(resource)
+
+   Run no other test using the machine-wide *resource* while the decorated test
+   runs.
+   It can decorate a test method or a :class:`~unittest.TestCase` subclass.
+
+   Tests using different resources still run in parallel.
+   A test run outside the test runner, or not in parallel, simply runs.
+
+
+.. function:: requires_exclusive(resource)
+
+   Run no other test using the machine-wide *resource* until this process ends.
+
+   For a resource which the whole test file needs, call it next to
+   :func:`requires`.
+
+   :func:`requires` and :func:`requires_resource` call it themselves for the
+   resources which are machine-wide, such as ``'gui'`` and ``'audio'``, so most
+   tests need neither this function nor :func:`exclusive`.
+
+
 .. decorator:: bigmemtest(size, memuse, dry_run=True)
 
    Decorator for bigmem tests.

Lib/test/libregrtest/run_workers.py

@@ -526,7 +526,18 @@ def __init__(self, num_workers: int, runtests: RunTests,
             # these worker threads would never get anything to do.
             self.num_workers = min(self.num_workers, jobs)
 
+    def start_exclusive_locks(self) -> None:
+        # Tests using a machine-wide resource lock each other out through files
+        # in this directory.  A single worker process runs one test at a time
+        # anyway, so it needs no locks.
+        if self.num_workers > 1:
+            os.environ[support.EXCLUSIVE_ENV] = tempfile.gettempdir()
+
+    def stop_exclusive_locks(self) -> None:
+        os.environ.pop(support.EXCLUSIVE_ENV, None)
+
     def start_workers(self) -> None:
+        self.start_exclusive_locks()
         self.workers = [WorkerThread(index, self)
                         for index in range(1, self.num_workers + 1)]
         jobs = self.runtests.get_jobs()
@@ -666,4 +677,5 @@ def run(self) -> None:
             # worker when we exit this function
             self.pending.stop()
             self.stop_workers()
+            self.stop_exclusive_locks()
             self.logger.get_mem_usage = None

Lib/test/support/__init__.py

@@ -40,6 +40,7 @@
     "has_fork_support", "requires_fork",
     "has_subprocess_support", "requires_subprocess",
     "has_socket_support", "requires_working_socket",
+    "exclusive", "requires_exclusive",
     "has_remote_subprocess_debugging", "requires_remote_subprocess_debugging",
     "anticipate_failure", "load_package_tests", "detect_api_mismatch",
     "check__all__", "skip_if_buggy_ucrt_strfptime",
@@ -336,6 +337,133 @@ def get_resource_value(resource):
         return None
     return use_resources.get(resource)
 
+# Resources which are machine-wide: two tests using one of them at the same
+# time interfere with each other.
+EXCLUSIVE_RESOURCES = frozenset({'audio', 'bigmem', 'console', 'curses',
+                                 'gui'})
+
+# The environment variable naming the directory in which the test runner keeps
+# the lock files.  It is set only when tests are run in parallel.
+EXCLUSIVE_ENV = '_PYTHON_TEST_EXCLUSIVE_DIR'
+
+# Resources locked for the lifetime of this process: resource -> file descriptor.
+_exclusive_locks = {}
+
+
+def _lock_file(fd):
+    try:
+        import fcntl
+    except ImportError:
+        import msvcrt
+        import time
+        while True:
+            try:
+                msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
+            except OSError:
+                # It returns at once if the file is locked, so wait ourselves.
+                time.sleep(0.1)
+            else:
+                return
+    else:
+        fcntl.flock(fd, fcntl.LOCK_EX)
+
+
+def _open_exclusive_lock(resource):
+    """Open and lock the file of *resource*, or return None if not applicable."""
+    from test.support import isolation
+    if isolation.runningInSubprocess:
+        # The process which started this one holds the lock already.
+        return None
+    directory = os.environ.get(EXCLUSIVE_ENV)
+    if not directory:
+        # The tests are not run in parallel, so there is nothing to exclude.
+        return None
+    # The file is left behind: another process may be waiting on a lock on it,
+    # and it is empty anyway.  Nothing is ever written in it.
+    path = os.path.join(directory, f'python-test-exclusive-{resource}.lock')
+    if MS_WINDOWS:
+        flags = os.O_WRONLY | os.O_CREAT
+    else:
+        # flock() locks a file open for reading too, so that the tests of
+        # another user, who cannot write the file, are locked out as well.
+        flags = os.O_RDONLY | os.O_CREAT | os.O_NOFOLLOW
+    try:
+        fd = os.open(path, flags, 0o666)
+    except OSError:
+        return None
+    try:
+        _lock_file(fd)
+    except (ImportError, OSError):
+        # Locking is not available here, so the tests are not serialized.
+        os.close(fd)
+        return None
+    except BaseException:
+        os.close(fd)
+        raise
+    return fd
+
+
+def requires_exclusive(resource):
+    """Run no other test using *resource* until this process ends.
+
+    For a resource which the whole test file needs, next to requires().
+    """
+    if resource not in _exclusive_locks:
+        fd = _open_exclusive_lock(resource)
+        if fd is not None:
+            _exclusive_locks[resource] = fd
+
+
+@contextlib.contextmanager
+def _exclusive_resource(resource):
+    """Run no other test using *resource* until the end of the block."""
+    if resource in _exclusive_locks:
+        # This process holds it already.
+        yield
+        return
+    fd = _open_exclusive_lock(resource)
+    if fd is None:
+        yield
+        return
+    _exclusive_locks[resource] = fd
+    try:
+        yield
+    finally:
+        del _exclusive_locks[resource]
+        os.close(fd)
+
+
+def exclusive(resource):
+    """Decorator to run no other test using *resource* while this one runs.
+
+    It can decorate a test method or a whole TestCase subclass.
+    """
+    def decorator(obj):
+        if isinstance(obj, type):
+            # Call the function itself, not the method bound to this class,
+            # so that a subclass gets its own class.
+            setup = obj.setUpClass.__func__
+            @classmethod
+            @functools.wraps(setup)
+            def setUpClass(cls):
+                cls.enterClassContext(_exclusive_resource(resource))
+                setup(cls)
+            obj.setUpClass = setUpClass
+            return obj
+        if inspect.iscoroutinefunction(obj):
+            @functools.wraps(obj)
+            async def wrapper(*args, **kwargs):
+                with _exclusive_resource(resource):
+                    return await obj(*args, **kwargs)
+        else:
+            @functools.wraps(obj)
+            def wrapper(*args, **kwargs):
+                with _exclusive_resource(resource):
+                    return obj(*args, **kwargs)
+        return wrapper
+    return decorator
+
+
 def requires(resource, msg=None):
     """Raise ResourceDenied if the specified resource is not available."""
     if not is_resource_enabled(resource):
@@ -346,6 +474,8 @@ def requires(resource, msg=None):
         raise ResourceDenied("No socket support")
     if resource == 'gui' and not _is_gui_available():
         raise ResourceDenied(_is_gui_available.reason)
+    if resource in EXCLUSIVE_RESOURCES:
+        requires_exclusive(resource)
 
 def _requires_unix_version(sysname, min_version):
     """Decorator raising SkipTest if the OS is `sysname` and the version is less
@@ -1336,25 +1466,28 @@ def wrapper(self):
                 # Watch it from here: the output of the subprocess is captured.
                 cls = type(self)
                 qualname = f'{cls.__qualname__}.{f.__name__}'
-                proc = isolation._start_test(cls.__module__, qualname)
-                watchdog = _memory_watchdog(proc.pid) if verbose else None
-                payload, output, returncode = proc.wait(tick=watchdog)
-                if watchdog:
-                    # The subprocess measures its own peak exactly.  What the
-                    # parent sampled is only a lower bound.
-                    maxrss = payload and payload.get('maxrss')
-                    peak = maxrss or watchdog.peak
-                    if peak:
-                        print(f" ... peak memory use: "
-                              f"{peak / (1024 ** 3):.1f} GiB"
-                              f"{'' if maxrss else ' or more'}", flush=True)
-                    majflt = payload and payload.get('majflt')
-                    if majflt:
-                        # The test did not fit in memory, so its timing means
-                        # little.
-                        print(f" ... {majflt} major page faults: the test "
-                              f"waited for the disk", flush=True)
-                isolation._replay_test(self, payload, output, returncode)
+                # A real run allocates most of the memory of the machine.
+                with _exclusive_resource('bigmem'):
+                    proc = isolation._start_test(cls.__module__, qualname)
+                    watchdog = _memory_watchdog(proc.pid) if verbose else None
+                    payload, output, returncode = proc.wait(tick=watchdog)
+                    if watchdog:
+                        # The subprocess measures its own peak exactly.
+                        # What the parent sampled is only a lower bound.
+                        maxrss = payload and payload.get('maxrss')
+                        peak = maxrss or watchdog.peak
+                        if peak:
+                            print(f" ... peak memory use: "
+                                  f"{peak / (1024 ** 3):.1f} GiB"
+                                  f"{'' if maxrss else ' or more'}",
+                                  flush=True)
+                        majflt = payload and payload.get('majflt')
+                        if majflt:
+                            # The test did not fit in memory, so its
+                            # timing means little.
+                            print(f" ... {majflt} major page faults: the test "
+                                  f"waited for the disk", flush=True)
+                    isolation._replay_test(self, payload, output, returncode)
                 return
 
             return f(self, maxsize)
@@ -1406,6 +1539,8 @@ def requires_resource(resource):
     if resource == 'gui' and not _is_gui_available():
         return unittest.skip(_is_gui_available.reason)
     if is_resource_enabled(resource):
+        if resource in EXCLUSIVE_RESOURCES:
+            return exclusive(resource)
         return _id
     else:
         return unittest.skip("resource {0!r} is not enabled".format(resource))

Misc/NEWS.d/next/Tests/2026-08-19-21-35-06.gh-issue-156082.fvnqct.rst

@@ -0,0 +1,5 @@
+Tests using a machine-wide resource, such as the display or the sound device,
+no longer run at the same time as other tests using the same resource.  Add
+:func:`~test.support.exclusive` and :func:`~test.support.requires_exclusive`;
+:func:`~test.support.requires`, :func:`~test.support.requires_resource` and
+:func:`~test.support.bigmemtest` use them for such resources.