GitLab

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 ()