Simon Jakobi pushed to branch wip/sjakobi/testsuite-atomic-output at Glasgow Haskell Compiler / GHC Commits: 5e4892c2 by Simon Jakobi at 2026-06-11T21:41:49+02:00 testsuite: Emit each test's output atomically Tests run concurrently (as asyncio tasks) and report progress and failures with many separate print() calls, some to stdout and some to stderr. As a result the lines of concurrently failing tests interleave arbitrarily, and since the two streams have different buffering when piped (as in CI), even a single test's lines arrive scrambled relative to each other. CI logs show diff headers separated from their diffs and failure messages wedged between unrelated tests' progress markers. Replace sys.stdout/sys.stderr with proxies that redirect writes into a per-asyncio-task buffer while a test is running (a context variable, following the existing testopts_ctx_var pattern). When the test completes, its accumulated output is written to the real stdout in a single call and flushed, so each test's output appears as one contiguous, correctly ordered block. Routing both streams through one buffer also eliminates the stdout/stderr reordering, and the explicit flush after each test means output now reaches CI logs as tests finish rather than sitting in a block buffer until the end of the run (cf #12934). Writes from contexts without an active buffer (driver preamble and summary, terminal-title updates from timer threads) pass through to the real streams unchanged. Closes #27367. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> - - - - - 2 changed files: - testsuite/driver/runtests.py - testsuite/driver/testlib.py Changes: ===================================== testsuite/driver/runtests.py ===================================== @@ -487,6 +487,7 @@ if config.list_broken: print('') else: # Now run all the tests + install_output_proxies() # avoid interleaved output from concurrent tests try: async def run_parallelTests(): sem = asyncio.Semaphore(config.threads) ===================================== testsuite/driver/testlib.py ===================================== @@ -88,6 +88,61 @@ def get_all_ways() -> Set[WayName]: global testopts_ctx_var testopts_ctx_var = contextvars.ContextVar('testopts_ctx_var') # type: ignore +# Pipe each test's output into a per-test buffer (set up by runTestAtomically) +# to avoid interleaving the output of concurrent tests. Writes from contexts +# with no active buffer pass straight through to the real streams. + +output_buffer_ctx_var = contextvars.ContextVar('output_buffer_ctx_var', default=None) # type: contextvars.ContextVar[Optional[io.StringIO]] + +class _OutputProxyBuffer: + """The .buffer of an _OutputProxy; takes bytes.""" + def __init__(self, real) -> None: + self._real = real + + def write(self, b: bytes) -> None: + buf = output_buffer_ctx_var.get() + if buf is None: + self._real.buffer.write(b) + else: + buf.write(b.decode('utf-8', errors='backslashreplace')) + + def flush(self) -> None: + if output_buffer_ctx_var.get() is None: + self._real.buffer.flush() + +class _OutputProxy: + def __init__(self, real) -> None: + self._real = real + self.buffer = _OutputProxyBuffer(real) + + @property + def encoding(self) -> str: + return self._real.encoding + + def write(self, s: str) -> int: + buf = output_buffer_ctx_var.get() + if buf is None: + return self._real.write(s) + return buf.write(s) + + def flush(self) -> None: + if output_buffer_ctx_var.get() is None: + self._real.flush() + + def isatty(self) -> bool: + return self._real.isatty() + + def fileno(self) -> int: + return self._real.fileno() + +def install_output_proxies() -> None: + # Both streams feed the same per-task buffer, so a test's stdout and + # stderr stay in print order. + if not isinstance(sys.stdout, _OutputProxy): + sys.stdout = _OutputProxy(sys.stdout) + if not isinstance(sys.stderr, _OutputProxy): + sys.stderr = _OutputProxy(sys.stderr) + def getTestOpts() -> TestOptions: return testopts_ctx_var.get() @@ -1502,9 +1557,25 @@ allTestNames = set([]) # type: Set[TestName] async def runTest(sem, opts, name: TestName, func, args): if sem is None: - return await test_common_work(name, opts, func, args) + return await runTestAtomically(opts, name, func, args) async with sem: + return await runTestAtomically(opts, name, func, args) + +async def runTestAtomically(opts, name: TestName, func, args): + # Buffer this test's output and emit it as one block at the end, so that + # concurrent tests' output does not interleave. + buf = io.StringIO() + token = output_buffer_ctx_var.set(buf) + try: return await test_common_work(name, opts, func, args) + finally: + output_buffer_ctx_var.reset(token) + s = buf.getvalue() + if s: + # The event loop is single-threaded and there is no await between + # these calls, so the block is written out atomically. + sys.stdout.write(s) + sys.stdout.flush() # name :: String # setup :: [TestOpt] -> IO () View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/5e4892c2a223f99e0b4156e2d212f5ff... -- View it on GitLab: https://gitlab.haskell.org/ghc/ghc/-/commit/5e4892c2a223f99e0b4156e2d212f5ff... You're receiving this email because of your account on gitlab.haskell.org.