59a0a3844c
PR Test AMD / cancel-on-close (push) Has been skipped
PR Test NVIDIA ARM / scan (push) Has been skipped
PR Test NVIDIA / cancel-on-close (push) Has been skipped
PR Test AMD / scan (push) Has been skipped
PR Test NVIDIA ARM / cancel-on-close (push) Has been skipped
PR Test NVIDIA / scan (push) Has been skipped
Release Docker Images / build (cu129-torch-2.11.0) (push) Has been skipped
Release Docker Images / build (cu130-torch-2.11.0) (push) Has been skipped
Release PyPI / publish (push) Has been skipped
Scheduler Python Test / test (push) Successful in 27m19s
Docs / build (push) Successful in 28m8s
Scheduler C++ Test / test (push) Successful in 28m19s
Scheduler C++ Test / test-flat (push) Successful in 28m18s
Docs / deploy (push) Has been cancelled
PR Test AMD / finish (push) Has been cancelled
PR Test NVIDIA / finish (push) Has been cancelled
PR Test NVIDIA ARM / finish (push) Has been cancelled
PR Test NVIDIA ARM / ${{ matrix.name }} (${{ matrix.runner }}) (push) Has been cancelled
PR Test AMD / ${{ matrix.name }} (${{ matrix.runner }}) (push) Has been cancelled
PR Test NVIDIA / ${{ matrix.name }} (${{ matrix.runner }}) (push) Has been cancelled
133 lines
5.3 KiB
Python
133 lines
5.3 KiB
Python
# Copyright (c) 2026 LightSeek Foundation
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
# of this software and associated documentation files (the "Software"), to deal
|
|
# in the Software without restriction, including without limitation the rights
|
|
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
# copies of the Software, and to permit persons to whom the Software is
|
|
# furnished to do so, subject to the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be included in
|
|
# all copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
|
# SOFTWARE.
|
|
|
|
"""Tests for the per-request timeout guards in :mod:`tokenspeed.bench`.
|
|
|
|
These tests don't talk to a real server; they exercise the timeout helper
|
|
directly. The point is to lock in the behaviour that one stuck
|
|
stream-response future cannot deadlock the outer ``asyncio.gather``: instead
|
|
it surfaces as a normal ``RequestFuncOutput`` marked failed.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from tokenspeed import bench
|
|
|
|
|
|
def test_aiohttp_timeout_has_sock_subtimeouts():
|
|
"""Regression guard: a future refactor must keep ``sock_read`` set.
|
|
|
|
Without ``sock_read``, ``aiohttp.StreamReader._wait`` awaits forever on
|
|
a silent socket and one stuck stream-response future blocks the entire
|
|
benchmark's ``asyncio.gather`` at high concurrency. The exact value is
|
|
user-tunable via env; we only assert it's bounded and strictly tighter
|
|
than ``total`` so an indefinitely silent socket still surfaces.
|
|
"""
|
|
timeout = bench.AIOHTTP_TIMEOUT
|
|
assert timeout.sock_read is not None, "AIOHTTP_TIMEOUT must set sock_read"
|
|
assert timeout.total is not None
|
|
assert 0 < timeout.sock_read < timeout.total, (
|
|
f"sock_read {timeout.sock_read} must be smaller than total "
|
|
f"{timeout.total} so it actually fires before the umbrella"
|
|
)
|
|
assert (
|
|
timeout.sock_connect is not None
|
|
), "AIOHTTP_TIMEOUT must also set sock_connect"
|
|
assert 0 < timeout.sock_connect <= timeout.sock_read
|
|
|
|
|
|
def test_per_request_timeout_constant_is_positive():
|
|
assert bench.PER_REQUEST_TIMEOUT_SEC > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_await_with_per_request_timeout_returns_failed_output_on_hang(
|
|
monkeypatch,
|
|
):
|
|
"""A stuck request must time out instead of blocking forever."""
|
|
monkeypatch.setattr(bench, "PER_REQUEST_TIMEOUT_SEC", 0.1)
|
|
|
|
async def stuck_request() -> bench.RequestFuncOutput:
|
|
await asyncio.sleep(60) # would block past the gather without the wrap
|
|
return bench.RequestFuncOutput() # pragma: no cover
|
|
|
|
start = time.perf_counter()
|
|
output = await bench.await_with_per_request_timeout(stuck_request(), prompt_len=42)
|
|
elapsed = time.perf_counter() - start
|
|
|
|
assert elapsed < 1.0, f"expected sub-second timeout, took {elapsed:.2f}s"
|
|
assert output.success is False
|
|
assert "per-request timeout" in output.error
|
|
assert output.prompt_len == 42
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_await_with_per_request_timeout_passes_through_success(monkeypatch):
|
|
"""The wrap must not perturb a request that completes normally."""
|
|
monkeypatch.setattr(bench, "PER_REQUEST_TIMEOUT_SEC", 5.0)
|
|
|
|
async def fast_request() -> bench.RequestFuncOutput:
|
|
output = bench.RequestFuncOutput()
|
|
output.success = True
|
|
output.prompt_len = 13
|
|
output.generated_text = "hello"
|
|
return output
|
|
|
|
result = await bench.await_with_per_request_timeout(fast_request(), prompt_len=13)
|
|
|
|
assert result.success is True
|
|
assert result.generated_text == "hello"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_concurrent_stuck_request_does_not_block_gather(monkeypatch):
|
|
"""End-to-end shape: stuck + healthy requests gather together cleanly."""
|
|
monkeypatch.setattr(bench, "PER_REQUEST_TIMEOUT_SEC", 0.2)
|
|
|
|
async def stuck() -> bench.RequestFuncOutput:
|
|
await asyncio.sleep(60)
|
|
return bench.RequestFuncOutput() # pragma: no cover
|
|
|
|
async def healthy(latency: float) -> bench.RequestFuncOutput:
|
|
await asyncio.sleep(latency)
|
|
out = bench.RequestFuncOutput()
|
|
out.success = True
|
|
return out
|
|
|
|
start = time.perf_counter()
|
|
results = await asyncio.gather(
|
|
bench.await_with_per_request_timeout(stuck(), prompt_len=1),
|
|
bench.await_with_per_request_timeout(healthy(0.05), prompt_len=2),
|
|
bench.await_with_per_request_timeout(stuck(), prompt_len=3),
|
|
bench.await_with_per_request_timeout(healthy(0.05), prompt_len=4),
|
|
)
|
|
elapsed = time.perf_counter() - start
|
|
|
|
# Without the timeout wrap this gather would block on the two stuck
|
|
# requests forever; with it the gather returns in roughly the timeout.
|
|
assert elapsed < 1.5, f"gather elapsed {elapsed:.2f}s, expected ~0.2s"
|
|
assert [r.success for r in results] == [False, True, False, True]
|
|
assert all("per-request timeout" in r.error for r in results if not r.success)
|