Function run_http

tools/loadtest.py:341–384

(args: argparse.Namespace)

async def run_http(args: argparse.Namespace) -> dict:
    import aiohttp

    rng = random.Random(args.seed)
    results: list[Result] = []
    test_start = time.monotonic()
    stop_at = test_start + args.duration

    connector = aiohttp.TCPConnector(limit=max(args.concurrency * 2, 64))
    timeout = aiohttp.ClientTimeout(total=None)

    loop_stats = LoopStats()
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        if args.rps:
            await _open_loop_scheduler(
                session,
                args.url,
                rng,
                results,
                test_start,
                stop_at,
                args.rps,
                args.simple_ratio,
                args.concurrency,
                loop_stats,
            )
        else:
            workers = [
                asyncio.create_task(
                    _closed_loop_worker(
                        session,
                        args.url,
                        rng,
                        results,
                        test_start,
                        stop_at,
                        args.simple_ratio,
                    )
                )
                for _ in range(args.concurrency)
            ]
            await asyncio.gather(*workers)

    return _build_report(results, loop_stats, args)


def _build_report(