mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-07 07:58:09 +00:00
test(caching): run loop-reuse tests until a loop address repeats
The collected-loop tests ran a fixed 50 loops and passed with the fix reverted whenever no loop reused an earlier address. They now stop at the first reuse and skip if none happens, through a shared helper.
This commit is contained in:
parent
4412f79e04
commit
f8e2fcdbbf
3 changed files with 42 additions and 26 deletions
20
tests/unit/caching/_event_loop_reuse.py
Normal file
20
tests/unit/caching/_event_loop_reuse.py
Normal file
|
|
@ -0,0 +1,20 @@
|
|||
from collections.abc import Callable
|
||||
from typing import Final, TypeVar
|
||||
|
||||
import pytest
|
||||
|
||||
LoopResult = TypeVar("LoopResult")
|
||||
|
||||
|
||||
def runs_until_a_loop_address_repeats(
|
||||
run_on_new_loop: Callable[[], tuple[int, LoopResult]],
|
||||
earlier: tuple[tuple[int, LoopResult], ...] = (),
|
||||
limit: int = 50,
|
||||
) -> tuple[tuple[int, LoopResult], ...]:
|
||||
if len(earlier) == limit:
|
||||
pytest.skip(f"none of {limit} collected loops had its address handed to a later loop")
|
||||
latest: Final = run_on_new_loop()
|
||||
runs: Final = (*earlier, latest)
|
||||
if latest[0] in {loop_id for loop_id, _ in earlier}:
|
||||
return runs
|
||||
return runs_until_a_loop_address_repeats(run_on_new_loop, runs, limit)
|
||||
|
|
@ -19,6 +19,7 @@ from redis.asyncio import ConnectionPool, Redis
|
|||
|
||||
from litellm.caching.evicted_client_closer import EvictedClientCloser
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
|
||||
from tests.unit.caching._event_loop_reuse import runs_until_a_loop_address_repeats
|
||||
|
||||
|
||||
class FakeClock:
|
||||
|
|
@ -278,37 +279,28 @@ def test_sync_client_evicted_outside_an_event_loop_is_still_closed():
|
|||
def test_async_client_from_collected_loop_is_not_closed_by_fresh_loops() -> None:
|
||||
clock: Final = FakeClock()
|
||||
closer: Final = EvictedClientCloser(grace_seconds=0, clock=clock)
|
||||
client: Final = RecordingAsyncClient()
|
||||
closer.mark_owned(client)
|
||||
loop_a: Final = asyncio.new_event_loop()
|
||||
|
||||
async def schedule() -> None:
|
||||
closer.schedule(client)
|
||||
|
||||
try:
|
||||
loop_a.run_until_complete(schedule())
|
||||
finally:
|
||||
loop_a.close()
|
||||
del loop_a
|
||||
gc.collect()
|
||||
|
||||
async def reap() -> None:
|
||||
async def reap_then_schedule(client: RecordingAsyncClient) -> None:
|
||||
closer.reap()
|
||||
await asyncio.sleep(0)
|
||||
closer.schedule(client)
|
||||
|
||||
def reap_on_new_loop() -> None:
|
||||
def run_on_new_loop() -> tuple[int, RecordingAsyncClient]:
|
||||
client: Final = RecordingAsyncClient()
|
||||
closer.mark_owned(client)
|
||||
loop: Final = asyncio.new_event_loop()
|
||||
loop_id: Final = id(loop)
|
||||
try:
|
||||
loop.run_until_complete(reap())
|
||||
loop.run_until_complete(reap_then_schedule(client))
|
||||
finally:
|
||||
loop.close()
|
||||
del loop
|
||||
gc.collect()
|
||||
return loop_id, client
|
||||
|
||||
for _ in range(50):
|
||||
reap_on_new_loop()
|
||||
runs: Final = runs_until_a_loop_address_repeats(run_on_new_loop)
|
||||
|
||||
assert client.close_calls == 0
|
||||
assert all(client.close_calls == 0 for _, client in runs)
|
||||
|
||||
|
||||
def test_async_client_queued_on_a_live_loop_is_closed_only_by_that_loop() -> None:
|
||||
|
|
|
|||
|
|
@ -30,6 +30,7 @@ from litellm.llms.custom_httpx.http_handler import(
|
|||
from litellm.types.utils import LlmProviders
|
||||
from litellm.utils import _invalidate_model_cost_lowercase_map
|
||||
from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
|
||||
from tests.unit.caching._event_loop_reuse import runs_until_a_loop_address_repeats
|
||||
|
||||
|
||||
async def _get_and_set_cache(cache: LLMClientCache) -> object | None:
|
||||
|
|
@ -38,30 +39,32 @@ async def _get_and_set_cache(cache: LLMClientCache) -> object | None:
|
|||
return cached
|
||||
|
||||
|
||||
def _cache_result_on_new_loop(cache: LLMClientCache) -> object | None:
|
||||
def _cache_result_on_new_loop(cache: LLMClientCache) -> tuple[int, object | None]:
|
||||
loop: Final = asyncio.new_event_loop()
|
||||
loop_id: Final = id(loop)
|
||||
try:
|
||||
result: Final = loop.run_until_complete(_get_and_set_cache(cache))
|
||||
finally:
|
||||
loop.close()
|
||||
del loop
|
||||
gc.collect()
|
||||
return result
|
||||
return loop_id, result
|
||||
|
||||
|
||||
async def _get_cache_key(cache: LLMClientCache) -> str:
|
||||
return cache.update_cache_key_with_event_loop("client")
|
||||
|
||||
|
||||
def _cache_key_on_new_loop(cache: LLMClientCache) -> str:
|
||||
def _cache_key_on_new_loop(cache: LLMClientCache) -> tuple[int, str]:
|
||||
loop: Final = asyncio.new_event_loop()
|
||||
loop_id: Final = id(loop)
|
||||
try:
|
||||
key: Final = loop.run_until_complete(_get_cache_key(cache))
|
||||
finally:
|
||||
loop.close()
|
||||
del loop
|
||||
gc.collect()
|
||||
return key
|
||||
return loop_id, key
|
||||
|
||||
|
||||
class MockAsyncClient:
|
||||
|
|
@ -86,14 +89,15 @@ class MockSyncClient:
|
|||
|
||||
def test_async_client_cache_does_not_reuse_clients_between_collected_loops() -> None:
|
||||
cache: Final = LLMClientCache()
|
||||
results: Final = tuple(_cache_result_on_new_loop(cache) for _ in range(50))
|
||||
runs: Final = runs_until_a_loop_address_repeats(lambda: _cache_result_on_new_loop(cache))
|
||||
|
||||
assert all(result is None for result in results)
|
||||
assert all(result is None for _, result in runs)
|
||||
|
||||
|
||||
def test_async_client_cache_keys_differ_for_every_collected_loop() -> None:
|
||||
cache: Final = LLMClientCache()
|
||||
keys: Final = tuple(_cache_key_on_new_loop(cache) for _ in range(50))
|
||||
runs: Final = runs_until_a_loop_address_repeats(lambda: _cache_key_on_new_loop(cache))
|
||||
keys: Final = tuple(key for _, key in runs)
|
||||
|
||||
assert len(set(keys)) == len(keys)
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue