async def METHOD_NAME(duration: int, args, tests, results, status, *, with_profiler: bool = False) -> None:
    """Run the listed tests either in parallel asynchronously or in a thread pool."""
    # Kick of a status monitoring thread.
    stop_status = threading.Event()
    status_thread = threading.Thread(
        target=_report_status,
        args=(status, tests, stop_status),
        daemon=True)
    status_thread.start()
    try:
        if args.sync:
            with ThreadPoolExecutor(max_workers=args.parallel) as ex:
                tasks = [ex.submit(test.run_all_sync, duration, run_profiler=with_profiler) for test in tests]
                wait(tasks, return_when=ALL_COMPLETED)
        else:
            tasks = [asyncio.create_task(test.run_all_async(duration, run_profiler=with_profiler)) for test in tests]
            await asyncio.wait(tasks, return_when=asyncio.ALL_COMPLETED)
        # If any of the parallel test runs raised an exception, let it be propagated, after all tasks have
        # completed.
        # TODO: This will only raise the first Exception encountered. Once we migrate the perf pipelines
        # to 3.11 we could refactor to use ExceptionGroups so all exceptions will be captured.
        for task in tasks:
            task.result()
        # Add final test results to the results queue to be accumulated by the parent process.
        for test in tests:
            results.put((test._parallel_index, test.completed_operations, test.last_completion_time))
    finally:
        # Clean up status reporting thread.
        stop_status.set()
        status_thread.join()