[Perf] Improvements for Async Success Handler (Logging Callbacks) - Approx +130 RPS (#13905)

* [Performance] Reduce Significant CPU overhead from litellm_logging.py (#13895)

* fix: litellm.configured_cold_storage_logger

* fix Session Management - Non-OpenAI Models docs

* ruff fix

* test fix

* create LoggingWorker

* add GLOBAL_LOGGING_WORKER for async task handling

* fix logging tests

* add conftest

* fix conftest

* test fix location of encode bedrock runtime modelid arn

* fix conftest.py

* tuning LoggingWorker

* conftest.py

* fix conftest batches/

* test_async_chat_azure

* event_loop

* test_bedrock_streaming_passthrough_test2

* fix GLOBAL_LOGGING_WORKER

* logging worker

* add flush for global logging worker

* Revert "fix GLOBAL_LOGGING_WORKER"

This reverts commit d254f508f4.

* fix conftest clear_queue

* fix conftest clear_queue

* setup_and_teardown for llm translation

* docs AWS_REGION

* test_async_chat_azure

* change test DIR

* run ci/cd again

* use 1 job for litellm_router_unit_testing

* fix space

* fix litellm_router_unit_testing

* test_aaarouter_dynamic_cooldown_message_retry_time

* litellm_router_unit_testing

* conftest.py clearing qu

* fixes litellm_router_unit_testing

* fixes clear_queue

* fix router_unit_tests

* remove conftest

* add back conftest for router

* fix event loop test

* test fix

* fixes for LoggingWorker

* ruff fix
This commit is contained in:
Ishaan Jaff 2025-08-23 13:13:23 -07:00 • committed by GitHub
parent 9750374081
commit b9132968b2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 549 additions and 17 deletions

View file

@ -439,6 +439,7 @@ jobs:
paths:
- auth_ui_unit_tests_coverage.xml
- auth_ui_unit_tests_coverage
litellm_router_testing: # Runs all tests with the "router" keyword
docker:
- image: cimg/python:3.11
@ -469,7 +470,55 @@ jobs:
command: |
pwd
ls
python -m pytest tests/local_testing tests/router_unit_tests --cov=litellm --cov-report=xml -vv -k "router" -x -v --junitxml=test-results/junit.xml --durations=5
python -m pytest tests/local_testing --cov=litellm --cov-report=xml -vv -k "router" -x -v --junitxml=test-results/junit.xml --durations=5
no_output_timeout: 120m
- run:
name: Rename the coverage files
command: |
mv coverage.xml litellm_router_coverage.xml
mv .coverage litellm_router_coverage
# Store test results
- store_test_results:
path: test-results
- persist_to_workspace:
root: .
paths:
- litellm_router_coverage.xml
- litellm_router_coverage
litellm_router_unit_testing: # Runs all tests with the "router" keyword
docker:
- image: cimg/python:3.11
auth:
username: ${DOCKERHUB_USERNAME}
password: ${DOCKERHUB_PASSWORD}
working_directory: ~/project
steps:
- checkout
- setup_google_dns
- run:
name: Install Dependencies
command: |
python -m pip install --upgrade pip
python -m pip install -r requirements.txt
pip install "pytest==7.3.1"
pip install "respx==0.22.0"
pip install "pytest-cov==5.0.0"
pip install "pytest-retry==1.6.3"
pip install "pytest-asyncio==0.21.1"
pip install semantic_router --no-deps
pip install aurelio_sdk --no-deps
pip install "pytest-xdist==3.6.1"
# Run pytest and generate JUnit XML report
- setup_litellm_enterprise_pip
- run:
name: Run tests
command: |
pwd
ls
python -m pytest -vv tests/router_unit_tests --cov=litellm --cov-report=xml -x -s -v --junitxml=test-results/junit.xml --durations=5
no_output_timeout: 120m
- run:
name: Rename the coverage files
@ -2982,6 +3031,12 @@ workflows:
only:
- main
- /litellm_.*/
- litellm_router_unit_testing:
filters:
branches:
only:
- main
- /litellm_.*/
- check_code_and_doc_quality:
filters:
branches:
@ -3128,6 +3183,7 @@ workflows:
- image_gen_testing
- logging_testing
- litellm_router_testing
- litellm_router_unit_testing
- caching_unit_tests
- litellm_proxy_unit_testing
- litellm_security_tests
@ -3187,6 +3243,7 @@ workflows:
- image_gen_testing
- logging_testing
- litellm_router_testing
- litellm_router_unit_testing
- caching_unit_tests
- langfuse_logging_unit_tests
- litellm_assistants_api_testing

View file

@ -0,0 +1,131 @@
import asyncio
import contextlib
from typing import Coroutine, Optional
from litellm._logging import verbose_logger
class LoggingWorker:
"""
A simple, async logging worker that processes log coroutines in the background.
Designed to be best-effort with bounded queues to prevent backpressure.
This leads to a +200 RPS performance improvement when using LiteLLM Python SDK or Proxy Server.
- Use this to queue coroutine tasks that are not critical to the main flow of the application. e.g Success/Error callbacks, logging, etc.
"""
LOGGING_WORKER_MAX_QUEUE_SIZE = 50_000
LOGGING_WORKER_MAX_TIME_PER_COROUTINE = 20.0
MAX_ITERATIONS_TO_CLEAR_QUEUE = 200
MAX_TIME_TO_CLEAR_QUEUE = 5.0
def __init__(
self,
timeout: float = LOGGING_WORKER_MAX_TIME_PER_COROUTINE,
max_queue_size: int = LOGGING_WORKER_MAX_QUEUE_SIZE,
):
self.timeout = timeout
self.max_queue_size = max_queue_size
self._queue: Optional[asyncio.Queue] = None
self._worker_task: Optional[asyncio.Task] = None
def _ensure_queue(self) -> None:
"""Initialize the queue if it doesn't exist."""
if self._queue is None:
self._queue = asyncio.Queue(maxsize=self.max_queue_size)
def start(self) -> None:
"""Start the logging worker. Idempotent - safe to call multiple times."""
self._ensure_queue()
if self._worker_task is None or self._worker_task.done():
self._worker_task = asyncio.create_task(self._worker_loop())
async def _worker_loop(self) -> None:
"""Main worker loop that processes log coroutines sequentially."""
try:
if self._queue is None:
return
while True:
# Process one coroutine at a time to keep event loop load predictable
coroutine = await self._queue.get()
try:
await asyncio.wait_for(coroutine, timeout=self.timeout)
except Exception as e:
verbose_logger.exception(f"LoggingWorker error: {e}")
pass
finally:
self._queue.task_done()
except asyncio.CancelledError as e:
verbose_logger.exception(f"LoggingWorker cancelled: {e}")
pass
def enqueue(self, coroutine: Coroutine) -> None:
"""
Add a coroutine to the logging queue.
Hot path: never blocks, drops logs if queue is full.
"""
if self._queue is None:
return
try:
self._queue.put_nowait(coroutine)
except asyncio.QueueFull as e:
verbose_logger.exception(f"LoggingWorker queue is full: {e}")
# Drop logs on overload to protect request throughput
pass
def ensure_initialized_and_enqueue(self, async_coroutine: Coroutine):
"""
Ensure the logging worker is initialized and enqueue the coroutine.
"""
self.start()
self.enqueue(async_coroutine)
async def stop(self) -> None:
"""Stop the logging worker and clean up resources."""
if self._worker_task:
self._worker_task.cancel()
with contextlib.suppress(Exception):
await self._worker_task
self._worker_task = None
async def flush(self) -> None:
"""Flush the logging queue."""
if self._queue is None:
return
while not self._queue.empty():
await self._queue.join()
async def clear_queue(self):
"""
Clear the queue with a maximum time limit.
"""
if self._queue is None:
return
start_time = asyncio.get_event_loop().time()
for _ in range(self.MAX_ITERATIONS_TO_CLEAR_QUEUE):
# Check if we've exceeded the maximum time
if asyncio.get_event_loop().time() - start_time >= self.MAX_TIME_TO_CLEAR_QUEUE:
verbose_logger.warning(f"clear_queue exceeded max_time of {self.MAX_TIME_TO_CLEAR_QUEUE}s, stopping early")
break
try:
coroutine = self._queue.get_nowait()
# Await the coroutine to properly execute and avoid "never awaited" warnings
try:
await asyncio.wait_for(coroutine, timeout=self.timeout)
except Exception:
# Suppress errors during cleanup
pass
self._queue.task_done() # If you're using join() elsewhere
except asyncio.QueueEmpty:
break
# Global instance for backward compatibility
GLOBAL_LOGGING_WORKER = LoggingWorker()

View file

@ -833,10 +833,22 @@ async def _client_async_logging_helper(
print_verbose(
f"Async Wrapper: Completed Call, calling async_success_handler: {logging_obj.async_success_handler}"
)
# check if user does not want this to be logged
asyncio.create_task(
logging_obj.async_success_handler(result, start_time, end_time)
################################################
# Async Logging Worker
################################################
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(
async_coroutine = logging_obj.async_success_handler(
result=result,
start_time=start_time,
end_time=end_time
)
)
################################################
# Sync Logging Worker
################################################
logging_obj.handle_sync_success_callbacks_for_async_calls(
result=result,
start_time=start_time,

View file

@ -0,0 +1,22 @@
# conftest.py
import importlib
import os
import sys
import pytest
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the parent directory to the system path
import litellm
import asyncio
@pytest.fixture(scope="session")
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()

View file

@ -0,0 +1,22 @@
import importlib
import os
import sys
import asyncio
import pytest
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the parent directory to the system path
import litellm
import asyncio
@pytest.fixture(scope="session")
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()

View file

@ -11,6 +11,17 @@ sys.path.insert(
) # Adds the parent directory to the system path
import litellm
import asyncio
@pytest.fixture(scope="session")
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="function", autouse=True)
def setup_and_teardown():

View file

@ -11,11 +11,14 @@ sys.path.insert(
) # Adds the parent directory to the system path
import litellm
import asyncio
@pytest.fixture(scope="session")
def event_loop():
"""Create an instance of the default event loop for each test session."""
loop = asyncio.get_event_loop_policy().new_event_loop()
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()
@ -28,6 +31,10 @@ def setup_and_teardown(event_loop): # Add event_loop as a dependency
import litellm
from litellm import Router
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
# flush all logs
asyncio.run(GLOBAL_LOGGING_WORKER.clear_queue())
importlib.reload(litellm)
# Set the event loop from the fixture

View file

@ -3308,7 +3308,7 @@ async def test_bedrock_converse__streaming_passthrough(monkeypatch):
@pytest.mark.asyncio
async def test_bedrock_streaming_passthrough(monkeypatch):
async def test_bedrock_streaming_passthrough_test2(monkeypatch):
import litellm
import time
import asyncio
@ -3350,7 +3350,7 @@ async def test_bedrock_streaming_passthrough(monkeypatch):
async for chunk in response:
print(chunk)
await asyncio.sleep(1)
await asyncio.sleep(5)
mock_callback.assert_called_once()
# check standard logging payload created
@ -3360,7 +3360,7 @@ async def test_bedrock_streaming_passthrough(monkeypatch):
@pytest.mark.asyncio
async def test_bedrock_streaming_passthrough(monkeypatch):
async def test_bedrock_streaming_passthrough_test1(monkeypatch):
import litellm
import time
import asyncio
@ -3402,7 +3402,7 @@ async def test_bedrock_streaming_passthrough(monkeypatch):
async for chunk in response:
print(chunk)
await asyncio.sleep(1)
await asyncio.sleep(5)
mock_callback.assert_called_once()
# check standard logging payload created

View file

@ -11,6 +11,19 @@ sys.path.insert(
) # Adds the parent directory to the system path
import litellm
import asyncio
@pytest.fixture(scope="session")
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="function", autouse=True)
def setup_and_teardown():
@ -24,6 +37,12 @@ def setup_and_teardown():
import litellm
from litellm import Router
import asyncio
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
# flush all logs
asyncio.run(GLOBAL_LOGGING_WORKER.clear_queue())
importlib.reload(litellm)

View file

@ -2328,7 +2328,7 @@ async def test_aaarouter_dynamic_cooldown_message_retry_time(sync_mode):
except litellm.RateLimitError:
pass
await asyncio.sleep(2)
await asyncio.sleep(5)
if sync_mode:
cooldown_deployments = _get_cooldown_deployments(

View file

@ -10,7 +10,16 @@ sys.path.insert(
0, os.path.abspath("../..")
) # Adds the parent directory to the system path
import litellm
import asyncio
@pytest.fixture(scope="session")
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="function", autouse=True)
def setup_and_teardown():
@ -24,8 +33,23 @@ def setup_and_teardown():
import litellm
from litellm import Router
import asyncio
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
# flush all logs
asyncio.run(GLOBAL_LOGGING_WORKER.clear_queue())
importlib.reload(litellm)
try:
if hasattr(litellm, "proxy") and hasattr(litellm.proxy, "proxy_server"):
import litellm.proxy.proxy_server
importlib.reload(litellm.proxy.proxy_server)
except Exception as e:
print(f"Error reloading litellm.proxy.proxy_server: {e}")
import asyncio
loop = asyncio.get_event_loop_policy().new_event_loop()
@ -39,6 +63,7 @@ def setup_and_teardown():
asyncio.set_event_loop(None) # Remove the reference to the loop
def pytest_collection_modifyitems(config, items):
# Separate tests in 'test_amazing_proxy_custom_logger.py' and other tests
custom_logger_tests = [

View file

@ -34,7 +34,7 @@ from litellm.integrations.custom_logger import CustomLogger
## 5. Azure OpenAI acompletion + streaming call with fallbacks
## 6. Azure OpenAI aembedding call with fallbacks
# Test interfaces
## Test interfaces
## 1. router.completion() + router.embeddings()
## 2. proxy.completions + proxy.embeddings
@ -265,8 +265,9 @@ class CompletionCustomHandler(
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
try:
print("CompletionCustomHandler.async_log_success_event, kwargs: ", kwargs)
self.states.append("async_success")
print("in async success, kwargs: ", kwargs)
print("############### CompletionCustomHandler async success, kwargs: ", kwargs)
## START TIME
assert isinstance(start_time, datetime)
## END TIME
@ -409,7 +410,8 @@ async def test_async_chat_azure():
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Hi 👋 - i'm openai"}],
)
await asyncio.sleep(2)
print("got response, sleeping 5 seconds....")
await asyncio.sleep(5)
assert len(customHandler_completion_azure_router.errors) == 0
assert (
len(customHandler_completion_azure_router.states) == 3
@ -427,7 +429,7 @@ async def test_async_chat_azure():
async for chunk in response:
print(f"async azure router chunk: {chunk}")
continue
await asyncio.sleep(2)
await asyncio.sleep(5)
print(f"customHandler.states: {customHandler_streaming_azure_router.states}")
assert len(customHandler_streaming_azure_router.errors) == 0
assert (
@ -459,7 +461,7 @@ async def test_async_chat_azure():
print(f"response in router3 acompletion: {response}")
except Exception:
pass
await asyncio.sleep(1)
await asyncio.sleep(5)
print(f"customHandler.states: {customHandler_failure.states}")
assert len(customHandler_failure.errors) == 0
assert len(customHandler_failure.states) == 3 # pre, post, failure

View file

@ -0,0 +1,64 @@
# conftest.py
import importlib
import os
import sys
import pytest
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the parent directory to the system path
import litellm
import asyncio
@pytest.fixture(scope="session")
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="function", autouse=True)
def setup_and_teardown():
"""
This fixture reloads litellm before every function. To speed up testing by removing callbacks being chained.
"""
curr_dir = os.getcwd() # Get the current working directory
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the project directory to the system path
import litellm
from litellm import Router
importlib.reload(litellm)
import asyncio
loop = asyncio.get_event_loop_policy().new_event_loop()
asyncio.set_event_loop(loop)
print(litellm)
# from litellm import Router, completion, aembedding, acompletion, embedding
yield
# Teardown code (executes after the yield point)
loop.close() # Close the loop created earlier
asyncio.set_event_loop(None) # Remove the reference to the loop
def pytest_collection_modifyitems(config, items):
# Separate tests in 'test_amazing_proxy_custom_logger.py' and other tests
custom_logger_tests = [
item for item in items if "custom_logger" in item.parent.name
]
other_tests = [item for item in items if "custom_logger" not in item.parent.name]
# Sort tests based on their names
custom_logger_tests.sort(key=lambda x: x.name)
other_tests.sort(key=lambda x: x.name)
# Reorder the items list
items[:] = custom_logger_tests + other_tests

View file

@ -0,0 +1,82 @@
# conftest.py
import importlib
import os
import sys
import pytest
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the parent directory to the system path
import litellm
import asyncio
@pytest.fixture(scope="session")
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="function", autouse=True)
def setup_and_teardown():
"""
This fixture reloads litellm before every function. To speed up testing by removing callbacks being chained.
"""
curr_dir = os.getcwd() # Get the current working directory
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the project directory to the system path
import litellm
from litellm import Router
import asyncio
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
# flush all logs
asyncio.run(GLOBAL_LOGGING_WORKER.clear_queue())
importlib.reload(litellm)
try:
if hasattr(litellm, "proxy") and hasattr(litellm.proxy, "proxy_server"):
import litellm.proxy.proxy_server
importlib.reload(litellm.proxy.proxy_server)
except Exception as e:
print(f"Error reloading litellm.proxy.proxy_server: {e}")
import asyncio
loop = asyncio.get_event_loop_policy().new_event_loop()
asyncio.set_event_loop(loop)
print(litellm)
# from litellm import Router, completion, aembedding, acompletion, embedding
yield
# Teardown code (executes after the yield point)
loop.close() # Close the loop created earlier
asyncio.set_event_loop(None) # Remove the reference to the loop
def pytest_collection_modifyitems(config, items):
# Separate tests in 'test_amazing_proxy_custom_logger.py' and other tests
custom_logger_tests = [
item for item in items if "custom_logger" in item.parent.name
]
other_tests = [item for item in items if "custom_logger" not in item.parent.name]
# Sort tests based on their names
custom_logger_tests.sort(key=lambda x: x.name)
other_tests.sort(key=lambda x: x.name)
# Reorder the items list
items[:] = custom_logger_tests + other_tests

View file

@ -14,7 +14,7 @@ sys.path.insert(
from litellm import Router, CustomLogger
from litellm.types.utils import StandardLoggingPayload
# Get the current directory of the file being run
## Get the current directory of the file being run
pwd = os.path.dirname(os.path.realpath(__file__))
print(pwd)

View file

@ -9,9 +9,23 @@ import pytest
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the parent directory to the system path
import asyncio
import litellm
@pytest.fixture(scope="session")
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="function", autouse=True)
def setup_and_teardown():
"""

View file

@ -0,0 +1,64 @@
# conftest.py
import importlib
import os
import sys
import pytest
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the parent directory to the system path
import litellm
import asyncio
@pytest.fixture(scope="session")
def event_loop():
try:
loop = asyncio.get_running_loop()
except RuntimeError:
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="function", autouse=True)
def setup_and_teardown():
"""
This fixture reloads litellm before every function. To speed up testing by removing callbacks being chained.
"""
curr_dir = os.getcwd() # Get the current working directory
sys.path.insert(
0, os.path.abspath("../..")
) # Adds the project directory to the system path
import litellm
from litellm import Router
importlib.reload(litellm)
import asyncio
loop = asyncio.get_event_loop_policy().new_event_loop()
asyncio.set_event_loop(loop)
print(litellm)
# from litellm import Router, completion, aembedding, acompletion, embedding
yield
# Teardown code (executes after the yield point)
loop.close() # Close the loop created earlier
asyncio.set_event_loop(None) # Remove the reference to the loop
def pytest_collection_modifyitems(config, items):
# Separate tests in 'test_amazing_proxy_custom_logger.py' and other tests
custom_logger_tests = [
item for item in items if "custom_logger" in item.parent.name
]
other_tests = [item for item in items if "custom_logger" not in item.parent.name]
# Sort tests based on their names
custom_logger_tests.sort(key=lambda x: x.name)
other_tests.sort(key=lambda x: x.name)
# Reorder the items list
items[:] = custom_logger_tests + other_tests