+    # I/O manager discovery.

+    #

+    # Which I/O managers (GHC's `--io-manager` RTS flag) a given RTS build

+    # supports depends on the target platform and on whether the RTS is the

+    # threaded or non-threaded flavour (see rts/IOManager.h's

+    # IOMGR_ENABLED_* macros). Rather than duplicate that platform/way

+    # dependent logic here, we ask the RTS itself: giving it a bogus

+    # '--io-manager=' choice makes it print, on stderr, the list of I/O

+    # managers it was actually built to support (see IOManager.h's

+    # IOMGRS_ENABLED_STR, used when reporting the error in RtsFlags.c),

+    # which we then parse out.

+    def discover_io_managers_generic(extra_flags):

+        import tempfile

+

+        test_src = 'main :: IO ()\nmain = return ()\n'

+

+        res = [None]

+

+        with tempfile.TemporaryDirectory() as d:

+            src = Path(d) / 'IOMgrProbe.hs'

+            src.write_text(test_src)

+            exe = Path(d) / ('IoMgrProbe' + config.exeext)

+

+            # -rtsopts is needed so that the resulting binary will even

+            # consider an --io-manager= flag (rather than immediately

+            # rejecting all "unsafe" RTS options).

+            cmd = '{} -v0 -rtsopts -o {} {} {}'.format(

+                                    config.compiler,

+                                    exe,

+                                    src,

+                                    ' '.join(extra_flags))

+            try:

+                cp = subprocess.run(

+                    cmd,

+                    shell=True,

+                    cwd=d,

+                    stdout=subprocess.DEVNULL,

+                    stderr=None if config.verbose >= 3 else subprocess.DEVNULL)

+                if cp.returncode != 0 or not exe.exists():

+                    raise Exception("Error in testsuite/config/ghc.discover_io_managers: failed to compile with: {}".format(cmd))

+                res = list(discover_io_managers(exe, d))

+

+            except Exception as err:

+                print("Exception thrown in testsuite/config/ghc.discover_io_managers_generic: %s", err)

+

+        return res

+

+    config.io_managers_nonthreaded = discover_io_managers_generic([])

+    config.io_managers_threaded = (

+        discover_io_managers_generic(['-threaded']) if config.ghc_with_threaded_rts

+        else [])

+    config.io_managers_all = list(set(config.io_managers_nonthreaded).union(set(config.io_managers_threaded)))

+

+from testutil import discover_io_managers, getStdout, str_warn, str_info, print_table, shorten_metric_name, str_removeprefix

+        # I/O managers (GHC's `--io-manager` RTS flag) supported by the threaded runtime.

+        self.io_managers_threaded = []  # type: List[str]

+

+        # I/O managers (GHC's `--io-manager` RTS flag) supported by the non-threaded runtime.

+        self.io_managers_nonthreaded = []  # type: List[str]

+

+        # All I/O managers (GHC's `--io-manager` RTS flag)

+        self.io_managers_all = []  # type: List[str]

+

+       # IO Managers to test

+       # * None - only default IO Manager

+       # * [...] - list of io managers e.g. "mio" or "poll"

+       self.io_managers = [None] # type: List[str|None]

+

+from testutil import discover_io_managers, strip_quotes, lndir, link_or_copy_file, passed, \

+# -----

+

+def _io_managers( name: TestName, opts, io_mgrs: List[str] ) -> None:

+    opts.io_managers = io_mgrs

+

+def io_managers( io_mgrs: List[str] ):

+    """

+    Run this test once for each of the given I/O managers (identified by

+    the name used with GHC's `--io-manager` RTS flag, e.g. 'mio',

+    'select', 'winio'), instead of the usual set of ways. I/O managers

+    that are not supported on the current platform/build are silently

+    skipped.

+    """

+    return lambda name, opts: _io_managers(name, opts, io_mgrs)

+

+def all_io_managers( name: TestName, opts ):

+    """

+    Run this test once for each I/O manager supported by this

+    platform/build.

+    """

+    _io_managers(name, opts, config.io_managers_all)

+

+def all_io_managers_except(exclude: List[str]):

+    """

+    Like all_io_managers, but excluding the given I/O managers.

+    """

+    return lambda name, opts: _io_managers(name, opts, list(set(config.io_managers_all) - set(exclude)))

+def way_supports_io_manager(way: WayName, io_manager: None | str) -> bool:

+    if not io_manager:

+        return True

+    way_is_threaded = '-threaded' in config.way_rts_flags[way]

+    return io_manager in config.io_managers_threaded if way_is_threaded else io_manager in config.io_managers_nonthreaded

+

+        # We run each way over each io manager (or just None meaning the default IO Manager)

+        all_ways_and_io_managers = dict(

+            (

+                way,

+                [io_manager

+                    for io_manager in opts.io_managers

+                    if way_supports_io_manager(way, io_manager)

+                ]

+            )

+            for way in all_ways

+        )

+

+        for io_managers in all_ways_and_io_managers.values():

+            t.total_test_cases += len(io_managers)

+            for io_manager in all_ways_and_io_managers[way]:

+                if stopping():

+                    break

+                try:

+                    await do_test(

+                        name,

+                        way,

+                        io_manager,

+                        func,

+                        args,

+                        files)

+                except KeyboardInterrupt:

+                    stopNow()

+                except Exception as e:

+                    traceback.print_exc()

+                    framework_fail(name, way, traceback.format_exc())

+

+        for skipped_way in set(all_ways) - set(do_ways):

+            t.n_tests_skipped += len(all_ways_and_io_managers[skipped_way])

+            io_manager: None | str,

+    if io_manager:

+        full_name += ' (IO Manager: {})'.format(io_manager)

+

+    func_has_io_manager_param = func in [

+        backpack_run,

+        compile_and_run,

+        ghci_multiunit_script,

+        ghci_script,

+        multi_compile_and_run,

+        multimod_compile_and_run,

+        warn_and_run

+    ]

+

+    io_manager_arg = [io_manager] if func_has_io_manager_param else []

+

+    result = await func(*[name,way] + io_manager_arg + args)

+    return await simple_run( name, WayName(''), None, override_options(cmd), '' )

+async def ghci_script( name, way, io_manager, script):

+    return await simple_run( name, way, io_manager, cmd, getTestOpts().extra_run_opts )

+async def ghci_multiunit_script( name, way, io_manager, units, script):

+    return await simple_run( name, way, io_manager, cmd, getTestOpts().extra_run_opts )

+                      io_manager: None | str,

+        return await simple_run( name, way, io_manager, cmd, getTestOpts().extra_run_opts )

+async def compile_and_run( name, way, io_manager, extra_hc_opts ):

+    return await compile_and_run__( name, way, io_manager, None, [], extra_hc_opts)

+async def backpack_run( name, way, io_manager, extra_hc_opts ):

+    return await compile_and_run__( name, way, io_manager, None, [], extra_hc_opts, backpack=True )

+async def multimod_compile_and_run( name, way, io_manager, top_mod, extra_hc_opts ):

+    return await compile_and_run__( name, way, io_manager, top_mod, [], extra_hc_opts)

+async def multi_compile_and_run( name, way, io_manager, top_mod, extra_mods, extra_hc_opts ):

+    return await compile_and_run__( name, way, io_manager, top_mod, extra_mods, extra_hc_opts)

+async def warn_and_run( name, way, io_manager, extra_hc_opts ):

+    return await compile_and_run__( name, way, io_manager, None, [], extra_hc_opts, compile_stderr = True)

+async def simple_run(name: TestName, way: WayName, io_manager: None | str, prog: str, extra_run_opts: str) -> Any:

+    if io_manager:

+        my_rts_flags += '+RTS --io-manager={0} -RTS'.format(io_manager)

+

+    # Check that IO Manager is available

+    if io_manager:

+        available_io_managers = discover_io_managers(prog, opts.testdir)

+        if io_manager not in available_io_managers:

+            framework_fail(name, way, 'IO Manager {} unexpectedly unavailable (available IO Managers: {}). `-non-threaded`/`-threaded` is only supported through way RTS flags (not manually added via test modifiers/args) when testing over IO Managers.'.format(io_manager, available_io_managers))

+

+import re

+

+# I/O manager discovery for a haskell binary. Returns the set of IO Manager

+# strings supported by the binary

+def discover_io_managers(hs_exe: str | Path, cwd: str | Path) -> Set[str]:

+    rp = subprocess.run(

+        [str(hs_exe), '+RTS', '--io-manager=not-a-real-io-manager', '-RTS'],

+        cwd=cwd,

+        stdout=subprocess.DEVNULL,

+        stderr=subprocess.PIPE)

+    stderr = rp.stderr.decode('utf-8', errors='replace')

+

+    m = re.search(r'choices are:\s*(.*)', stderr)

+    if not m:

+        raise Exception("Expected stderr output of form 'choices are: ...' but got: {}".format(stderr))

+

+    # 'auto' is a meta-choice (let the RTS pick a default for us), not

+    # an I/O manager implementation in its own right.

+    return set(m.group(1).split()) - {'auto'}

+test('IOManager', [js_skip, when(arch('wasm32'), skip),

+                   pre_cmd('$MAKE -s --no-print-directory IOManager.hs'), all_io_managers_except('winio')],