Function _fire_one

tools/loadtest.py:239–277

(
    session,
    url: str,
    rng: random.Random,
    results: list[Result],
    test_start: float,
    simple_ratio: float,
)

async def _fire_one(
    session,
    url: str,
    rng: random.Random,
    results: list[Result],
    test_start: float,
    simple_ratio: float,
) -> None:
    if rng.random() < simple_ratio:
        cmd = rng.choice(SIMPLE_CMDS)
    else:
        cmd = gen_bot_cmd(rng)
    target = f"{url}/explain"
    t0 = time.perf_counter()
    try:
        async with session.get(target, params={"cmd": cmd}, timeout=60) as resp:
            body = await resp.read()
            lat = (time.perf_counter() - t0) * 1000.0
            results.append(
                Result(
                    ok=resp.status == 200,
                    status=resp.status,
                    latency_ms=lat,
                    bytes=len(body),
                    t_end_s=time.monotonic() - test_start,
                )
            )
    except Exception as e:
        lat = (time.perf_counter() - t0) * 1000.0
        results.append(
            Result(
                ok=False,
                status=0,
                latency_ms=lat,
                bytes=0,
                t_end_s=time.monotonic() - test_start,
                err=str(e)[:80],
            )
        )


async def _closed_loop_worker(