From 446fadc4c71ea4cc95ab8e311bfdab1c3ace7bfe Mon Sep 17 00:00:00 2001 From: yassin Date: Thu, 17 Sep 2026 02:50:45 +0000 Subject: [PATCH 1/5] feat(router): bound the max_parallel_requests wait queue and return 429 on overflow Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/constants.py | 2 + litellm/proxy/proxy_server.py | 16 +- litellm/router.py | 452 +++++------------- .../client_initalization_utils.py | 84 +++- .../router_settings_endpoints.py | 11 + litellm/types/router.py | 2 + litellm/types/utils.py | 1 + .../router_code_coverage.py | 1 + tests/test_litellm/proxy/test_proxy_server.py | 59 +++ .../test_client_initalization_utils.py | 189 ++++++++ tests/test_litellm/test_router.py | 221 +++++++++ tests/test_litellm/test_utils.py | 30 ++ .../components/router_settings/index.test.tsx | 35 ++ .../src/components/router_settings/index.tsx | 11 +- 14 files changed, 782 insertions(+), 332 deletions(-) create mode 100644 tests/test_litellm/router_utils/test_client_initalization_utils.py diff --git a/litellm/constants.py b/litellm/constants.py index 8409a161800..0cd59706015 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -30,8 +30,10 @@ RUNTIME_UPDATABLE_ROUTER_SETTINGS: Final[frozenset[str]] = frozenset( "enable_tag_filtering", "tag_routing_prefix", "optional_pre_call_checks", + "default_max_parallel_requests_queue_size", } ) +NULLABLE_RUNTIME_ROUTER_SETTINGS: Final[frozenset[str]] = frozenset({"default_max_parallel_requests_queue_size"}) ROUTER_SETTINGS_MANAGED_OUTSIDE_CONFIG: Final[frozenset[str]] = frozenset( { "model_list", diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 7bc36e175c0..dc78ec75a6e 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -70,6 +70,7 @@ from litellm.constants import ( LITELLM_SETTINGS_SAFE_DB_OVERRIDES, LITELLM_UI_ALLOW_HEADERS, LITELLM_UI_SESSION_DURATION, + NULLABLE_RUNTIME_ROUTER_SETTINGS, RUNTIME_UPDATABLE_ROUTER_SETTINGS, ) from litellm.litellm_core_utils.asyncify import asyncify @@ -6900,13 +6901,20 @@ class ProxyConfig: ): from litellm.utils import _update_dictionary + db_settings: Final = db_router_settings.param_value db_overlay_deferring_empty_lists_to_config: Final = { k: v - for k, v in db_router_settings.param_value.items() + for k, v in db_settings.items() if not (k in config_router_settings and isinstance(v, list) and len(v) == 0) } - combined_router_settings = _update_dictionary( - config_router_settings, db_overlay_deferring_empty_lists_to_config + cleared_nullable_settings: Final = MappingProxyType( + {k: None for k in NULLABLE_RUNTIME_ROUTER_SETTINGS if k in db_settings and db_settings[k] is None} + ) + combined_router_settings = MappingProxyType( + { + **_update_dictionary(config_router_settings, db_overlay_deferring_empty_lists_to_config), + **cleared_nullable_settings, + } ) elif config_router_settings is not None and isinstance(config_router_settings, dict): combined_router_settings = config_router_settings @@ -17039,7 +17047,7 @@ async def update_config( raw_router_settings_without_none: Final = { key: value for key, value in raw_router_settings.items() - if key not in typed_router_settings and value is not None + if key not in typed_router_settings and (value is not None or key in NULLABLE_RUNTIME_ROUTER_SETTINGS) } router_settings_updates: Final = {**typed_router_settings, **raw_router_settings_without_none} new_router_settings: Final = {**existing, **router_settings_updates} diff --git a/litellm/router.py b/litellm/router.py index 5f5522e9fd4..b62c83b8ab1 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -738,6 +738,7 @@ class Router: stream_timeout: float | None = None, default_litellm_params: dict | None = None, # default params for Router.chat.completion.create default_max_parallel_requests: int | None = None, + default_max_parallel_requests_queue_size: int | None = None, set_verbose: bool = False, debug_level: Literal["DEBUG", "INFO"] = "INFO", default_fallbacks: list[str] | None = None, # generic fallbacks, works across all deployments @@ -935,6 +936,7 @@ class Router: None # use this to track the users default deployment, when they want to use model = * ) self.default_max_parallel_requests = default_max_parallel_requests + self._default_max_parallel_requests_queue_size = default_max_parallel_requests_queue_size self.provider_default_deployment_ids: list[str] = [] self.pattern_router = PatternMatchRouter() self.team_pattern_routers: dict[str, PatternMatchRouter] = {} # {"TEAM_ID": PatternMatchRouter} @@ -3630,8 +3632,6 @@ class Router: input_kwargs.pop("silent_model", None) input_kwargs.pop("include_fallback_errors", None) - _response: Final = litellm.acompletion(**input_kwargs) - logging_obj: Final[LiteLLMLogging | None] = kwargs.get("litellm_logging_obj", None) rpm_semaphore: Final = self._get_client( @@ -3647,7 +3647,7 @@ class Router: logging_obj=logging_obj, parent_otel_span=parent_otel_span, ) - response = await _response + response = await litellm.acompletion(**input_kwargs) ## CHECK CONTENT FILTER ERROR ## if isinstance(response, ModelResponse): @@ -4574,38 +4574,16 @@ class Router: ) self.total_calls[model_name] += 1 - response = litellm.aimage_generation( - **{ - **data, - "prompt": prompt, - "caching": self.cache_responses, - "client": model_client, - **kwargs, - } - ) - - ### CONCURRENCY-SAFE RPM CHECKS ### - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - """ - - Check rpm limits before making the call - - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe) - """ - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span + async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span): + response = await litellm.aimage_generation( + **{ + **data, + "prompt": prompt, + "caching": self.cache_responses, + "client": model_client, + **kwargs, + } ) - response = await response self.success_calls[model_name] += 1 verbose_router_logger.info("litellm.aimage_generation(model=%s)\x1b[32m 200 OK\x1b[0m", model_name) @@ -4679,38 +4657,16 @@ class Router: ) self.total_calls[model_name] += 1 - response = litellm.atranscription( - **{ - **data, - "file": file, - "caching": self.cache_responses, - "client": model_client, - **kwargs, - } - ) - - ### CONCURRENCY-SAFE RPM CHECKS ### - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - """ - - Check rpm limits before making the call - - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe) - """ - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span + async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span): + response = await litellm.atranscription( + **{ + **data, + "file": file, + "caching": self.cache_responses, + "client": model_client, + **kwargs, + } ) - response = await response self.success_calls[model_name] += 1 verbose_router_logger.info("litellm.atranscription(model=%s)\x1b[32m 200 OK\x1b[0m", model_name) @@ -4794,38 +4750,16 @@ class Router: ) self.total_calls[model_name] += 1 - response = litellm.aspeech( - **{ - **data, - "input": input, - "voice": data.get("voice") if voice is None else voice, - "client": model_client, - **kwargs, - } - ) - - ### CONCURRENCY-SAFE RPM CHECKS ### - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - """ - - Check rpm limits before making the call - - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe) - """ - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span + async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span): + response = await litellm.aspeech( + **{ + **data, + "input": input, + "voice": data.get("voice") if voice is None else voice, + "client": model_client, + **kwargs, + } ) - response = await response self.success_calls[model_name] += 1 verbose_router_logger.info("litellm.aspeech(model=%s)\x1b[32m 200 OK\x1b[0m", model_name) @@ -4990,37 +4924,16 @@ class Router: ) self.total_calls[model_name] += 1 - response = litellm.atext_completion( - **{ - **data, - "prompt": prompt, - "caching": self.cache_responses, - "client": model_client, - **kwargs, - } - ) - - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - """ - - Check rpm limits before making the call - - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe) - """ - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span + async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span): + response = await litellm.atext_completion( + **{ + **data, + "prompt": prompt, + "caching": self.cache_responses, + "client": model_client, + **kwargs, + } ) - response = await response self.success_calls[model_name] += 1 verbose_router_logger.info("litellm.atext_completion(model=%s)\x1b[32m 200 OK\x1b[0m", model_name) @@ -5081,37 +4994,16 @@ class Router: ) self.total_calls[model_name] += 1 - response = litellm.aadapter_completion( - **{ - **data, - "adapter_id": adapter_id, - "caching": self.cache_responses, - "client": model_client, - **kwargs, - } - ) - - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - """ - - Check rpm limits before making the call - - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe) - """ - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span + async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span): + response = await litellm.aadapter_completion( + **{ + **data, + "adapter_id": adapter_id, + "caching": self.cache_responses, + "client": model_client, + **kwargs, + } ) - response = await response self.success_calls[model_name] += 1 verbose_router_logger.info("litellm.aadapter_completion(model=%s)\x1b[32m 200 OK\x1b[0m", model_name) @@ -5341,29 +5233,8 @@ class Router: if custom_llm_provider is not None: response_kwargs["custom_llm_provider"] = custom_llm_provider - response = original_generic_function(**response_kwargs) - - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - """ - - Check rpm limits before making the call - - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe) - """ - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response + async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span): + response = await original_generic_function(**response_kwargs) if self._should_raise_anthropic_refusal_error( model=model, @@ -5971,38 +5842,16 @@ class Router: ) self.total_calls[model_name] += 1 - response = litellm.aembedding( - **{ - **data, - "input": input, - "caching": self.cache_responses, - "client": model_client, - **kwargs, - } - ) - - ### CONCURRENCY-SAFE RPM CHECKS ### - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - """ - - Check rpm limits before making the call - - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe) - """ - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span + async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span): + response = await litellm.aembedding( + **{ + **data, + "input": input, + "caching": self.cache_responses, + "client": model_client, + **kwargs, + } ) - response = await response self.success_calls[model_name] += 1 verbose_router_logger.info("litellm.aembedding(model=%s)\x1b[32m 200 OK\x1b[0m", model_name) @@ -6111,37 +5960,18 @@ class Router: "gcs_bucket_name" in data ): # TODO: Remove this once we have a better way to handle GCS bucket name: Problem is that we need to pass the gcs_bucket_name to the router for the create_file call but it doesn't show up there kwargs_copy.setdefault("litellm_metadata", {})["gcs_bucket_name"] = data["gcs_bucket_name"] - response = litellm.acreate_file( - **{ - **data, - "custom_llm_provider": custom_llm_provider, - "caching": self.cache_responses, - "client": model_client, - **kwargs_copy, - } - ) - - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs_copy, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - """ - - Check rpm limits before making the call - - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe) - """ - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span + async with self._deployment_slot( + deployment=deployment, kwargs=kwargs_copy, parent_otel_span=parent_otel_span + ): + response = await litellm.acreate_file( + **{ + **data, + "custom_llm_provider": custom_llm_provider, + "caching": self.cache_responses, + "client": model_client, + **kwargs_copy, + } ) - response = await response self.success_calls[model_name] += 1 verbose_router_logger.info("litellm.acreate_file(model=%s)\x1b[32m 200 OK\x1b[0m", model_name) @@ -6231,33 +6061,16 @@ class Router: ) custom_llm_provider = custom_llm_provider or inferred_custom_llm_provider - response = avector_store_create_sdk( - **{ - **data, - "custom_llm_provider": custom_llm_provider, - "caching": self.cache_responses, - "client": model_client, - **kwargs, - } - ) - - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span + async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span): + response = await avector_store_create_sdk( + **{ + **data, + "custom_llm_provider": custom_llm_provider, + "caching": self.cache_responses, + "client": model_client, + **kwargs, + } ) - response = await response self.success_calls[model_name] += 1 verbose_router_logger.info("litellm.avector_store_create(model=%s)\x1b[32m 200 OK\x1b[0m", model_name) @@ -6343,37 +6156,16 @@ class Router: ) custom_llm_provider = custom_llm_provider or inferred_custom_llm_provider - response = litellm.acreate_batch( - **{ - **data, - "custom_llm_provider": custom_llm_provider, - "caching": self.cache_responses, - "client": model_client, - **kwargs, - } - ) - - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - """ - - Check rpm limits before making the call - - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe) - """ - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span + async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span): + response = await litellm.acreate_batch( + **{ + **data, + "custom_llm_provider": custom_llm_provider, + "caching": self.cache_responses, + "client": model_client, + **kwargs, + } ) - response = await response self.success_calls[model_name] += 1 verbose_router_logger.info("litellm.acreate_batch(model=%s)\x1b[32m 200 OK\x1b[0m", model_name) @@ -6564,37 +6356,16 @@ class Router: ) custom_llm_provider = custom_llm_provider or inferred_custom_llm_provider - response = litellm.acancel_batch( - **{ - **data, - "custom_llm_provider": custom_llm_provider, - "caching": self.cache_responses, - "client": model_client, - **kwargs, - } - ) - - rpm_semaphore: Final = self._get_client( - deployment=deployment, - kwargs=kwargs, - client_type="max_parallel_requests", - ) - - if rpm_semaphore is not None and isinstance(rpm_semaphore, asyncio.Semaphore): - async with rpm_semaphore: - """ - - Check rpm limits before making the call - - If allowed, increment the rpm limit (allows global value to be updated, concurrency-safe) - """ - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span - ) - response = await response - else: - await self.async_routing_strategy_pre_call_checks( - deployment=deployment, parent_otel_span=parent_otel_span + async with self._deployment_slot(deployment=deployment, kwargs=kwargs, parent_otel_span=parent_otel_span): + response = await litellm.acancel_batch( + **{ + **data, + "custom_llm_provider": custom_llm_provider, + "caching": self.cache_responses, + "client": model_client, + **kwargs, + } ) - response = await response self.success_calls[model_name] += 1 verbose_router_logger.info("litellm.acancel_batch(model=%s)\x1b[32m 200 OK\x1b[0m", model_name) @@ -8729,6 +8500,23 @@ class Router: ) raise e + @contextlib.asynccontextmanager + async def _deployment_slot( + self, deployment: dict, kwargs: Mapping[str, object], parent_otel_span: Span | None + ) -> AsyncGenerator[None, None]: + """Holds the deployment's max_parallel_requests slot, if it has one, around the provider call. Routing + strategy pre-call checks run inside the slot so their rpm accounting stays concurrency-safe.""" + rpm_semaphore: Final = self._get_client( + deployment=deployment, + kwargs=kwargs, + client_type="max_parallel_requests", + ) + async with contextlib.AsyncExitStack() as slot: + if isinstance(rpm_semaphore, asyncio.Semaphore): + await slot.enter_async_context(rpm_semaphore) + await self.async_routing_strategy_pre_call_checks(deployment=deployment, parent_otel_span=parent_otel_span) + yield + async def async_callback_filter_deployments( self, model: str, @@ -12055,8 +11843,20 @@ class Router: _settings_to_return[var] = self.lowestlatency_logger.routing_args.json() _settings_to_return["routing_groups"] = [group.model_dump() for group in self._routing_groups.values()] + _settings_to_return["default_max_parallel_requests_queue_size"] = self.default_max_parallel_requests_queue_size return _settings_to_return + @property + def default_max_parallel_requests_queue_size(self) -> int | None: + return self._default_max_parallel_requests_queue_size + + @default_max_parallel_requests_queue_size.setter + def default_max_parallel_requests_queue_size(self, queue_size: int | None) -> None: + self._default_max_parallel_requests_queue_size = None if queue_size is None else int(queue_size) + InitalizeCachedClient.apply_default_max_parallel_requests_queue_size( + litellm_router_instance=self, queue_size=self._default_max_parallel_requests_queue_size + ) + def update_settings(self, **kwargs): """ Update the router settings. diff --git a/litellm/router_utils/client_initalization_utils.py b/litellm/router_utils/client_initalization_utils.py index 24324334a86..a135978d09e 100644 --- a/litellm/router_utils/client_initalization_utils.py +++ b/litellm/router_utils/client_initalization_utils.py @@ -1,6 +1,10 @@ import asyncio +import time from typing import TYPE_CHECKING, Any, Final +from litellm._logging import verbose_router_logger +from litellm.exceptions import RateLimitError, RateLimitErrorCategory, RateLimitType +from litellm.types.router import RouterErrors from litellm.utils import calculate_max_parallel_requests if TYPE_CHECKING: @@ -11,6 +15,59 @@ else: LitellmRouter = Any +class DeploymentSemaphore(asyncio.Semaphore): + """A deployment's max_parallel_requests slots. ``queue_size=None`` parks callers without bound, like a plain + ``asyncio.Semaphore``; otherwise a caller arriving while all slots are busy and ``queue_size`` callers already + wait gets a 429 instead of being parked.""" + + def __init__(self, max_parallel_requests: int, model_id: str, model_group: str, queue_size: int | None) -> None: + super().__init__(max_parallel_requests) + self.max_parallel_requests = max_parallel_requests + self.model_id = model_id + self.model_group = model_group + self.queue_size = queue_size + self.waiting = 0 + + async def acquire(self) -> bool: + if not self.locked(): + return await super().acquire() + if self.queue_size is not None and self.waiting >= self.queue_size: + raise RateLimitError( + message=( + f"{RouterErrors.max_parallel_requests_queue_full.value} Deployment model_group={self.model_group}, " + f"id={self.model_id} has all max_parallel_requests={self.max_parallel_requests} slots in use and " + f"{self.waiting} requests already waiting, which is its max_parallel_requests_queue_size=" + f"{self.queue_size}. Raise max_parallel_requests or max_parallel_requests_queue_size for this " + "deployment, or unset max_parallel_requests_queue_size to queue without a bound" + ), + llm_provider="", + model=self.model_group, + category=RateLimitErrorCategory.LITELLM_RATE_LIMIT, + rate_limit_type=RateLimitType.CONCURRENT_REQUESTS, + ) + self.waiting += 1 + queued_at: Final = time.perf_counter() + verbose_router_logger.debug( + "Deployment model_group=%s, id=%s has all max_parallel_requests=%s slots in use, request queued " + "(waiting=%s, max_parallel_requests_queue_size=%s)", + self.model_group, + self.model_id, + self.max_parallel_requests, + self.waiting, + self.queue_size, + ) + try: + return await super().acquire() + finally: + self.waiting -= 1 + verbose_router_logger.debug( + "Deployment model_group=%s, id=%s request left the max_parallel_requests queue after %.1f ms", + self.model_group, + self.model_id, + (time.perf_counter() - queued_at) * 1000, + ) + + class InitalizeCachedClient: @staticmethod def set_max_parallel_requests_client(litellm_router_instance: LitellmRouter, model: dict): @@ -26,10 +83,35 @@ class InitalizeCachedClient: default_max_parallel_requests=litellm_router_instance.default_max_parallel_requests, ) if calculated_max_parallel_requests: - semaphore: Final = asyncio.Semaphore(calculated_max_parallel_requests) + deployment_queue_size: Final = litellm_params.get("max_parallel_requests_queue_size", None) + semaphore: Final = DeploymentSemaphore( + max_parallel_requests=calculated_max_parallel_requests, + model_id=model_id, + model_group=model.get("model_name", ""), + queue_size=( + deployment_queue_size + if deployment_queue_size is not None + else litellm_router_instance.default_max_parallel_requests_queue_size + ), + ) cache_key: Final = f"{model_id}_max_parallel_requests_client" litellm_router_instance.cache.set_cache( key=cache_key, value=semaphore, local_only=True, ) + + @staticmethod + def apply_default_max_parallel_requests_queue_size( + litellm_router_instance: LitellmRouter, queue_size: int | None + ) -> None: + inheriting_semaphores: Final = ( + litellm_router_instance.cache.get_cache( + key=f"{model['model_info']['id']}_max_parallel_requests_client", local_only=True + ) + for model in litellm_router_instance.model_list + if model["litellm_params"].get("max_parallel_requests_queue_size") is None + ) + for semaphore in inheriting_semaphores: + if isinstance(semaphore, DeploymentSemaphore): + semaphore.queue_size = queue_size diff --git a/litellm/types/management_endpoints/router_settings_endpoints.py b/litellm/types/management_endpoints/router_settings_endpoints.py index cef180b202a..fe715e45b2f 100644 --- a/litellm/types/management_endpoints/router_settings_endpoints.py +++ b/litellm/types/management_endpoints/router_settings_endpoints.py @@ -244,6 +244,17 @@ ROUTER_SETTINGS_FIELDS: Final[list[RouterSettingsField]] = [ field_default=None, ui_field_name="Max Parallel Requests", ), + RouterSettingsField( + field_name="default_max_parallel_requests_queue_size", + field_type="Integer", + field_value=None, + field_description=( + "Default cap on how many requests may wait for a deployment's max_parallel_requests slot before " + "further requests get a 429. Unset queues without a bound" + ), + field_default=None, + ui_field_name="Max Parallel Requests Queue Size", + ), RouterSettingsField( field_name="enable_tag_filtering", field_type="Boolean", diff --git a/litellm/types/router.py b/litellm/types/router.py index 584d2494db4..8f788b5f933 100644 --- a/litellm/types/router.py +++ b/litellm/types/router.py @@ -497,6 +497,7 @@ class LiteLLMParamsTypedDict(TypedDict, total=False): order: int | None weight: int | None max_parallel_requests: int | None + max_parallel_requests_queue_size: ReadOnly[int | None] api_key: str | None api_base: str | None api_version: str | None @@ -647,6 +648,7 @@ class RouterErrors(enum.Enum): """ user_defined_ratelimit_error = "Deployment over user-defined ratelimit." + max_parallel_requests_queue_full = "Deployment max_parallel_requests queue is full." no_deployments_available = "No deployments available for selected model" all_deployments_in_cooldown = "All deployments for selected model are in cooldown" no_deployments_with_tag_routing = "Not allowed to access model due to tags configuration" diff --git a/litellm/types/utils.py b/litellm/types/utils.py index aaa16fd2d44..8f902f34548 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -3841,6 +3841,7 @@ all_litellm_params = ( "itpm", "otpm", "max_parallel_requests", + "max_parallel_requests_queue_size", "input_cost_per_token", "output_cost_per_token", "input_cost_per_second", diff --git a/tests/code_coverage_tests/router_code_coverage.py b/tests/code_coverage_tests/router_code_coverage.py index a11f015743b..057e82a24c8 100644 --- a/tests/code_coverage_tests/router_code_coverage.py +++ b/tests/code_coverage_tests/router_code_coverage.py @@ -88,6 +88,7 @@ ignored_function_names = [ "_resolve_claude_code_session_router", # Tested through Claude Code session routing in test_router.py "_get_claude_code_session_router_binding", # Tested through the two-worker session routing test in test_router.py "_apply_updated_routing_strategy_args", # Tested via update_settings in test_lowest_latency.py (file lacks "router" in name) + "default_max_parallel_requests_queue_size", # Property, so its reads and assignments in test_router.py are never an ast.Call ] diff --git a/tests/test_litellm/proxy/test_proxy_server.py b/tests/test_litellm/proxy/test_proxy_server.py index 41c4956dba6..5754301ac4a 100644 --- a/tests/test_litellm/proxy/test_proxy_server.py +++ b/tests/test_litellm/proxy/test_proxy_server.py @@ -5051,6 +5051,39 @@ async def test_add_router_settings_from_db_config_empty_db_list_still_clears_unc assert combined_settings["num_retries"] == 1 +@pytest.mark.asyncio +async def test_add_router_settings_from_db_config_null_queue_size_reaches_router(): + """A cleared Admin UI field is stored as null. The reload must hand that None to the + router so a config.yaml bound is lifted, while an unrelated null still falls back to + the config value.""" + from unittest.mock import AsyncMock, MagicMock + + from litellm.proxy.proxy_server import ProxyConfig + + proxy_config = ProxyConfig() + mock_router = MagicMock() + mock_router.update_settings = MagicMock() + + config_data = {"router_settings": {"default_max_parallel_requests_queue_size": 2, "num_retries": 1}} + + mock_db_config = MagicMock() + mock_db_config.param_value = {"default_max_parallel_requests_queue_size": None, "num_retries": None} + + mock_prisma_client = MagicMock() + mock_prisma_client.db.litellm_config.find_first = AsyncMock(return_value=mock_db_config) + + await proxy_config._add_router_settings_from_db_config( + config_data=config_data, + llm_router=mock_router, + prisma_client=mock_prisma_client, + ) + + combined_settings = mock_router.update_settings.call_args.kwargs + assert "default_max_parallel_requests_queue_size" in combined_settings + assert combined_settings["default_max_parallel_requests_queue_size"] is None + assert combined_settings["num_retries"] == 1 + + @pytest.mark.asyncio async def test_add_router_settings_from_db_config_edge_cases(): """ @@ -9334,6 +9367,32 @@ def test_update_config_litellm_settings_request_wins_for_non_callback_keys( restore() +def test_update_config_router_settings_null_clears_max_parallel_requests_queue_size( + _update_config_setup, +): + """Clearing the Admin UI field sends null. The stored row must hold null so the + reload hands None to the router and queueing becomes unbounded again, while an + unrelated null is still dropped rather than persisted.""" + client, prisma, restore = _update_config_setup( + initial_rows={ + "router_settings": {"default_max_parallel_requests_queue_size": 3, "num_retries": 2}, + } + ) + try: + resp = client.post( + "/config/update", + json={"router_settings": {"default_max_parallel_requests_queue_size": None, "timeout": None}}, + ) + assert resp.status_code == 200 + stored = prisma.db.litellm_config.rows["router_settings"] + assert "default_max_parallel_requests_queue_size" in stored + assert stored["default_max_parallel_requests_queue_size"] is None + assert stored["num_retries"] == 2 + assert "timeout" not in stored + finally: + restore() + + def test_update_config_success_callback_normalizes_existing_mixed_case( _update_config_setup, ): diff --git a/tests/test_litellm/router_utils/test_client_initalization_utils.py b/tests/test_litellm/router_utils/test_client_initalization_utils.py new file mode 100644 index 00000000000..332f2f1503a --- /dev/null +++ b/tests/test_litellm/router_utils/test_client_initalization_utils.py @@ -0,0 +1,189 @@ +import asyncio +from typing import Final + +import pytest + +import litellm +from litellm import Router +from litellm.router_utils.client_initalization_utils import DeploymentSemaphore + + +def _semaphore(queue_size: int | None, max_parallel_requests: int = 1) -> DeploymentSemaphore: + return DeploymentSemaphore( + max_parallel_requests=max_parallel_requests, + model_id="deployment-1", + model_group="gpt-5.6", + queue_size=queue_size, + ) + + +async def _hold(semaphore: DeploymentSemaphore, release: asyncio.Event) -> str: + async with semaphore: + await release.wait() + return "ok" + + +async def _expect_rejection(semaphore: DeploymentSemaphore) -> litellm.RateLimitError: + with pytest.raises(litellm.RateLimitError) as excinfo: + await asyncio.wait_for(semaphore.acquire(), timeout=1) + return excinfo.value + + +@pytest.mark.asyncio +async def test_queue_full_rejects_new_caller_while_queued_callers_still_complete(): + semaphore: Final = _semaphore(queue_size=2) + release: Final = asyncio.Event() + holder: Final = asyncio.create_task(_hold(semaphore, release)) + await asyncio.sleep(0) + queued: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(2)] + await asyncio.sleep(0) + assert semaphore.locked() and semaphore.waiting == 2 + + rejection: Final = await _expect_rejection(semaphore) + + assert rejection.status_code == 429 + assert "deployment-1" in rejection.message + assert "gpt-5.6" in rejection.message + assert "max_parallel_requests=1" in rejection.message + assert "max_parallel_requests_queue_size=2" in rejection.message + assert semaphore.waiting == 2 + + release.set() + assert await asyncio.wait_for(asyncio.gather(holder, *queued), timeout=2) == ["ok", "ok", "ok"] + assert semaphore.waiting == 0 + assert not semaphore.locked() + + +@pytest.mark.asyncio +async def test_zero_queue_size_rejects_as_soon_as_every_slot_is_busy(): + semaphore: Final = _semaphore(queue_size=0, max_parallel_requests=2) + release: Final = asyncio.Event() + holders: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(2)] + await asyncio.sleep(0) + + await _expect_rejection(semaphore) + assert semaphore.waiting == 0 + + release.set() + assert await asyncio.wait_for(asyncio.gather(*holders), timeout=2) == ["ok", "ok"] + + +@pytest.mark.asyncio +async def test_unset_queue_size_parks_every_caller_until_a_slot_frees(): + semaphore: Final = _semaphore(queue_size=None) + release: Final = asyncio.Event() + callers: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(50)] + await asyncio.sleep(0) + assert semaphore.waiting == 49 + + release.set() + assert await asyncio.wait_for(asyncio.gather(*callers), timeout=2) == ["ok"] * 50 + assert semaphore.waiting == 0 + + +@pytest.mark.asyncio +async def test_cancelled_waiter_gives_its_queue_slot_back(): + semaphore: Final = _semaphore(queue_size=1) + release: Final = asyncio.Event() + holder: Final = asyncio.create_task(_hold(semaphore, release)) + await asyncio.sleep(0) + cancelled: Final = asyncio.create_task(_hold(semaphore, release)) + await asyncio.sleep(0) + assert semaphore.waiting == 1 + + cancelled.cancel() + with pytest.raises(asyncio.CancelledError): + await cancelled + assert semaphore.waiting == 0 + + replacement: Final = asyncio.create_task(_hold(semaphore, release)) + await asyncio.sleep(0) + assert semaphore.waiting == 1 + release.set() + assert await asyncio.wait_for(asyncio.gather(holder, replacement), timeout=2) == ["ok", "ok"] + + +def _router_semaphore(router: Router, model_name: str) -> DeploymentSemaphore: + deployment: Final = router.get_deployment_by_model_group_name(model_group_name=model_name) + assert deployment is not None + client: Final = router._get_client(deployment=deployment.model_dump(), kwargs={}, client_type="max_parallel_requests") + assert isinstance(client, DeploymentSemaphore) + return client + + +@pytest.mark.asyncio +async def test_deployment_queue_size_overrides_router_default_and_zero_is_honored(): + router: Final = Router( + model_list=[ + {"model_name": "inherits-default", "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1}}, + { + "model_name": "no-queue", + "litellm_params": {"model": "openai/gpt-5.6", "tpm": 100, "max_parallel_requests_queue_size": 0}, + }, + ], + default_max_parallel_requests_queue_size=1, + ) + release: Final = asyncio.Event() + + inherits: Final = _router_semaphore(router, "inherits-default") + inherits_holder: Final = asyncio.create_task(_hold(inherits, release)) + await asyncio.sleep(0) + inherits_waiter: Final = asyncio.create_task(_hold(inherits, release)) + await asyncio.sleep(0) + assert "max_parallel_requests_queue_size=1" in (await _expect_rejection(inherits)).message + + no_queue: Final = _router_semaphore(router, "no-queue") + no_queue_holder: Final = asyncio.create_task(_hold(no_queue, release)) + await asyncio.sleep(0) + assert "max_parallel_requests_queue_size=0" in (await _expect_rejection(no_queue)).message + + release.set() + await asyncio.wait_for(asyncio.gather(inherits_holder, inherits_waiter, no_queue_holder), timeout=2) + + +@pytest.mark.asyncio +async def test_router_without_queue_size_keeps_unbounded_queueing(): + router: Final = Router( + model_list=[{"model_name": "gpt-5.6", "litellm_params": {"model": "openai/gpt-5.6", "max_parallel_requests": 1}}] + ) + semaphore: Final = _router_semaphore(router, "gpt-5.6") + release: Final = asyncio.Event() + callers: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(20)] + await asyncio.sleep(0) + assert semaphore.waiting == 19 + release.set() + assert await asyncio.wait_for(asyncio.gather(*callers), timeout=2) == ["ok"] * 20 + + +@pytest.mark.asyncio +async def test_update_settings_applies_default_queue_size_to_live_semaphores_without_an_override(): + router: Final = Router( + model_list=[ + {"model_name": "inherits-default", "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1}}, + { + "model_name": "pinned", + "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1, "max_parallel_requests_queue_size": 5}, + }, + ], + ) + inherits: Final = _router_semaphore(router, "inherits-default") + pinned: Final = _router_semaphore(router, "pinned") + assert router.get_settings()["default_max_parallel_requests_queue_size"] is None + + router.update_settings(default_max_parallel_requests_queue_size="0") + assert router.get_settings()["default_max_parallel_requests_queue_size"] == 0 + assert (inherits.queue_size, pinned.queue_size) == (0, 5) + + release: Final = asyncio.Event() + holder: Final = asyncio.create_task(_hold(inherits, release)) + await asyncio.sleep(0) + assert "max_parallel_requests_queue_size=0" in (await _expect_rejection(inherits)).message + + router.update_settings(default_max_parallel_requests_queue_size=None) + assert (inherits.queue_size, pinned.queue_size) == (None, 5) + waiter: Final = asyncio.create_task(_hold(inherits, release)) + await asyncio.sleep(0) + assert inherits.waiting == 1 + + release.set() + assert await asyncio.wait_for(asyncio.gather(holder, waiter), timeout=2) == ["ok", "ok"] diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index 1e6636ec3d6..b094312808e 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -1,11 +1,13 @@ import asyncio import copy import functools +import gc import json import logging import os import sys import threading +import warnings from collections.abc import Awaitable, Callable, Mapping from datetime import datetime, timedelta from types import SimpleNamespace @@ -45,6 +47,7 @@ from litellm.router import ( _is_retriable_anthropic_status, ) from litellm.router_strategy import simple_shuffle +from litellm.router_utils.cooldown_handlers import _async_get_cooldown_deployments from litellm.types.llms.openai import ChatCompletionRequest from litellm.types.router import Deployment, DeploymentTypedDict, LiteLLM_Params, ModelInfo, PreRoutingHookResponse, RetryPolicy @@ -16038,6 +16041,224 @@ async def test_router_max_parallel_requests_slot_released_when_stream_closed_ear assert tracker.current == 0 +@pytest.mark.asyncio +async def test_router_max_parallel_requests_queue_size_turns_overflow_into_429(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + router: Final = Router( + model_list=[ + { + "model_name": "gpt-5.6", + "litellm_params": { + "model": "openai/gpt-5.6", + "api_key": "sk-fake", + "api_base": "https://max-parallel.local/v1", + "max_parallel_requests": 1, + "max_parallel_requests_queue_size": 1, + }, + "model_info": {"id": "queue-bounded-deployment"}, + }, + { + "model_name": "gpt-5.6", + "litellm_params": { + "model": "openai/gpt-5.6", + "api_key": "sk-fake", + "api_base": "https://max-parallel-sibling.local/v1", + }, + "model_info": {"id": "queue-sibling-deployment"}, + }, + ], + num_retries=0, + ) + + async def upstream(request: httpx.Request) -> httpx.Response: + await asyncio.sleep(0.2) + return httpx.Response( + 200, + json={ + "id": "c", + "object": "chat.completion", + "created": 1, + "model": "gpt-5.6", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "x"}, "finish_reason": "stop"}], + }, + ) + + with respx.mock(assert_all_called=False) as respx_mock: + route: Final = respx_mock.post("https://max-parallel.local/v1/chat/completions").mock(side_effect=upstream) + sibling_route: Final = respx_mock.post("https://max-parallel-sibling.local/v1/chat/completions").mock( + side_effect=upstream + ) + results: Final = await asyncio.wait_for( + asyncio.gather( + *( + router.acompletion(model="queue-bounded-deployment", messages=[{"role": "user", "content": "hi"}]) + for _ in range(3) + ), + return_exceptions=True, + ), + timeout=10, + ) + + rejected: Final = [r for r in results if isinstance(r, BaseException)] + assert len(rejected) == 1 and len(results) == 3 + assert isinstance(rejected[0], litellm.RateLimitError) + assert rejected[0].status_code == 429 + assert "queue-bounded-deployment" in rejected[0].message + assert "max_parallel_requests_queue_size=1" in rejected[0].message + assert route.call_count == 2 + assert sibling_route.call_count == 0 + assert all("max_parallel_requests_queue_size" not in call.request.content.decode() for call in route.calls) + assert await _async_get_cooldown_deployments(litellm_router_instance=router, parent_otel_span=None) == [] + + +@pytest.mark.asyncio +async def test_router_embedding_path_honors_max_parallel_requests_queue_size(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + router: Final = Router( + model_list=[ + { + "model_name": "embed", + "litellm_params": { + "model": "openai/text-embedding-3-small", + "api_key": "sk-fake", + "api_base": "https://max-parallel-embed.local/v1", + "max_parallel_requests": 1, + }, + "model_info": {"id": "embed-bounded-deployment"}, + } + ], + default_max_parallel_requests_queue_size=1, + num_retries=0, + ) + + async def upstream(request: httpx.Request) -> httpx.Response: + await asyncio.sleep(0.2) + return httpx.Response( + 200, + json={ + "object": "list", + "data": [{"object": "embedding", "index": 0, "embedding": [0.1, 0.2]}], + "model": "text-embedding-3-small", + "usage": {"prompt_tokens": 1, "total_tokens": 1}, + }, + ) + + with respx.mock() as respx_mock, warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + route: Final = respx_mock.post("https://max-parallel-embed.local/v1/embeddings").mock(side_effect=upstream) + results: Final = await asyncio.wait_for( + asyncio.gather( + *(router.aembedding(model="embed", input=["hi"]) for _ in range(3)), + return_exceptions=True, + ), + timeout=10, + ) + gc.collect() + + rejected: Final = [r for r in results if isinstance(r, BaseException)] + assert len(rejected) == 1 and len(results) == 3 + assert isinstance(rejected[0], litellm.RateLimitError) and rejected[0].status_code == 429 + assert "embed-bounded-deployment" in rejected[0].message + assert route.call_count == 2 + assert [str(w.message) for w in caught if "never awaited" in str(w.message)] == [] + + +@pytest.mark.asyncio +async def test_router_max_parallel_requests_queue_overflow_takes_the_ordinary_429_fallback_path( + monkeypatch: pytest.MonkeyPatch, +): + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + router: Final = Router( + model_list=[ + { + "model_name": "gpt-5.6", + "litellm_params": { + "model": "openai/gpt-5.6", + "api_key": "sk-fake", + "api_base": "https://max-parallel-primary.local/v1", + "max_parallel_requests": 1, + "max_parallel_requests_queue_size": 0, + }, + "model_info": {"id": "queue-primary-deployment"}, + }, + { + "model_name": "gpt-5.6-fallback", + "litellm_params": { + "model": "openai/gpt-5.6", + "api_key": "sk-fake", + "api_base": "https://max-parallel-fallback.local/v1", + }, + "model_info": {"id": "queue-fallback-deployment"}, + }, + ], + fallbacks=[{"gpt-5.6": ["gpt-5.6-fallback"]}], + num_retries=0, + ) + + async def upstream(request: httpx.Request) -> httpx.Response: + await asyncio.sleep(0.2) + return httpx.Response( + 200, + json={ + "id": "c", + "object": "chat.completion", + "created": 1, + "model": "gpt-5.6", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "x"}, "finish_reason": "stop"}], + }, + ) + + with respx.mock() as respx_mock: + primary: Final = respx_mock.post("https://max-parallel-primary.local/v1/chat/completions").mock( + side_effect=upstream + ) + fallback: Final = respx_mock.post("https://max-parallel-fallback.local/v1/chat/completions").mock( + side_effect=upstream + ) + results: Final = await asyncio.wait_for( + asyncio.gather( + *(router.acompletion(model="gpt-5.6", messages=[{"role": "user", "content": "hi"}]) for _ in range(3)) + ), + timeout=10, + ) + + assert len(results) == 3 + assert primary.call_count == 1 + assert fallback.call_count == 2 + assert await _async_get_cooldown_deployments(litellm_router_instance=router, parent_otel_span=None) == [] + + +@pytest.mark.asyncio +async def test_router_deployment_slot_rejects_once_queue_is_full_and_frees_slot_on_exit(): + router: Final = Router( + model_list=[ + { + "model_name": "gpt-5.6", + "litellm_params": { + "model": "openai/gpt-5.6", + "api_key": "sk-fake", + "max_parallel_requests": 1, + "max_parallel_requests_queue_size": 0, + }, + "model_info": {"id": "slot-deployment"}, + } + ] + ) + deployment: Final = router.get_deployment(model_id="slot-deployment") + assert deployment is not None + kwargs: Final = {"model": "gpt-5.6"} + + async with router._deployment_slot(deployment=deployment.model_dump(), kwargs=kwargs, parent_otel_span=None): + with pytest.raises(litellm.RateLimitError) as overflow: + async with router._deployment_slot(deployment=deployment.model_dump(), kwargs=kwargs, parent_otel_span=None): + pass + assert overflow.value.status_code == 429 + assert "slot-deployment" in overflow.value.message + + async with router._deployment_slot(deployment=deployment.model_dump(), kwargs=kwargs, parent_otel_span=None): + pass + + @pytest.mark.asyncio async def test_router_deployment_drop_params_string_true_is_honored(monkeypatch): from litellm import Router diff --git a/tests/test_litellm/test_utils.py b/tests/test_litellm/test_utils.py index f219f26b353..07da804c0f9 100644 --- a/tests/test_litellm/test_utils.py +++ b/tests/test_litellm/test_utils.py @@ -60,6 +60,7 @@ from litellm.utils import ( _snapshot_exception_for_hook, async_post_call_failure_deployment_hook, async_post_call_success_deployment_hook, + calculate_max_parallel_requests, client, get_non_default_completion_params, get_optional_params_image_gen, @@ -6213,3 +6214,32 @@ def test_provider_prefixed_lookup_never_outranks_an_existing_row(local_model_cos ("openrouter/openai/gpt-4o", "openrouter", "openrouter/openai/gpt-4o"), ): assert litellm.get_model_info(model=model, custom_llm_provider=provider)["key"] == expected_key + + +@pytest.mark.parametrize( + ("max_parallel_requests", "rpm", "tpm", "default_max_parallel_requests", "expected"), + [ + (3, 100, 100_000, 7, 3), + (None, 100, 100_000, 7, 100), + (None, None, 100_000, 7, 600), + (None, None, 50, 7, 1), + (None, None, None, 7, 7), + (None, None, None, None, None), + ], +) +def test_calculate_max_parallel_requests_precedence( + max_parallel_requests: int | None, + rpm: int | None, + tpm: int | None, + default_max_parallel_requests: int | None, + expected: int | None, +) -> None: + assert ( + calculate_max_parallel_requests( + max_parallel_requests=max_parallel_requests, + rpm=rpm, + tpm=tpm, + default_max_parallel_requests=default_max_parallel_requests, + ) + == expected + ) diff --git a/ui/litellm-dashboard/src/components/router_settings/index.test.tsx b/ui/litellm-dashboard/src/components/router_settings/index.test.tsx index 1875085231a..f2740bbd1e0 100644 --- a/ui/litellm-dashboard/src/components/router_settings/index.test.tsx +++ b/ui/litellm-dashboard/src/components/router_settings/index.test.tsx @@ -137,6 +137,41 @@ describe("RouterSettings", () => { ); }); + it("should save default_max_parallel_requests_queue_size as a number and an empty field as null", async () => { + vi.mocked(getCallbacksCall).mockResolvedValue({ + router_settings: { ...mockCallbacksResponse.router_settings, default_max_parallel_requests_queue_size: null }, + }); + const user = userEvent.setup(); + renderWithProviders(); + + await findStrategySelect(); + + const queueSize = await screen.findByRole("textbox", { name: /default_max_parallel_requests_queue_size/i }); + fireEvent.change(queueSize, { target: { value: "4" } }); + await user.click(screen.getByRole("button", { name: /save changes/i })); + + await waitFor(() => + expect(setCallbacksCall).toHaveBeenLastCalledWith( + "test-token", + expect.objectContaining({ + router_settings: expect.objectContaining({ default_max_parallel_requests_queue_size: 4 }), + }), + ), + ); + + fireEvent.change(queueSize, { target: { value: "" } }); + await user.click(screen.getByRole("button", { name: /save changes/i })); + + await waitFor(() => + expect(setCallbacksCall).toHaveBeenLastCalledWith( + "test-token", + expect.objectContaining({ + router_settings: expect.objectContaining({ default_max_parallel_requests_queue_size: null }), + }), + ), + ); + }); + it("should show a success notification after saving", async () => { const user = userEvent.setup(); renderWithProviders(); diff --git a/ui/litellm-dashboard/src/components/router_settings/index.tsx b/ui/litellm-dashboard/src/components/router_settings/index.tsx index 53d35b81cec..4170d48361d 100644 --- a/ui/litellm-dashboard/src/components/router_settings/index.tsx +++ b/ui/litellm-dashboard/src/components/router_settings/index.tsx @@ -86,7 +86,15 @@ const RouterSettings: React.FC = ({ accessToken, userRole, const router_settings = formValue.routerSettings; - const numberKeys = new Set(["allowed_fails", "cooldown_time", "num_retries", "timeout", "retry_after"]); + const numberKeys = new Set([ + "allowed_fails", + "cooldown_time", + "num_retries", + "timeout", + "retry_after", + "default_max_parallel_requests_queue_size", + ]); + const unsettableNumberKeys = new Set(["default_max_parallel_requests_queue_size"]); const jsonKeys = new Set(["model_group_alias"]); // retry_policy and model_group_retry_policy are owned by the Model Retry Settings tab; // routing_groups is owned by the Routing Groups tab. This page must not read or write them. @@ -100,6 +108,7 @@ const RouterSettings: React.FC = ({ accessToken, userRole, if (v.toLowerCase() === "null") return null; if (numberKeys.has(key)) { + if (v === "" && unsettableNumberKeys.has(key)) return null; const n = Number(v); return Number.isNaN(n) ? fallback : n; } From fe0eee64512ff1832b227dde5a1cd0649ce4a5bd Mon Sep 17 00:00:00 2001 From: yassin Date: Thu, 17 Sep 2026 03:02:42 +0000 Subject: [PATCH 2/5] fix(anthropic): keep prompt cache prediction supported for queue-bounded deployments Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/llms/anthropic/prompt_cache_prediction.py | 1 + .../anthropic/test_anthropic_prompt_cache_prediction.py | 9 +++++++++ 2 files changed, 10 insertions(+) diff --git a/litellm/llms/anthropic/prompt_cache_prediction.py b/litellm/llms/anthropic/prompt_cache_prediction.py index e69a02bd93a..a0ce5bf0360 100644 --- a/litellm/llms/anthropic/prompt_cache_prediction.py +++ b/litellm/llms/anthropic/prompt_cache_prediction.py @@ -50,6 +50,7 @@ _DEPLOYMENT_OPTIONS: Final = frozenset( "max_retries", "num_retries", "max_parallel_requests", + "max_parallel_requests_queue_size", "input_cost_per_token", "output_cost_per_token", "cache_read_input_token_cost", diff --git a/tests/test_litellm/llms/anthropic/test_anthropic_prompt_cache_prediction.py b/tests/test_litellm/llms/anthropic/test_anthropic_prompt_cache_prediction.py index 2b36866a1a0..c13217a0d46 100644 --- a/tests/test_litellm/llms/anthropic/test_anthropic_prompt_cache_prediction.py +++ b/tests/test_litellm/llms/anthropic/test_anthropic_prompt_cache_prediction.py @@ -181,6 +181,15 @@ async def test_environment_credential_matches_native_count_and_observed_scope( assert observed.scope == cache_scope(_CALLER, _DEPLOYMENT, target.api_key, target.model) +def test_deployment_concurrency_knobs_keep_native_prediction_supported() -> None: + target: Final = resolve_prediction_target(LiteLLM_Params( + model=f"anthropic/{_MODEL}", api_key=_KEY, api_base="https://api.anthropic.com", + max_parallel_requests=1, max_parallel_requests_queue_size=0, + )) + assert isinstance(target, NativePredictionTarget) + assert (target.model, target.api_key) == (_MODEL, _KEY) + + @pytest.mark.parametrize("inline_key", [None, _KEY]) @pytest.mark.asyncio async def test_named_credential_is_explicitly_unsupported_before_count( From 6be9c4a978c2a35d9cae254e2c05a1dc536d2fd4 Mon Sep 17 00:00:00 2001 From: yassin Date: Thu, 17 Sep 2026 03:17:49 +0000 Subject: [PATCH 3/5] refactor(router): compose DeploymentSemaphore over asyncio.Semaphore instead of subclassing it Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/router.py | 6 ++-- .../client_initalization_utils.py | 31 ++++++++++++++----- tests/test_litellm/test_router.py | 5 ++- 3 files changed, 30 insertions(+), 12 deletions(-) diff --git a/litellm/router.py b/litellm/router.py index b62c83b8ab1..132f48730df 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -148,7 +148,7 @@ from litellm.router_utils.batch_utils import ( replace_model_in_jsonl, should_replace_model_in_jsonl, ) -from litellm.router_utils.client_initalization_utils import InitalizeCachedClient +from litellm.router_utils.client_initalization_utils import DeploymentSemaphore, InitalizeCachedClient from litellm.router_utils.clientside_credential_handler import ( get_dynamic_litellm_params, is_clientside_credential, @@ -3640,7 +3640,7 @@ class Router: client_type="max_parallel_requests", ) async with contextlib.AsyncExitStack() as deployment_slot: - if isinstance(rpm_semaphore, asyncio.Semaphore): + if isinstance(rpm_semaphore, DeploymentSemaphore): await deployment_slot.enter_async_context(rpm_semaphore) await self.async_routing_strategy_pre_call_checks( deployment=deployment, @@ -8512,7 +8512,7 @@ class Router: client_type="max_parallel_requests", ) async with contextlib.AsyncExitStack() as slot: - if isinstance(rpm_semaphore, asyncio.Semaphore): + if isinstance(rpm_semaphore, DeploymentSemaphore): await slot.enter_async_context(rpm_semaphore) await self.async_routing_strategy_pre_call_checks(deployment=deployment, parent_otel_span=parent_otel_span) yield diff --git a/litellm/router_utils/client_initalization_utils.py b/litellm/router_utils/client_initalization_utils.py index a135978d09e..72854cba028 100644 --- a/litellm/router_utils/client_initalization_utils.py +++ b/litellm/router_utils/client_initalization_utils.py @@ -1,5 +1,6 @@ import asyncio import time +from types import TracebackType from typing import TYPE_CHECKING, Any, Final from litellm._logging import verbose_router_logger @@ -15,22 +16,36 @@ else: LitellmRouter = Any -class DeploymentSemaphore(asyncio.Semaphore): +class DeploymentSemaphore: """A deployment's max_parallel_requests slots. ``queue_size=None`` parks callers without bound, like a plain ``asyncio.Semaphore``; otherwise a caller arriving while all slots are busy and ``queue_size`` callers already wait gets a 429 instead of being parked.""" def __init__(self, max_parallel_requests: int, model_id: str, model_group: str, queue_size: int | None) -> None: - super().__init__(max_parallel_requests) - self.max_parallel_requests = max_parallel_requests - self.model_id = model_id - self.model_group = model_group + self._slots: Final = asyncio.Semaphore(max_parallel_requests) + self.max_parallel_requests: Final = max_parallel_requests + self.model_id: Final = model_id + self.model_group: Final = model_group self.queue_size = queue_size self.waiting = 0 + def locked(self) -> bool: + return self._slots.locked() + + def release(self) -> None: + self._slots.release() + + async def __aenter__(self) -> None: + await self.acquire() + + async def __aexit__( + self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None + ) -> None: + self._slots.release() + async def acquire(self) -> bool: - if not self.locked(): - return await super().acquire() + if not self._slots.locked(): + return await self._slots.acquire() if self.queue_size is not None and self.waiting >= self.queue_size: raise RateLimitError( message=( @@ -57,7 +72,7 @@ class DeploymentSemaphore(asyncio.Semaphore): self.queue_size, ) try: - return await super().acquire() + return await self._slots.acquire() finally: self.waiting -= 1 verbose_router_logger.debug( diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index b094312808e..a674f767dde 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -47,6 +47,7 @@ from litellm.router import ( _is_retriable_anthropic_status, ) from litellm.router_strategy import simple_shuffle +from litellm.router_utils.client_initalization_utils import DeploymentSemaphore from litellm.router_utils.cooldown_handlers import _async_get_cooldown_deployments from litellm.types.llms.openai import ChatCompletionRequest from litellm.types.router import Deployment, DeploymentTypedDict, LiteLLM_Params, ModelInfo, PreRoutingHookResponse, RetryPolicy @@ -1520,7 +1521,9 @@ async def test_router_ageneric_api_call_with_fallbacks_helper(): }, } - mock_semaphore = asyncio.Semaphore(1) + mock_semaphore = DeploymentSemaphore( + max_parallel_requests=1, model_id="deployment-1", model_group="gpt-3.5-turbo", queue_size=None + ) with patch.object( router, "_update_kwargs_with_deployment" From 45ceb5611015fa368f4f89ab403bb9aeb05fb05a Mon Sep 17 00:00:00 2001 From: yassin Date: Thu, 17 Sep 2026 04:05:41 +0000 Subject: [PATCH 4/5] fix(router): validate max_parallel_requests_queue_size as a non-negative integer Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/proxy/proxy_server.py | 12 +++++++ litellm/router.py | 7 ++-- .../client_initalization_utils.py | 4 +-- litellm/types/router.py | 15 ++++++-- .../router_code_coverage.py | 2 +- tests/test_litellm/proxy/test_proxy_server.py | 19 +++++++++++ .../test_client_initalization_utils.py | 34 ++++++++++++++++++- ui/litellm-dashboard/src/lib/http/schema.d.ts | 4 +++ 8 files changed, 88 insertions(+), 9 deletions(-) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index dc78ec75a6e..216a146143d 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -776,6 +776,7 @@ from litellm.types.router import ( RoutingPlugin, SearchToolTypedDict, updateDeployment, + validate_max_parallel_requests_queue_size, ) from litellm.types.router import ModelInfo as RouterModelInfo from litellm.types.scheduler import DefaultPriorities @@ -16936,6 +16937,17 @@ async def update_config( ) }, ) + raw_queue_size: Final = raw_router_settings.get("default_max_parallel_requests_queue_size") + try: + validate_max_parallel_requests_queue_size(raw_queue_size) + except ValueError as invalid_queue_size: + raise HTTPException( + status_code=400, + detail=( + f"default_max_parallel_requests_queue_size={raw_queue_size!r} is not valid, " + "it must be a non-negative integer or null" + ), + ) from invalid_queue_size if prisma_client is None: raise Exception("No DB Connected") diff --git a/litellm/router.py b/litellm/router.py index 132f48730df..97325e6c450 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -263,6 +263,7 @@ from litellm.types.router import ( RoutingStrategy, SearchToolTypedDict, TaggedPreRoutingStrategy, + validate_max_parallel_requests_queue_size, ) from litellm.types.services import ServiceTypes from litellm.types.utils import ( @@ -936,7 +937,9 @@ class Router: None # use this to track the users default deployment, when they want to use model = * ) self.default_max_parallel_requests = default_max_parallel_requests - self._default_max_parallel_requests_queue_size = default_max_parallel_requests_queue_size + self._default_max_parallel_requests_queue_size = validate_max_parallel_requests_queue_size( + default_max_parallel_requests_queue_size + ) self.provider_default_deployment_ids: list[str] = [] self.pattern_router = PatternMatchRouter() self.team_pattern_routers: dict[str, PatternMatchRouter] = {} # {"TEAM_ID": PatternMatchRouter} @@ -11852,7 +11855,7 @@ class Router: @default_max_parallel_requests_queue_size.setter def default_max_parallel_requests_queue_size(self, queue_size: int | None) -> None: - self._default_max_parallel_requests_queue_size = None if queue_size is None else int(queue_size) + self._default_max_parallel_requests_queue_size = validate_max_parallel_requests_queue_size(queue_size) InitalizeCachedClient.apply_default_max_parallel_requests_queue_size( litellm_router_instance=self, queue_size=self._default_max_parallel_requests_queue_size ) diff --git a/litellm/router_utils/client_initalization_utils.py b/litellm/router_utils/client_initalization_utils.py index 72854cba028..be5f71a4e70 100644 --- a/litellm/router_utils/client_initalization_utils.py +++ b/litellm/router_utils/client_initalization_utils.py @@ -5,7 +5,7 @@ from typing import TYPE_CHECKING, Any, Final from litellm._logging import verbose_router_logger from litellm.exceptions import RateLimitError, RateLimitErrorCategory, RateLimitType -from litellm.types.router import RouterErrors +from litellm.types.router import RouterErrors, validate_max_parallel_requests_queue_size from litellm.utils import calculate_max_parallel_requests if TYPE_CHECKING: @@ -26,7 +26,7 @@ class DeploymentSemaphore: self.max_parallel_requests: Final = max_parallel_requests self.model_id: Final = model_id self.model_group: Final = model_group - self.queue_size = queue_size + self.queue_size = validate_max_parallel_requests_queue_size(queue_size) self.waiting = 0 def locked(self) -> bool: diff --git a/litellm/types/router.py b/litellm/types/router.py index 8f788b5f933..848dd28aaac 100644 --- a/litellm/types/router.py +++ b/litellm/types/router.py @@ -6,10 +6,10 @@ import datetime import enum from collections.abc import Mapping, Sequence from dataclasses import dataclass -from typing import TYPE_CHECKING, Any, ClassVar, Final, Generic, Literal, TypeVar, get_type_hints +from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Final, Generic, Literal, TypeVar, get_type_hints import httpx -from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, field_validator, model_validator from typing_extensions import Protocol, ReadOnly, Required, TypedDict, runtime_checkable from litellm._logging import verbose_logger @@ -314,6 +314,14 @@ class CredentialLiteLLMParams(BaseModel): _RESERVED_INIT_KEYS: Final = frozenset({"self", "params", "__class__"}) +MaxParallelRequestsQueueSize = Annotated[int, Field(strict=True, ge=0)] +_MAX_PARALLEL_REQUESTS_QUEUE_SIZE_ADAPTER: Final = TypeAdapter(MaxParallelRequestsQueueSize | None) + + +def validate_max_parallel_requests_queue_size(value: object) -> int | None: + return _MAX_PARALLEL_REQUESTS_QUEUE_SIZE_ADAPTER.validate_python(value) + + class GenericLiteLLMParams(CredentialLiteLLMParams, CustomPricingLiteLLMParams): """ LiteLLM Params without 'model' arg (used across completion / assistants api) @@ -324,6 +332,7 @@ class GenericLiteLLMParams(CredentialLiteLLMParams, CustomPricingLiteLLMParams): rpm: int | None = None itpm: int | None = None otpm: int | None = None + max_parallel_requests_queue_size: MaxParallelRequestsQueueSize | None = None timeout: float | str | httpx.Timeout | None = None # if str, pass in as os.environ/ stream_timeout: float | str | None = None # timeout when making stream=True calls, if str, pass in as os.environ/ max_retries: int | None = None @@ -497,7 +506,7 @@ class LiteLLMParamsTypedDict(TypedDict, total=False): order: int | None weight: int | None max_parallel_requests: int | None - max_parallel_requests_queue_size: ReadOnly[int | None] + max_parallel_requests_queue_size: ReadOnly[MaxParallelRequestsQueueSize | None] api_key: str | None api_base: str | None api_version: str | None diff --git a/tests/code_coverage_tests/router_code_coverage.py b/tests/code_coverage_tests/router_code_coverage.py index 057e82a24c8..582977d613b 100644 --- a/tests/code_coverage_tests/router_code_coverage.py +++ b/tests/code_coverage_tests/router_code_coverage.py @@ -88,7 +88,7 @@ ignored_function_names = [ "_resolve_claude_code_session_router", # Tested through Claude Code session routing in test_router.py "_get_claude_code_session_router_binding", # Tested through the two-worker session routing test in test_router.py "_apply_updated_routing_strategy_args", # Tested via update_settings in test_lowest_latency.py (file lacks "router" in name) - "default_max_parallel_requests_queue_size", # Property, so its reads and assignments in test_router.py are never an ast.Call + "default_max_parallel_requests_queue_size", ] diff --git a/tests/test_litellm/proxy/test_proxy_server.py b/tests/test_litellm/proxy/test_proxy_server.py index 5754301ac4a..5a2e76039e8 100644 --- a/tests/test_litellm/proxy/test_proxy_server.py +++ b/tests/test_litellm/proxy/test_proxy_server.py @@ -9393,6 +9393,25 @@ def test_update_config_router_settings_null_clears_max_parallel_requests_queue_s restore() +@pytest.mark.parametrize("invalid_queue_size", [-1, 2.5, "3"]) +def test_update_config_rejects_invalid_max_parallel_requests_queue_size_before_persisting( + _update_config_setup, invalid_queue_size +): + client, prisma, restore = _update_config_setup( + initial_rows={"router_settings": {"default_max_parallel_requests_queue_size": 3}}, + ) + try: + resp = client.post( + "/config/update", + json={"router_settings": {"default_max_parallel_requests_queue_size": invalid_queue_size}}, + ) + assert resp.status_code == 400 + assert "default_max_parallel_requests_queue_size" in resp.json()["error"]["message"] + assert prisma.db.litellm_config.rows["router_settings"] == {"default_max_parallel_requests_queue_size": 3} + finally: + restore() + + def test_update_config_success_callback_normalizes_existing_mixed_case( _update_config_setup, ): diff --git a/tests/test_litellm/router_utils/test_client_initalization_utils.py b/tests/test_litellm/router_utils/test_client_initalization_utils.py index 332f2f1503a..a6626d2f975 100644 --- a/tests/test_litellm/router_utils/test_client_initalization_utils.py +++ b/tests/test_litellm/router_utils/test_client_initalization_utils.py @@ -2,6 +2,7 @@ import asyncio from typing import Final import pytest +from pydantic import ValidationError import litellm from litellm import Router @@ -111,6 +112,37 @@ def _router_semaphore(router: Router, model_name: str) -> DeploymentSemaphore: return client +@pytest.mark.parametrize("invalid_queue_size", [-1, 2.5, True, "3"]) +def test_invalid_queue_sizes_are_rejected_instead_of_coerced(invalid_queue_size: object): + """A negative bound would reject every busy request and a fraction would be truncated, so + neither may reach a semaphore, the router default, or a live update of that default.""" + with pytest.raises(ValidationError): + _semaphore(queue_size=invalid_queue_size) + model_list: Final = [{"model_name": "gpt-5.6", "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1}}] + with pytest.raises(ValidationError): + Router(model_list=model_list, default_max_parallel_requests_queue_size=invalid_queue_size) + with pytest.raises(ValidationError): + Router( + model_list=[ + { + "model_name": "gpt-5.6", + "litellm_params": { + "model": "openai/gpt-5.6", + "rpm": 1, + "max_parallel_requests_queue_size": invalid_queue_size, + }, + } + ] + ) + + router: Final = Router(model_list=model_list, default_max_parallel_requests_queue_size=4) + semaphore: Final = _router_semaphore(router, "gpt-5.6") + with pytest.raises(ValidationError): + router.update_settings(default_max_parallel_requests_queue_size=invalid_queue_size) + assert router.default_max_parallel_requests_queue_size == 4 + assert semaphore.queue_size == 4 + + @pytest.mark.asyncio async def test_deployment_queue_size_overrides_router_default_and_zero_is_honored(): router: Final = Router( @@ -170,7 +202,7 @@ async def test_update_settings_applies_default_queue_size_to_live_semaphores_wit pinned: Final = _router_semaphore(router, "pinned") assert router.get_settings()["default_max_parallel_requests_queue_size"] is None - router.update_settings(default_max_parallel_requests_queue_size="0") + router.update_settings(default_max_parallel_requests_queue_size=0) assert router.get_settings()["default_max_parallel_requests_queue_size"] == 0 assert (inherits.queue_size, pinned.queue_size) == (0, 5) diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 872875cc535..84ca93eccfe 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -30441,6 +30441,8 @@ export interface components { max_budget?: number | null; /** Max File Size Mb */ max_file_size_mb?: number | null; + /** Max Parallel Requests Queue Size */ + max_parallel_requests_queue_size?: number | null; /** Max Retries */ max_retries?: number | null; /** @@ -40893,6 +40895,8 @@ export interface components { max_budget?: number | null; /** Max File Size Mb */ max_file_size_mb?: number | null; + /** Max Parallel Requests Queue Size */ + max_parallel_requests_queue_size?: number | null; /** Max Retries */ max_retries?: number | null; /** From 8d972eefc7404d4f626899ab0d222ce8134d0e02 Mon Sep 17 00:00:00 2001 From: yassin Date: Thu, 17 Sep 2026 18:14:09 +0000 Subject: [PATCH 5/5] feat(router): reject with 429 when a deployment's max_parallel_requests slots are all in use Replace the per-deployment asyncio.Semaphore with MaxParallelRequestsLimit, which admits a call synchronously or raises the router's RateLimitError (429) right away. Nothing waits for a slot any more, so the max_parallel_requests_queue_size and default_max_parallel_requests_queue_size settings from the earlier commits are dropped along with their proxy validation, dashboard control and generated schema entries. The rpm/tpm derivation of the cap is unchanged. Every router endpoint family now enters the slot through one _deployment_slot context, and the provider coroutine is only created once the slot is held Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/constants.py | 2 - .../llms/anthropic/prompt_cache_prediction.py | 1 - litellm/proxy/proxy_server.py | 28 +-- litellm/router.py | 31 +-- .../client_initalization_utils.py | 96 ++----- .../router_settings_endpoints.py | 11 - litellm/types/router.py | 16 +- litellm/types/utils.py | 1 - .../router_code_coverage.py | 1 - .../test_router_max_parallel_requests.py | 13 +- .../test_anthropic_prompt_cache_prediction.py | 9 - tests/test_litellm/proxy/test_proxy_server.py | 78 ------ .../test_client_initalization_utils.py | 236 ++++++------------ tests/test_litellm/test_router.py | 87 ++++--- .../components/router_settings/index.test.tsx | 35 --- .../src/components/router_settings/index.tsx | 11 +- ui/litellm-dashboard/src/lib/http/schema.d.ts | 4 - 17 files changed, 160 insertions(+), 500 deletions(-) diff --git a/litellm/constants.py b/litellm/constants.py index 0cd59706015..8409a161800 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -30,10 +30,8 @@ RUNTIME_UPDATABLE_ROUTER_SETTINGS: Final[frozenset[str]] = frozenset( "enable_tag_filtering", "tag_routing_prefix", "optional_pre_call_checks", - "default_max_parallel_requests_queue_size", } ) -NULLABLE_RUNTIME_ROUTER_SETTINGS: Final[frozenset[str]] = frozenset({"default_max_parallel_requests_queue_size"}) ROUTER_SETTINGS_MANAGED_OUTSIDE_CONFIG: Final[frozenset[str]] = frozenset( { "model_list", diff --git a/litellm/llms/anthropic/prompt_cache_prediction.py b/litellm/llms/anthropic/prompt_cache_prediction.py index a0ce5bf0360..e69a02bd93a 100644 --- a/litellm/llms/anthropic/prompt_cache_prediction.py +++ b/litellm/llms/anthropic/prompt_cache_prediction.py @@ -50,7 +50,6 @@ _DEPLOYMENT_OPTIONS: Final = frozenset( "max_retries", "num_retries", "max_parallel_requests", - "max_parallel_requests_queue_size", "input_cost_per_token", "output_cost_per_token", "cache_read_input_token_cost", diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 216a146143d..7bc36e175c0 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -70,7 +70,6 @@ from litellm.constants import ( LITELLM_SETTINGS_SAFE_DB_OVERRIDES, LITELLM_UI_ALLOW_HEADERS, LITELLM_UI_SESSION_DURATION, - NULLABLE_RUNTIME_ROUTER_SETTINGS, RUNTIME_UPDATABLE_ROUTER_SETTINGS, ) from litellm.litellm_core_utils.asyncify import asyncify @@ -776,7 +775,6 @@ from litellm.types.router import ( RoutingPlugin, SearchToolTypedDict, updateDeployment, - validate_max_parallel_requests_queue_size, ) from litellm.types.router import ModelInfo as RouterModelInfo from litellm.types.scheduler import DefaultPriorities @@ -6902,20 +6900,13 @@ class ProxyConfig: ): from litellm.utils import _update_dictionary - db_settings: Final = db_router_settings.param_value db_overlay_deferring_empty_lists_to_config: Final = { k: v - for k, v in db_settings.items() + for k, v in db_router_settings.param_value.items() if not (k in config_router_settings and isinstance(v, list) and len(v) == 0) } - cleared_nullable_settings: Final = MappingProxyType( - {k: None for k in NULLABLE_RUNTIME_ROUTER_SETTINGS if k in db_settings and db_settings[k] is None} - ) - combined_router_settings = MappingProxyType( - { - **_update_dictionary(config_router_settings, db_overlay_deferring_empty_lists_to_config), - **cleared_nullable_settings, - } + combined_router_settings = _update_dictionary( + config_router_settings, db_overlay_deferring_empty_lists_to_config ) elif config_router_settings is not None and isinstance(config_router_settings, dict): combined_router_settings = config_router_settings @@ -16937,17 +16928,6 @@ async def update_config( ) }, ) - raw_queue_size: Final = raw_router_settings.get("default_max_parallel_requests_queue_size") - try: - validate_max_parallel_requests_queue_size(raw_queue_size) - except ValueError as invalid_queue_size: - raise HTTPException( - status_code=400, - detail=( - f"default_max_parallel_requests_queue_size={raw_queue_size!r} is not valid, " - "it must be a non-negative integer or null" - ), - ) from invalid_queue_size if prisma_client is None: raise Exception("No DB Connected") @@ -17059,7 +17039,7 @@ async def update_config( raw_router_settings_without_none: Final = { key: value for key, value in raw_router_settings.items() - if key not in typed_router_settings and (value is not None or key in NULLABLE_RUNTIME_ROUTER_SETTINGS) + if key not in typed_router_settings and value is not None } router_settings_updates: Final = {**typed_router_settings, **raw_router_settings_without_none} new_router_settings: Final = {**existing, **router_settings_updates} diff --git a/litellm/router.py b/litellm/router.py index 97325e6c450..d645fe0fab8 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -148,7 +148,7 @@ from litellm.router_utils.batch_utils import ( replace_model_in_jsonl, should_replace_model_in_jsonl, ) -from litellm.router_utils.client_initalization_utils import DeploymentSemaphore, InitalizeCachedClient +from litellm.router_utils.client_initalization_utils import InitalizeCachedClient, MaxParallelRequestsLimit from litellm.router_utils.clientside_credential_handler import ( get_dynamic_litellm_params, is_clientside_credential, @@ -263,7 +263,6 @@ from litellm.types.router import ( RoutingStrategy, SearchToolTypedDict, TaggedPreRoutingStrategy, - validate_max_parallel_requests_queue_size, ) from litellm.types.services import ServiceTypes from litellm.types.utils import ( @@ -739,7 +738,6 @@ class Router: stream_timeout: float | None = None, default_litellm_params: dict | None = None, # default params for Router.chat.completion.create default_max_parallel_requests: int | None = None, - default_max_parallel_requests_queue_size: int | None = None, set_verbose: bool = False, debug_level: Literal["DEBUG", "INFO"] = "INFO", default_fallbacks: list[str] | None = None, # generic fallbacks, works across all deployments @@ -937,9 +935,6 @@ class Router: None # use this to track the users default deployment, when they want to use model = * ) self.default_max_parallel_requests = default_max_parallel_requests - self._default_max_parallel_requests_queue_size = validate_max_parallel_requests_queue_size( - default_max_parallel_requests_queue_size - ) self.provider_default_deployment_ids: list[str] = [] self.pattern_router = PatternMatchRouter() self.team_pattern_routers: dict[str, PatternMatchRouter] = {} # {"TEAM_ID": PatternMatchRouter} @@ -3637,14 +3632,14 @@ class Router: logging_obj: Final[LiteLLMLogging | None] = kwargs.get("litellm_logging_obj", None) - rpm_semaphore: Final = self._get_client( + max_parallel_requests_limit: Final = self._get_client( deployment=deployment, kwargs=kwargs, client_type="max_parallel_requests", ) async with contextlib.AsyncExitStack() as deployment_slot: - if isinstance(rpm_semaphore, DeploymentSemaphore): - await deployment_slot.enter_async_context(rpm_semaphore) + if isinstance(max_parallel_requests_limit, MaxParallelRequestsLimit): + deployment_slot.enter_context(max_parallel_requests_limit) await self.async_routing_strategy_pre_call_checks( deployment=deployment, logging_obj=logging_obj, @@ -8509,14 +8504,14 @@ class Router: ) -> AsyncGenerator[None, None]: """Holds the deployment's max_parallel_requests slot, if it has one, around the provider call. Routing strategy pre-call checks run inside the slot so their rpm accounting stays concurrency-safe.""" - rpm_semaphore: Final = self._get_client( + max_parallel_requests_limit: Final = self._get_client( deployment=deployment, kwargs=kwargs, client_type="max_parallel_requests", ) async with contextlib.AsyncExitStack() as slot: - if isinstance(rpm_semaphore, DeploymentSemaphore): - await slot.enter_async_context(rpm_semaphore) + if isinstance(max_parallel_requests_limit, MaxParallelRequestsLimit): + slot.enter_context(max_parallel_requests_limit) await self.async_routing_strategy_pre_call_checks(deployment=deployment, parent_otel_span=parent_otel_span) yield @@ -11846,20 +11841,8 @@ class Router: _settings_to_return[var] = self.lowestlatency_logger.routing_args.json() _settings_to_return["routing_groups"] = [group.model_dump() for group in self._routing_groups.values()] - _settings_to_return["default_max_parallel_requests_queue_size"] = self.default_max_parallel_requests_queue_size return _settings_to_return - @property - def default_max_parallel_requests_queue_size(self) -> int | None: - return self._default_max_parallel_requests_queue_size - - @default_max_parallel_requests_queue_size.setter - def default_max_parallel_requests_queue_size(self, queue_size: int | None) -> None: - self._default_max_parallel_requests_queue_size = validate_max_parallel_requests_queue_size(queue_size) - InitalizeCachedClient.apply_default_max_parallel_requests_queue_size( - litellm_router_instance=self, queue_size=self._default_max_parallel_requests_queue_size - ) - def update_settings(self, **kwargs): """ Update the router settings. diff --git a/litellm/router_utils/client_initalization_utils.py b/litellm/router_utils/client_initalization_utils.py index be5f71a4e70..55b4c071cb0 100644 --- a/litellm/router_utils/client_initalization_utils.py +++ b/litellm/router_utils/client_initalization_utils.py @@ -1,11 +1,8 @@ -import asyncio -import time from types import TracebackType from typing import TYPE_CHECKING, Any, Final -from litellm._logging import verbose_router_logger from litellm.exceptions import RateLimitError, RateLimitErrorCategory, RateLimitType -from litellm.types.router import RouterErrors, validate_max_parallel_requests_queue_size +from litellm.types.router import RouterErrors from litellm.utils import calculate_max_parallel_requests if TYPE_CHECKING: @@ -16,71 +13,41 @@ else: LitellmRouter = Any -class DeploymentSemaphore: - """A deployment's max_parallel_requests slots. ``queue_size=None`` parks callers without bound, like a plain - ``asyncio.Semaphore``; otherwise a caller arriving while all slots are busy and ``queue_size`` callers already - wait gets a 429 instead of being parked.""" +class MaxParallelRequestsLimit: + """A deployment's max_parallel_requests slots. A caller arriving while every slot is in use gets a 429 instead + of waiting for one to free up.""" - def __init__(self, max_parallel_requests: int, model_id: str, model_group: str, queue_size: int | None) -> None: - self._slots: Final = asyncio.Semaphore(max_parallel_requests) + def __init__(self, max_parallel_requests: int, model_id: str, model_group: str) -> None: self.max_parallel_requests: Final = max_parallel_requests self.model_id: Final = model_id self.model_group: Final = model_group - self.queue_size = validate_max_parallel_requests_queue_size(queue_size) - self.waiting = 0 + self.in_flight = 0 - def locked(self) -> bool: - return self._slots.locked() + def __enter__(self) -> None: + self.acquire() - def release(self) -> None: - self._slots.release() - - async def __aenter__(self) -> None: - await self.acquire() - - async def __aexit__( + def __exit__( self, exc_type: type[BaseException] | None, exc: BaseException | None, tb: TracebackType | None ) -> None: - self._slots.release() + self.release() - async def acquire(self) -> bool: - if not self._slots.locked(): - return await self._slots.acquire() - if self.queue_size is not None and self.waiting >= self.queue_size: + def acquire(self) -> None: + if self.in_flight >= self.max_parallel_requests: raise RateLimitError( message=( - f"{RouterErrors.max_parallel_requests_queue_full.value} Deployment model_group={self.model_group}, " - f"id={self.model_id} has all max_parallel_requests={self.max_parallel_requests} slots in use and " - f"{self.waiting} requests already waiting, which is its max_parallel_requests_queue_size=" - f"{self.queue_size}. Raise max_parallel_requests or max_parallel_requests_queue_size for this " - "deployment, or unset max_parallel_requests_queue_size to queue without a bound" + f"{RouterErrors.max_parallel_requests_exceeded.value} Deployment model_group={self.model_group}, " + f"id={self.model_id} already has max_parallel_requests={self.max_parallel_requests} requests in " + "flight. Raise max_parallel_requests (or the rpm/tpm it is derived from) for this deployment" ), llm_provider="", model=self.model_group, category=RateLimitErrorCategory.LITELLM_RATE_LIMIT, rate_limit_type=RateLimitType.CONCURRENT_REQUESTS, ) - self.waiting += 1 - queued_at: Final = time.perf_counter() - verbose_router_logger.debug( - "Deployment model_group=%s, id=%s has all max_parallel_requests=%s slots in use, request queued " - "(waiting=%s, max_parallel_requests_queue_size=%s)", - self.model_group, - self.model_id, - self.max_parallel_requests, - self.waiting, - self.queue_size, - ) - try: - return await self._slots.acquire() - finally: - self.waiting -= 1 - verbose_router_logger.debug( - "Deployment model_group=%s, id=%s request left the max_parallel_requests queue after %.1f ms", - self.model_group, - self.model_id, - (time.perf_counter() - queued_at) * 1000, - ) + self.in_flight += 1 + + def release(self) -> None: + self.in_flight -= 1 class InitalizeCachedClient: @@ -98,35 +65,14 @@ class InitalizeCachedClient: default_max_parallel_requests=litellm_router_instance.default_max_parallel_requests, ) if calculated_max_parallel_requests: - deployment_queue_size: Final = litellm_params.get("max_parallel_requests_queue_size", None) - semaphore: Final = DeploymentSemaphore( + limit: Final = MaxParallelRequestsLimit( max_parallel_requests=calculated_max_parallel_requests, model_id=model_id, model_group=model.get("model_name", ""), - queue_size=( - deployment_queue_size - if deployment_queue_size is not None - else litellm_router_instance.default_max_parallel_requests_queue_size - ), ) cache_key: Final = f"{model_id}_max_parallel_requests_client" litellm_router_instance.cache.set_cache( key=cache_key, - value=semaphore, + value=limit, local_only=True, ) - - @staticmethod - def apply_default_max_parallel_requests_queue_size( - litellm_router_instance: LitellmRouter, queue_size: int | None - ) -> None: - inheriting_semaphores: Final = ( - litellm_router_instance.cache.get_cache( - key=f"{model['model_info']['id']}_max_parallel_requests_client", local_only=True - ) - for model in litellm_router_instance.model_list - if model["litellm_params"].get("max_parallel_requests_queue_size") is None - ) - for semaphore in inheriting_semaphores: - if isinstance(semaphore, DeploymentSemaphore): - semaphore.queue_size = queue_size diff --git a/litellm/types/management_endpoints/router_settings_endpoints.py b/litellm/types/management_endpoints/router_settings_endpoints.py index fe715e45b2f..cef180b202a 100644 --- a/litellm/types/management_endpoints/router_settings_endpoints.py +++ b/litellm/types/management_endpoints/router_settings_endpoints.py @@ -244,17 +244,6 @@ ROUTER_SETTINGS_FIELDS: Final[list[RouterSettingsField]] = [ field_default=None, ui_field_name="Max Parallel Requests", ), - RouterSettingsField( - field_name="default_max_parallel_requests_queue_size", - field_type="Integer", - field_value=None, - field_description=( - "Default cap on how many requests may wait for a deployment's max_parallel_requests slot before " - "further requests get a 429. Unset queues without a bound" - ), - field_default=None, - ui_field_name="Max Parallel Requests Queue Size", - ), RouterSettingsField( field_name="enable_tag_filtering", field_type="Boolean", diff --git a/litellm/types/router.py b/litellm/types/router.py index 848dd28aaac..29f3c3681e0 100644 --- a/litellm/types/router.py +++ b/litellm/types/router.py @@ -6,10 +6,10 @@ import datetime import enum from collections.abc import Mapping, Sequence from dataclasses import dataclass -from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Final, Generic, Literal, TypeVar, get_type_hints +from typing import TYPE_CHECKING, Any, ClassVar, Final, Generic, Literal, TypeVar, get_type_hints import httpx -from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, field_validator, model_validator +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from typing_extensions import Protocol, ReadOnly, Required, TypedDict, runtime_checkable from litellm._logging import verbose_logger @@ -314,14 +314,6 @@ class CredentialLiteLLMParams(BaseModel): _RESERVED_INIT_KEYS: Final = frozenset({"self", "params", "__class__"}) -MaxParallelRequestsQueueSize = Annotated[int, Field(strict=True, ge=0)] -_MAX_PARALLEL_REQUESTS_QUEUE_SIZE_ADAPTER: Final = TypeAdapter(MaxParallelRequestsQueueSize | None) - - -def validate_max_parallel_requests_queue_size(value: object) -> int | None: - return _MAX_PARALLEL_REQUESTS_QUEUE_SIZE_ADAPTER.validate_python(value) - - class GenericLiteLLMParams(CredentialLiteLLMParams, CustomPricingLiteLLMParams): """ LiteLLM Params without 'model' arg (used across completion / assistants api) @@ -332,7 +324,6 @@ class GenericLiteLLMParams(CredentialLiteLLMParams, CustomPricingLiteLLMParams): rpm: int | None = None itpm: int | None = None otpm: int | None = None - max_parallel_requests_queue_size: MaxParallelRequestsQueueSize | None = None timeout: float | str | httpx.Timeout | None = None # if str, pass in as os.environ/ stream_timeout: float | str | None = None # timeout when making stream=True calls, if str, pass in as os.environ/ max_retries: int | None = None @@ -506,7 +497,6 @@ class LiteLLMParamsTypedDict(TypedDict, total=False): order: int | None weight: int | None max_parallel_requests: int | None - max_parallel_requests_queue_size: ReadOnly[MaxParallelRequestsQueueSize | None] api_key: str | None api_base: str | None api_version: str | None @@ -657,7 +647,7 @@ class RouterErrors(enum.Enum): """ user_defined_ratelimit_error = "Deployment over user-defined ratelimit." - max_parallel_requests_queue_full = "Deployment max_parallel_requests queue is full." + max_parallel_requests_exceeded = "Deployment has all max_parallel_requests slots in use." no_deployments_available = "No deployments available for selected model" all_deployments_in_cooldown = "All deployments for selected model are in cooldown" no_deployments_with_tag_routing = "Not allowed to access model due to tags configuration" diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 8f902f34548..aaa16fd2d44 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -3841,7 +3841,6 @@ all_litellm_params = ( "itpm", "otpm", "max_parallel_requests", - "max_parallel_requests_queue_size", "input_cost_per_token", "output_cost_per_token", "input_cost_per_second", diff --git a/tests/code_coverage_tests/router_code_coverage.py b/tests/code_coverage_tests/router_code_coverage.py index 582977d613b..a11f015743b 100644 --- a/tests/code_coverage_tests/router_code_coverage.py +++ b/tests/code_coverage_tests/router_code_coverage.py @@ -88,7 +88,6 @@ ignored_function_names = [ "_resolve_claude_code_session_router", # Tested through Claude Code session routing in test_router.py "_get_claude_code_session_router_binding", # Tested through the two-worker session routing test in test_router.py "_apply_updated_routing_strategy_args", # Tested via update_settings in test_lowest_latency.py (file lacks "router" in name) - "default_max_parallel_requests_queue_size", ] diff --git a/tests/local_testing/test_router_max_parallel_requests.py b/tests/local_testing/test_router_max_parallel_requests.py index 65602c968bc..051c69c9322 100644 --- a/tests/local_testing/test_router_max_parallel_requests.py +++ b/tests/local_testing/test_router_max_parallel_requests.py @@ -11,6 +11,7 @@ import pytest from typing import Optional import litellm +from litellm.router_utils.client_initalization_utils import MaxParallelRequestsLimit from litellm.utils import calculate_max_parallel_requests """ @@ -93,26 +94,26 @@ def test_setting_mpr_limits_per_model( default_max_parallel_requests=default_max_parallel_requests, ) - mpr_client: Optional[asyncio.Semaphore] = router._get_client( + mpr_client: Optional[MaxParallelRequestsLimit] = router._get_client( deployment=deployment, kwargs={}, client_type="max_parallel_requests", ) if max_parallel_requests is not None: - assert max_parallel_requests == mpr_client._value + assert max_parallel_requests == mpr_client.max_parallel_requests elif rpm is not None: - assert rpm == mpr_client._value + assert rpm == mpr_client.max_parallel_requests elif tpm is not None: calculated_rpm = int(tpm / 1000 * 6) if calculated_rpm == 0: calculated_rpm = 1 print( - f"test calculated_rpm: {calculated_rpm}, calculated_max_parallel_requests={mpr_client._value}" + f"test calculated_rpm: {calculated_rpm}, calculated_max_parallel_requests={mpr_client.max_parallel_requests}" ) - assert calculated_rpm == mpr_client._value + assert calculated_rpm == mpr_client.max_parallel_requests elif default_max_parallel_requests is not None: - assert mpr_client._value == default_max_parallel_requests + assert mpr_client.max_parallel_requests == default_max_parallel_requests else: assert mpr_client is None diff --git a/tests/test_litellm/llms/anthropic/test_anthropic_prompt_cache_prediction.py b/tests/test_litellm/llms/anthropic/test_anthropic_prompt_cache_prediction.py index c13217a0d46..2b36866a1a0 100644 --- a/tests/test_litellm/llms/anthropic/test_anthropic_prompt_cache_prediction.py +++ b/tests/test_litellm/llms/anthropic/test_anthropic_prompt_cache_prediction.py @@ -181,15 +181,6 @@ async def test_environment_credential_matches_native_count_and_observed_scope( assert observed.scope == cache_scope(_CALLER, _DEPLOYMENT, target.api_key, target.model) -def test_deployment_concurrency_knobs_keep_native_prediction_supported() -> None: - target: Final = resolve_prediction_target(LiteLLM_Params( - model=f"anthropic/{_MODEL}", api_key=_KEY, api_base="https://api.anthropic.com", - max_parallel_requests=1, max_parallel_requests_queue_size=0, - )) - assert isinstance(target, NativePredictionTarget) - assert (target.model, target.api_key) == (_MODEL, _KEY) - - @pytest.mark.parametrize("inline_key", [None, _KEY]) @pytest.mark.asyncio async def test_named_credential_is_explicitly_unsupported_before_count( diff --git a/tests/test_litellm/proxy/test_proxy_server.py b/tests/test_litellm/proxy/test_proxy_server.py index 5a2e76039e8..41c4956dba6 100644 --- a/tests/test_litellm/proxy/test_proxy_server.py +++ b/tests/test_litellm/proxy/test_proxy_server.py @@ -5051,39 +5051,6 @@ async def test_add_router_settings_from_db_config_empty_db_list_still_clears_unc assert combined_settings["num_retries"] == 1 -@pytest.mark.asyncio -async def test_add_router_settings_from_db_config_null_queue_size_reaches_router(): - """A cleared Admin UI field is stored as null. The reload must hand that None to the - router so a config.yaml bound is lifted, while an unrelated null still falls back to - the config value.""" - from unittest.mock import AsyncMock, MagicMock - - from litellm.proxy.proxy_server import ProxyConfig - - proxy_config = ProxyConfig() - mock_router = MagicMock() - mock_router.update_settings = MagicMock() - - config_data = {"router_settings": {"default_max_parallel_requests_queue_size": 2, "num_retries": 1}} - - mock_db_config = MagicMock() - mock_db_config.param_value = {"default_max_parallel_requests_queue_size": None, "num_retries": None} - - mock_prisma_client = MagicMock() - mock_prisma_client.db.litellm_config.find_first = AsyncMock(return_value=mock_db_config) - - await proxy_config._add_router_settings_from_db_config( - config_data=config_data, - llm_router=mock_router, - prisma_client=mock_prisma_client, - ) - - combined_settings = mock_router.update_settings.call_args.kwargs - assert "default_max_parallel_requests_queue_size" in combined_settings - assert combined_settings["default_max_parallel_requests_queue_size"] is None - assert combined_settings["num_retries"] == 1 - - @pytest.mark.asyncio async def test_add_router_settings_from_db_config_edge_cases(): """ @@ -9367,51 +9334,6 @@ def test_update_config_litellm_settings_request_wins_for_non_callback_keys( restore() -def test_update_config_router_settings_null_clears_max_parallel_requests_queue_size( - _update_config_setup, -): - """Clearing the Admin UI field sends null. The stored row must hold null so the - reload hands None to the router and queueing becomes unbounded again, while an - unrelated null is still dropped rather than persisted.""" - client, prisma, restore = _update_config_setup( - initial_rows={ - "router_settings": {"default_max_parallel_requests_queue_size": 3, "num_retries": 2}, - } - ) - try: - resp = client.post( - "/config/update", - json={"router_settings": {"default_max_parallel_requests_queue_size": None, "timeout": None}}, - ) - assert resp.status_code == 200 - stored = prisma.db.litellm_config.rows["router_settings"] - assert "default_max_parallel_requests_queue_size" in stored - assert stored["default_max_parallel_requests_queue_size"] is None - assert stored["num_retries"] == 2 - assert "timeout" not in stored - finally: - restore() - - -@pytest.mark.parametrize("invalid_queue_size", [-1, 2.5, "3"]) -def test_update_config_rejects_invalid_max_parallel_requests_queue_size_before_persisting( - _update_config_setup, invalid_queue_size -): - client, prisma, restore = _update_config_setup( - initial_rows={"router_settings": {"default_max_parallel_requests_queue_size": 3}}, - ) - try: - resp = client.post( - "/config/update", - json={"router_settings": {"default_max_parallel_requests_queue_size": invalid_queue_size}}, - ) - assert resp.status_code == 400 - assert "default_max_parallel_requests_queue_size" in resp.json()["error"]["message"] - assert prisma.db.litellm_config.rows["router_settings"] == {"default_max_parallel_requests_queue_size": 3} - finally: - restore() - - def test_update_config_success_callback_normalizes_existing_mixed_case( _update_config_setup, ): diff --git a/tests/test_litellm/router_utils/test_client_initalization_utils.py b/tests/test_litellm/router_utils/test_client_initalization_utils.py index a6626d2f975..6f9a7b730ac 100644 --- a/tests/test_litellm/router_utils/test_client_initalization_utils.py +++ b/tests/test_litellm/router_utils/test_client_initalization_utils.py @@ -2,220 +2,124 @@ import asyncio from typing import Final import pytest -from pydantic import ValidationError import litellm from litellm import Router -from litellm.router_utils.client_initalization_utils import DeploymentSemaphore +from litellm.router_utils.client_initalization_utils import MaxParallelRequestsLimit -def _semaphore(queue_size: int | None, max_parallel_requests: int = 1) -> DeploymentSemaphore: - return DeploymentSemaphore( - max_parallel_requests=max_parallel_requests, - model_id="deployment-1", - model_group="gpt-5.6", - queue_size=queue_size, +def _limit(max_parallel_requests: int = 1) -> MaxParallelRequestsLimit: + return MaxParallelRequestsLimit( + max_parallel_requests=max_parallel_requests, model_id="deployment-1", model_group="gpt-5.6" ) -async def _hold(semaphore: DeploymentSemaphore, release: asyncio.Event) -> str: - async with semaphore: +async def _hold(limit: MaxParallelRequestsLimit, release: asyncio.Event) -> str: + with limit: await release.wait() return "ok" -async def _expect_rejection(semaphore: DeploymentSemaphore) -> litellm.RateLimitError: +def _expect_rejection(limit: MaxParallelRequestsLimit) -> litellm.RateLimitError: with pytest.raises(litellm.RateLimitError) as excinfo: - await asyncio.wait_for(semaphore.acquire(), timeout=1) + limit.acquire() return excinfo.value @pytest.mark.asyncio -async def test_queue_full_rejects_new_caller_while_queued_callers_still_complete(): - semaphore: Final = _semaphore(queue_size=2) +async def test_request_arriving_while_every_slot_is_in_use_gets_429_without_waiting(): + limit: Final = _limit(max_parallel_requests=2) release: Final = asyncio.Event() - holder: Final = asyncio.create_task(_hold(semaphore, release)) + holders: Final = [asyncio.create_task(_hold(limit, release)) for _ in range(2)] await asyncio.sleep(0) - queued: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(2)] - await asyncio.sleep(0) - assert semaphore.locked() and semaphore.waiting == 2 + assert limit.in_flight == 2 - rejection: Final = await _expect_rejection(semaphore) + rejection: Final = _expect_rejection(limit) assert rejection.status_code == 429 assert "deployment-1" in rejection.message assert "gpt-5.6" in rejection.message - assert "max_parallel_requests=1" in rejection.message - assert "max_parallel_requests_queue_size=2" in rejection.message - assert semaphore.waiting == 2 - - release.set() - assert await asyncio.wait_for(asyncio.gather(holder, *queued), timeout=2) == ["ok", "ok", "ok"] - assert semaphore.waiting == 0 - assert not semaphore.locked() - - -@pytest.mark.asyncio -async def test_zero_queue_size_rejects_as_soon_as_every_slot_is_busy(): - semaphore: Final = _semaphore(queue_size=0, max_parallel_requests=2) - release: Final = asyncio.Event() - holders: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(2)] - await asyncio.sleep(0) - - await _expect_rejection(semaphore) - assert semaphore.waiting == 0 + assert "max_parallel_requests=2" in rejection.message + assert limit.in_flight == 2 release.set() assert await asyncio.wait_for(asyncio.gather(*holders), timeout=2) == ["ok", "ok"] + assert limit.in_flight == 0 + with limit: + assert limit.in_flight == 1 + assert limit.in_flight == 0 @pytest.mark.asyncio -async def test_unset_queue_size_parks_every_caller_until_a_slot_frees(): - semaphore: Final = _semaphore(queue_size=None) +async def test_burst_over_the_cap_admits_exactly_max_parallel_requests_and_rejects_the_rest(): + limit: Final = _limit(max_parallel_requests=3) release: Final = asyncio.Event() - callers: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(50)] - await asyncio.sleep(0) - assert semaphore.waiting == 49 + async def attempt() -> str: + try: + return await _hold(limit, release) + except litellm.RateLimitError as e: + return f"rejected:{e.status_code}" + + callers: Final = [asyncio.create_task(attempt()) for _ in range(10)] + await asyncio.sleep(0) + assert limit.in_flight == 3 release.set() - assert await asyncio.wait_for(asyncio.gather(*callers), timeout=2) == ["ok"] * 50 - assert semaphore.waiting == 0 + outcomes: Final = await asyncio.wait_for(asyncio.gather(*callers), timeout=2) + assert outcomes.count("ok") == 3 + assert outcomes.count("rejected:429") == 7 + assert limit.in_flight == 0 -@pytest.mark.asyncio -async def test_cancelled_waiter_gives_its_queue_slot_back(): - semaphore: Final = _semaphore(queue_size=1) - release: Final = asyncio.Event() - holder: Final = asyncio.create_task(_hold(semaphore, release)) - await asyncio.sleep(0) - cancelled: Final = asyncio.create_task(_hold(semaphore, release)) - await asyncio.sleep(0) - assert semaphore.waiting == 1 - - cancelled.cancel() - with pytest.raises(asyncio.CancelledError): - await cancelled - assert semaphore.waiting == 0 - - replacement: Final = asyncio.create_task(_hold(semaphore, release)) - await asyncio.sleep(0) - assert semaphore.waiting == 1 - release.set() - assert await asyncio.wait_for(asyncio.gather(holder, replacement), timeout=2) == ["ok", "ok"] +def test_slot_is_released_when_the_held_call_raises(): + limit: Final = _limit() + with pytest.raises(RuntimeError): + with limit: + raise RuntimeError("provider blew up") + assert limit.in_flight == 0 + with limit: + assert limit.in_flight == 1 -def _router_semaphore(router: Router, model_name: str) -> DeploymentSemaphore: +def _router_limit(router: Router, model_name: str) -> MaxParallelRequestsLimit: deployment: Final = router.get_deployment_by_model_group_name(model_group_name=model_name) assert deployment is not None - client: Final = router._get_client(deployment=deployment.model_dump(), kwargs={}, client_type="max_parallel_requests") - assert isinstance(client, DeploymentSemaphore) + client: Final = router._get_client( + deployment=deployment.model_dump(), kwargs={}, client_type="max_parallel_requests" + ) + assert isinstance(client, MaxParallelRequestsLimit) return client -@pytest.mark.parametrize("invalid_queue_size", [-1, 2.5, True, "3"]) -def test_invalid_queue_sizes_are_rejected_instead_of_coerced(invalid_queue_size: object): - """A negative bound would reject every busy request and a fraction would be truncated, so - neither may reach a semaphore, the router default, or a live update of that default.""" - with pytest.raises(ValidationError): - _semaphore(queue_size=invalid_queue_size) - model_list: Final = [{"model_name": "gpt-5.6", "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1}}] - with pytest.raises(ValidationError): - Router(model_list=model_list, default_max_parallel_requests_queue_size=invalid_queue_size) - with pytest.raises(ValidationError): - Router( - model_list=[ - { - "model_name": "gpt-5.6", - "litellm_params": { - "model": "openai/gpt-5.6", - "rpm": 1, - "max_parallel_requests_queue_size": invalid_queue_size, - }, - } - ] - ) - - router: Final = Router(model_list=model_list, default_max_parallel_requests_queue_size=4) - semaphore: Final = _router_semaphore(router, "gpt-5.6") - with pytest.raises(ValidationError): - router.update_settings(default_max_parallel_requests_queue_size=invalid_queue_size) - assert router.default_max_parallel_requests_queue_size == 4 - assert semaphore.queue_size == 4 - - +@pytest.mark.parametrize( + ("litellm_params", "expected_cap"), + [ + ({"max_parallel_requests": 2, "rpm": 7, "tpm": 100_000}, 2), + ({"rpm": 7, "tpm": 100_000}, 7), + ({"tpm": 100_000}, 600), + ({"tpm": 100}, 1), + ], +) @pytest.mark.asyncio -async def test_deployment_queue_size_overrides_router_default_and_zero_is_honored(): +async def test_router_deployment_rejects_past_its_derived_cap(litellm_params: dict[str, int], expected_cap: int): router: Final = Router( - model_list=[ - {"model_name": "inherits-default", "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1}}, - { - "model_name": "no-queue", - "litellm_params": {"model": "openai/gpt-5.6", "tpm": 100, "max_parallel_requests_queue_size": 0}, - }, - ], - default_max_parallel_requests_queue_size=1, + model_list=[{"model_name": "gpt-5.6", "litellm_params": {"model": "openai/gpt-5.6", **litellm_params}}] ) + limit: Final = _router_limit(router, "gpt-5.6") + assert limit.max_parallel_requests == expected_cap release: Final = asyncio.Event() - - inherits: Final = _router_semaphore(router, "inherits-default") - inherits_holder: Final = asyncio.create_task(_hold(inherits, release)) + holders: Final = [asyncio.create_task(_hold(limit, release)) for _ in range(expected_cap)] await asyncio.sleep(0) - inherits_waiter: Final = asyncio.create_task(_hold(inherits, release)) - await asyncio.sleep(0) - assert "max_parallel_requests_queue_size=1" in (await _expect_rejection(inherits)).message - - no_queue: Final = _router_semaphore(router, "no-queue") - no_queue_holder: Final = asyncio.create_task(_hold(no_queue, release)) - await asyncio.sleep(0) - assert "max_parallel_requests_queue_size=0" in (await _expect_rejection(no_queue)).message - + assert limit.in_flight == expected_cap + assert f"max_parallel_requests={expected_cap}" in _expect_rejection(limit).message release.set() - await asyncio.wait_for(asyncio.gather(inherits_holder, inherits_waiter, no_queue_holder), timeout=2) + assert await asyncio.wait_for(asyncio.gather(*holders), timeout=2) == ["ok"] * expected_cap -@pytest.mark.asyncio -async def test_router_without_queue_size_keeps_unbounded_queueing(): - router: Final = Router( - model_list=[{"model_name": "gpt-5.6", "litellm_params": {"model": "openai/gpt-5.6", "max_parallel_requests": 1}}] +def test_router_without_any_concurrency_setting_has_no_limit(): + router: Final = Router(model_list=[{"model_name": "gpt-5.6", "litellm_params": {"model": "openai/gpt-5.6"}}]) + deployment: Final = router.get_deployment_by_model_group_name(model_group_name="gpt-5.6") + assert deployment is not None + assert ( + router._get_client(deployment=deployment.model_dump(), kwargs={}, client_type="max_parallel_requests") is None ) - semaphore: Final = _router_semaphore(router, "gpt-5.6") - release: Final = asyncio.Event() - callers: Final = [asyncio.create_task(_hold(semaphore, release)) for _ in range(20)] - await asyncio.sleep(0) - assert semaphore.waiting == 19 - release.set() - assert await asyncio.wait_for(asyncio.gather(*callers), timeout=2) == ["ok"] * 20 - - -@pytest.mark.asyncio -async def test_update_settings_applies_default_queue_size_to_live_semaphores_without_an_override(): - router: Final = Router( - model_list=[ - {"model_name": "inherits-default", "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1}}, - { - "model_name": "pinned", - "litellm_params": {"model": "openai/gpt-5.6", "rpm": 1, "max_parallel_requests_queue_size": 5}, - }, - ], - ) - inherits: Final = _router_semaphore(router, "inherits-default") - pinned: Final = _router_semaphore(router, "pinned") - assert router.get_settings()["default_max_parallel_requests_queue_size"] is None - - router.update_settings(default_max_parallel_requests_queue_size=0) - assert router.get_settings()["default_max_parallel_requests_queue_size"] == 0 - assert (inherits.queue_size, pinned.queue_size) == (0, 5) - - release: Final = asyncio.Event() - holder: Final = asyncio.create_task(_hold(inherits, release)) - await asyncio.sleep(0) - assert "max_parallel_requests_queue_size=0" in (await _expect_rejection(inherits)).message - - router.update_settings(default_max_parallel_requests_queue_size=None) - assert (inherits.queue_size, pinned.queue_size) == (None, 5) - waiter: Final = asyncio.create_task(_hold(inherits, release)) - await asyncio.sleep(0) - assert inherits.waiting == 1 - - release.set() - assert await asyncio.wait_for(asyncio.gather(holder, waiter), timeout=2) == ["ok", "ok"] diff --git a/tests/test_litellm/test_router.py b/tests/test_litellm/test_router.py index a674f767dde..26f9803a022 100644 --- a/tests/test_litellm/test_router.py +++ b/tests/test_litellm/test_router.py @@ -47,7 +47,7 @@ from litellm.router import ( _is_retriable_anthropic_status, ) from litellm.router_strategy import simple_shuffle -from litellm.router_utils.client_initalization_utils import DeploymentSemaphore +from litellm.router_utils.client_initalization_utils import MaxParallelRequestsLimit from litellm.router_utils.cooldown_handlers import _async_get_cooldown_deployments from litellm.types.llms.openai import ChatCompletionRequest from litellm.types.router import Deployment, DeploymentTypedDict, LiteLLM_Params, ModelInfo, PreRoutingHookResponse, RetryPolicy @@ -1521,8 +1521,8 @@ async def test_router_ageneric_api_call_with_fallbacks_helper(): }, } - mock_semaphore = DeploymentSemaphore( - max_parallel_requests=1, model_id="deployment-1", model_group="gpt-3.5-turbo", queue_size=None + mock_semaphore = MaxParallelRequestsLimit( + max_parallel_requests=1, model_id="deployment-1", model_group="gpt-3.5-turbo" ) with patch.object( @@ -15968,7 +15968,7 @@ def _max_parallel_router(max_parallel_requests: int) -> Router: @pytest.mark.asyncio @pytest.mark.parametrize("stream", [False, True]) -async def test_router_max_parallel_requests_bounds_in_flight_upstream_calls( +async def test_router_max_parallel_requests_admits_the_cap_and_rejects_the_rest_with_429( monkeypatch: pytest.MonkeyPatch, stream: bool ): monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) @@ -15994,24 +15994,33 @@ async def test_router_max_parallel_requests_bounds_in_flight_upstream_calls( }, ) - async def one_call() -> None: - response = await router.acompletion( - model="gpt-5.6", messages=[{"role": "user", "content": "hi"}], stream=stream - ) + async def one_call() -> str: + try: + response = await router.acompletion( + model="gpt-5.6", messages=[{"role": "user", "content": "hi"}], stream=stream + ) + except litellm.RateLimitError as e: + return f"rejected:{e.status_code}" if stream: async for _ in response: pass + return "ok" with respx.mock(assert_all_called=True) as respx_mock: - respx_mock.post("https://max-parallel.local/v1/chat/completions").mock(side_effect=upstream) - await asyncio.wait_for(asyncio.gather(*(one_call() for _ in range(10))), timeout=10) + route: Final = respx_mock.post("https://max-parallel.local/v1/chat/completions").mock(side_effect=upstream) + outcomes: Final = await asyncio.wait_for(asyncio.gather(*(one_call() for _ in range(10))), timeout=10) - assert tracker.peak <= 2 + assert outcomes.count("ok") == 2 + assert outcomes.count("rejected:429") == 8 + assert route.call_count == 2 + assert tracker.peak == 2 assert tracker.current == 0 @pytest.mark.asyncio -async def test_router_max_parallel_requests_slot_released_when_stream_closed_early(monkeypatch: pytest.MonkeyPatch): +async def test_router_max_parallel_requests_slot_held_until_stream_closed_then_released( + monkeypatch: pytest.MonkeyPatch, +): monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) tracker: Final = _InFlightTracker() router: Final = _max_parallel_router(max_parallel_requests=1) @@ -16034,18 +16043,21 @@ async def test_router_max_parallel_requests_slot_released_when_stream_closed_ear async for _ in second: pass - second_task: Final = asyncio.create_task(second_call()) - await asyncio.sleep(0.05) assert tracker.current == 1 + with pytest.raises(litellm.RateLimitError) as while_streaming: + await second_call() + assert while_streaming.value.status_code == 429 await first.aclose() - await asyncio.wait_for(second_task, timeout=2) + await asyncio.wait_for(second_call(), timeout=2) assert tracker.peak == 1 assert tracker.current == 0 @pytest.mark.asyncio -async def test_router_max_parallel_requests_queue_size_turns_overflow_into_429(monkeypatch: pytest.MonkeyPatch): +async def test_router_max_parallel_requests_overflow_is_429_without_cooldown_or_provider_call( + monkeypatch: pytest.MonkeyPatch, +): monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) router: Final = Router( model_list=[ @@ -16056,9 +16068,8 @@ async def test_router_max_parallel_requests_queue_size_turns_overflow_into_429(m "api_key": "sk-fake", "api_base": "https://max-parallel.local/v1", "max_parallel_requests": 1, - "max_parallel_requests_queue_size": 1, }, - "model_info": {"id": "queue-bounded-deployment"}, + "model_info": {"id": "capped-deployment"}, }, { "model_name": "gpt-5.6", @@ -16067,7 +16078,7 @@ async def test_router_max_parallel_requests_queue_size_turns_overflow_into_429(m "api_key": "sk-fake", "api_base": "https://max-parallel-sibling.local/v1", }, - "model_info": {"id": "queue-sibling-deployment"}, + "model_info": {"id": "sibling-deployment"}, }, ], num_retries=0, @@ -16094,7 +16105,7 @@ async def test_router_max_parallel_requests_queue_size_turns_overflow_into_429(m results: Final = await asyncio.wait_for( asyncio.gather( *( - router.acompletion(model="queue-bounded-deployment", messages=[{"role": "user", "content": "hi"}]) + router.acompletion(model="capped-deployment", messages=[{"role": "user", "content": "hi"}]) for _ in range(3) ), return_exceptions=True, @@ -16103,19 +16114,18 @@ async def test_router_max_parallel_requests_queue_size_turns_overflow_into_429(m ) rejected: Final = [r for r in results if isinstance(r, BaseException)] - assert len(rejected) == 1 and len(results) == 3 - assert isinstance(rejected[0], litellm.RateLimitError) - assert rejected[0].status_code == 429 - assert "queue-bounded-deployment" in rejected[0].message - assert "max_parallel_requests_queue_size=1" in rejected[0].message - assert route.call_count == 2 + assert len(rejected) == 2 and len(results) == 3 + assert all(isinstance(r, litellm.RateLimitError) and r.status_code == 429 for r in rejected) + assert all("capped-deployment" in r.message and "max_parallel_requests=1" in r.message for r in rejected) + assert route.call_count == 1 assert sibling_route.call_count == 0 - assert all("max_parallel_requests_queue_size" not in call.request.content.decode() for call in route.calls) assert await _async_get_cooldown_deployments(litellm_router_instance=router, parent_otel_span=None) == [] @pytest.mark.asyncio -async def test_router_embedding_path_honors_max_parallel_requests_queue_size(monkeypatch: pytest.MonkeyPatch): +async def test_router_embedding_path_rejects_past_max_parallel_requests_without_orphan_coroutines( + monkeypatch: pytest.MonkeyPatch, +): monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) router: Final = Router( model_list=[ @@ -16127,10 +16137,9 @@ async def test_router_embedding_path_honors_max_parallel_requests_queue_size(mon "api_base": "https://max-parallel-embed.local/v1", "max_parallel_requests": 1, }, - "model_info": {"id": "embed-bounded-deployment"}, + "model_info": {"id": "embed-capped-deployment"}, } ], - default_max_parallel_requests_queue_size=1, num_retries=0, ) @@ -16159,15 +16168,15 @@ async def test_router_embedding_path_honors_max_parallel_requests_queue_size(mon gc.collect() rejected: Final = [r for r in results if isinstance(r, BaseException)] - assert len(rejected) == 1 and len(results) == 3 - assert isinstance(rejected[0], litellm.RateLimitError) and rejected[0].status_code == 429 - assert "embed-bounded-deployment" in rejected[0].message - assert route.call_count == 2 + assert len(rejected) == 2 and len(results) == 3 + assert all(isinstance(r, litellm.RateLimitError) and r.status_code == 429 for r in rejected) + assert all("embed-capped-deployment" in r.message for r in rejected) + assert route.call_count == 1 assert [str(w.message) for w in caught if "never awaited" in str(w.message)] == [] @pytest.mark.asyncio -async def test_router_max_parallel_requests_queue_overflow_takes_the_ordinary_429_fallback_path( +async def test_router_max_parallel_requests_overflow_takes_the_ordinary_429_fallback_path( monkeypatch: pytest.MonkeyPatch, ): monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) @@ -16180,9 +16189,8 @@ async def test_router_max_parallel_requests_queue_overflow_takes_the_ordinary_42 "api_key": "sk-fake", "api_base": "https://max-parallel-primary.local/v1", "max_parallel_requests": 1, - "max_parallel_requests_queue_size": 0, }, - "model_info": {"id": "queue-primary-deployment"}, + "model_info": {"id": "capped-primary-deployment"}, }, { "model_name": "gpt-5.6-fallback", @@ -16191,7 +16199,7 @@ async def test_router_max_parallel_requests_queue_overflow_takes_the_ordinary_42 "api_key": "sk-fake", "api_base": "https://max-parallel-fallback.local/v1", }, - "model_info": {"id": "queue-fallback-deployment"}, + "model_info": {"id": "fallback-deployment"}, }, ], fallbacks=[{"gpt-5.6": ["gpt-5.6-fallback"]}], @@ -16232,7 +16240,7 @@ async def test_router_max_parallel_requests_queue_overflow_takes_the_ordinary_42 @pytest.mark.asyncio -async def test_router_deployment_slot_rejects_once_queue_is_full_and_frees_slot_on_exit(): +async def test_router_deployment_slot_rejects_while_held_and_frees_slot_on_exit(): router: Final = Router( model_list=[ { @@ -16241,7 +16249,6 @@ async def test_router_deployment_slot_rejects_once_queue_is_full_and_frees_slot_ "model": "openai/gpt-5.6", "api_key": "sk-fake", "max_parallel_requests": 1, - "max_parallel_requests_queue_size": 0, }, "model_info": {"id": "slot-deployment"}, } diff --git a/ui/litellm-dashboard/src/components/router_settings/index.test.tsx b/ui/litellm-dashboard/src/components/router_settings/index.test.tsx index f2740bbd1e0..1875085231a 100644 --- a/ui/litellm-dashboard/src/components/router_settings/index.test.tsx +++ b/ui/litellm-dashboard/src/components/router_settings/index.test.tsx @@ -137,41 +137,6 @@ describe("RouterSettings", () => { ); }); - it("should save default_max_parallel_requests_queue_size as a number and an empty field as null", async () => { - vi.mocked(getCallbacksCall).mockResolvedValue({ - router_settings: { ...mockCallbacksResponse.router_settings, default_max_parallel_requests_queue_size: null }, - }); - const user = userEvent.setup(); - renderWithProviders(); - - await findStrategySelect(); - - const queueSize = await screen.findByRole("textbox", { name: /default_max_parallel_requests_queue_size/i }); - fireEvent.change(queueSize, { target: { value: "4" } }); - await user.click(screen.getByRole("button", { name: /save changes/i })); - - await waitFor(() => - expect(setCallbacksCall).toHaveBeenLastCalledWith( - "test-token", - expect.objectContaining({ - router_settings: expect.objectContaining({ default_max_parallel_requests_queue_size: 4 }), - }), - ), - ); - - fireEvent.change(queueSize, { target: { value: "" } }); - await user.click(screen.getByRole("button", { name: /save changes/i })); - - await waitFor(() => - expect(setCallbacksCall).toHaveBeenLastCalledWith( - "test-token", - expect.objectContaining({ - router_settings: expect.objectContaining({ default_max_parallel_requests_queue_size: null }), - }), - ), - ); - }); - it("should show a success notification after saving", async () => { const user = userEvent.setup(); renderWithProviders(); diff --git a/ui/litellm-dashboard/src/components/router_settings/index.tsx b/ui/litellm-dashboard/src/components/router_settings/index.tsx index 4170d48361d..53d35b81cec 100644 --- a/ui/litellm-dashboard/src/components/router_settings/index.tsx +++ b/ui/litellm-dashboard/src/components/router_settings/index.tsx @@ -86,15 +86,7 @@ const RouterSettings: React.FC = ({ accessToken, userRole, const router_settings = formValue.routerSettings; - const numberKeys = new Set([ - "allowed_fails", - "cooldown_time", - "num_retries", - "timeout", - "retry_after", - "default_max_parallel_requests_queue_size", - ]); - const unsettableNumberKeys = new Set(["default_max_parallel_requests_queue_size"]); + const numberKeys = new Set(["allowed_fails", "cooldown_time", "num_retries", "timeout", "retry_after"]); const jsonKeys = new Set(["model_group_alias"]); // retry_policy and model_group_retry_policy are owned by the Model Retry Settings tab; // routing_groups is owned by the Routing Groups tab. This page must not read or write them. @@ -108,7 +100,6 @@ const RouterSettings: React.FC = ({ accessToken, userRole, if (v.toLowerCase() === "null") return null; if (numberKeys.has(key)) { - if (v === "" && unsettableNumberKeys.has(key)) return null; const n = Number(v); return Number.isNaN(n) ? fallback : n; } diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 84ca93eccfe..872875cc535 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -30441,8 +30441,6 @@ export interface components { max_budget?: number | null; /** Max File Size Mb */ max_file_size_mb?: number | null; - /** Max Parallel Requests Queue Size */ - max_parallel_requests_queue_size?: number | null; /** Max Retries */ max_retries?: number | null; /** @@ -40895,8 +40893,6 @@ export interface components { max_budget?: number | null; /** Max File Size Mb */ max_file_size_mb?: number | null; - /** Max Parallel Requests Queue Size */ - max_parallel_requests_queue_size?: number | null; /** Max Retries */ max_retries?: number | null; /**