Function _witness

tools/loadtest.py:543–583

(
    url: str,
    rng: random.Random,
    deadline: float,
    interval_s: float,
    witness_results: list[Result],
    test_start: float,
)

async def _witness(
    url: str,
    rng: random.Random,
    deadline: float,
    interval_s: float,
    witness_results: list[Result],
    test_start: float,
) -> None:
    import aiohttp

    timeout = aiohttp.ClientTimeout(total=120)
    connector = aiohttp.TCPConnector(limit=4, force_close=True)
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        while time.perf_counter() < deadline:
            t0 = time.perf_counter()
            status = 0
            err = ""
            nbytes = 0
            try:
                async with session.get(
                    f"{url}/", timeout=aiohttp.ClientTimeout(total=60)
                ) as resp:
                    status = resp.status
                    body = await resp.read()
                    nbytes = len(body)
                    ok = 200 <= status < 400
            except Exception as e:
                ok = False
                err = f"{type(e).__name__}: {e}"[:120]
            latency_ms = (time.perf_counter() - t0) * 1000.0
            witness_results.append(
                Result(
                    ok=ok,
                    status=status,
                    latency_ms=latency_ms,
                    bytes=nbytes,
                    t_end_s=time.perf_counter() - test_start,
                    err=err,
                )
            )
            await asyncio.sleep(max(0.0, interval_s - (time.perf_counter() - t0)))


async def run_slowread(args: argparse.Namespace) -> dict: