Tiny pass routing key to scheduler processes (#16839)
This commit is contained in:
@@ -282,3 +282,8 @@ class OpenAIServingBase(ABC):
|
||||
if label in self.allowed_custom_labels
|
||||
}
|
||||
return custom_labels
|
||||
|
||||
def extract_routing_key(self, raw_request):
|
||||
if raw_request is None:
|
||||
return None
|
||||
return raw_request.headers.get("x-smg-routing-key")
|
||||
|
||||
@@ -246,6 +246,7 @@ class OpenAIServingChat(OpenAIServingBase):
|
||||
extra_key=self._compute_extra_key(request),
|
||||
require_reasoning=self._get_reasoning_from_request(request),
|
||||
priority=request.priority,
|
||||
routing_key=self.extract_routing_key(raw_request),
|
||||
custom_labels=custom_labels,
|
||||
custom_logit_processor=request.custom_logit_processor,
|
||||
image_max_dynamic_patch=img_max_dynamic_patch,
|
||||
|
||||
@@ -121,6 +121,7 @@ class OpenAIServingCompletion(OpenAIServingBase):
|
||||
rid=request.rid,
|
||||
extra_key=self._compute_extra_key(request),
|
||||
priority=request.priority,
|
||||
routing_key=self.extract_routing_key(raw_request),
|
||||
custom_labels=custom_labels,
|
||||
custom_logit_processor=request.custom_logit_processor,
|
||||
)
|
||||
|
||||
@@ -130,6 +130,7 @@ class OpenAIServingEmbedding(OpenAIServingBase):
|
||||
**prompt_kwargs,
|
||||
rid=request.rid,
|
||||
priority=request.priority,
|
||||
routing_key=self.extract_routing_key(raw_request),
|
||||
dimensions=request.dimensions,
|
||||
)
|
||||
|
||||
|
||||
@@ -243,6 +243,9 @@ class GenerateReqInput(BaseReq, APIServingTimingMixin):
|
||||
# Extra key for classifying the request (e.g. cache_salt)
|
||||
extra_key: Optional[Union[List[str], str]] = None
|
||||
|
||||
# Routing key for routing-key schedule policy
|
||||
routing_key: Optional[str] = None
|
||||
|
||||
# Whether to disallow logging for this request (e.g. due to ZDR)
|
||||
no_logs: bool = False
|
||||
|
||||
@@ -740,6 +743,9 @@ class TokenizedGenerateReqInput(BaseReq):
|
||||
# Extra key for classifying the request (e.g. cache_salt)
|
||||
extra_key: Optional[str] = None
|
||||
|
||||
# Routing key for routing-key schedule policy
|
||||
routing_key: Optional[str] = None
|
||||
|
||||
# Whether to disallow logging for this request (e.g. due to ZDR)
|
||||
no_logs: bool = False
|
||||
|
||||
@@ -802,6 +808,8 @@ class EmbeddingReqInput(BaseReq, APIServingTimingMixin):
|
||||
is_cross_encoder_request: bool = False
|
||||
# Priority for the request
|
||||
priority: Optional[int] = None
|
||||
# Routing key for routing-key schedule policy
|
||||
routing_key: Optional[str] = None
|
||||
|
||||
# For background responses (OpenAI responses API)
|
||||
background: bool = False
|
||||
|
||||
@@ -521,6 +521,7 @@ class Req:
|
||||
priority: Optional[int] = None,
|
||||
metrics_collector: Optional[SchedulerMetricsCollector] = None,
|
||||
extra_key: Optional[str] = None,
|
||||
routing_key: Optional[str] = None,
|
||||
dimensions: Optional[int] = None,
|
||||
http_worker_ipc: Optional[str] = None,
|
||||
):
|
||||
@@ -580,6 +581,7 @@ class Req:
|
||||
|
||||
self.extra_key = extra_key
|
||||
self.lora_id = lora_id
|
||||
self.routing_key = routing_key
|
||||
|
||||
# Memory pool info
|
||||
self.req_pool_idx: Optional[int] = None
|
||||
|
||||
@@ -925,6 +925,7 @@ class TokenizerManager(TokenizerCommunicatorMixin, TokenizerManagerMultiItemMixi
|
||||
data_parallel_rank=obj.data_parallel_rank,
|
||||
priority=obj.priority,
|
||||
extra_key=obj.extra_key,
|
||||
routing_key=obj.routing_key,
|
||||
need_wait_for_image=obj.need_wait_for_image,
|
||||
num_items_assigned=obj.num_items_assigned,
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user