Function _closed_loop_worker

tools/loadtest.py:280–290

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

async def _closed_loop_worker(
    session,
    url: str,
    rng: random.Random,
    results: list[Result],
    test_start: float,
    stop_at: float,
    simple_ratio: float,
) -> None:
    while time.monotonic() < stop_at:
        await _fire_one(session, url, rng, results, test_start, simple_ratio)


@dataclass