"mmdet/models/task_modules/samplers/random_sampler.py" did not exist on "3e89e2569a866545f404bfd4c11cc9bfb12cad68"
router.py 9.43 KB
Newer Older
1
from typing import Dict, List, Optional
2
3
4
5
6
7
8
9
10
11

from sglang_router_rs import PolicyType
from sglang_router_rs import Router as _Router


class Router:
    """
    A high-performance router for distributing requests across worker nodes.

    Args:
12
13
        worker_urls: List of URLs for worker nodes that will handle requests. Each URL should include
            the protocol, host, and port (e.g., ['http://worker1:8000', 'http://worker2:8000'])
14
15
16
        policy: Load balancing policy to use. Options:
            - PolicyType.Random: Randomly select workers
            - PolicyType.RoundRobin: Distribute requests in round-robin fashion
17
            - PolicyType.CacheAware: Distribute requests based on cache state and load balance
18
            - PolicyType.PowerOfTwo: Select best of two random workers based on load (PD mode only)
19
20
        host: Host address to bind the router server. Default: '127.0.0.1'
        port: Port number to bind the router server. Default: 3001
21
        worker_startup_timeout_secs: Timeout in seconds for worker startup. Default: 300
22
        worker_startup_check_interval: Interval in seconds between checks for worker initialization. Default: 10
23
24
25
        cache_threshold: Cache threshold (0.0-1.0) for cache-aware routing. Routes to cached worker
            if the match rate exceeds threshold, otherwise routes to the worker with the smallest
            tree. Default: 0.5
26
27
28
29
        balance_abs_threshold: Load balancing is triggered when (max_load - min_load) > abs_threshold
            AND max_load > min_load * rel_threshold. Otherwise, use cache aware. Default: 32
        balance_rel_threshold: Load balancing is triggered when (max_load - min_load) > abs_threshold
            AND max_load > min_load * rel_threshold. Otherwise, use cache aware. Default: 1.0001
30
31
        eviction_interval_secs: Interval in seconds between cache eviction operations in cache-aware
            routing. Default: 60
32
        max_payload_size: Maximum payload size in bytes. Default: 256MB
33
        max_tree_size: Maximum size of the approximation tree for cache-aware routing. Default: 2^24
34
35
36
37
        dp_aware: Enable data parallelism aware schedule. Default: False
        api_key: The api key used for the authorization with the worker.
            Useful when the dp aware scheduling strategy is enabled.
            Default: None
38
        log_dir: Directory to store log files. If None, logs are only output to console. Default: None
39
        log_level: Logging level. Options: 'debug', 'info', 'warning', 'error', 'critical'.
40
41
42
43
44
45
46
47
        service_discovery: Enable Kubernetes service discovery. When enabled, the router will
            automatically discover worker pods based on the selector. Default: False
        selector: Dictionary mapping of label keys to values for Kubernetes pod selection.
            Example: {"app": "sglang-worker"}. Default: {}
        service_discovery_port: Port to use for service discovery. The router will generate
            worker URLs using this port. Default: 80
        service_discovery_namespace: Kubernetes namespace to watch for pods. If not provided,
            watches pods across all namespaces (requires cluster-wide permissions). Default: None
48
49
50
51
        prefill_selector: Dictionary mapping of label keys to values for Kubernetes pod selection
            for prefill servers (PD mode only). Default: {}
        decode_selector: Dictionary mapping of label keys to values for Kubernetes pod selection
            for decode servers (PD mode only). Default: {}
52
53
        prometheus_port: Port to expose Prometheus metrics. Default: None
        prometheus_host: Host address to bind the Prometheus metrics server. Default: None
54
        pd_disaggregation: Enable PD (Prefill-Decode) disaggregated mode. Default: False
55
56
        prefill_urls: List of (url, bootstrap_port) tuples for prefill servers (PD mode only)
        decode_urls: List of URLs for decode servers (PD mode only)
57
58
59
60
        prefill_policy: Specific load balancing policy for prefill nodes (PD mode only).
            If not specified, uses the main policy. Default: None
        decode_policy: Specific load balancing policy for decode nodes (PD mode only).
            If not specified, uses the main policy. Default: None
61
62
63
        request_id_headers: List of HTTP headers to check for request IDs. If not specified,
            uses common defaults: ['x-request-id', 'x-correlation-id', 'x-trace-id', 'request-id'].
            Example: ['x-my-request-id', 'x-custom-trace-id']. Default: None
64
65
66
67
68
        bootstrap_port_annotation: Kubernetes annotation name for bootstrap port (PD mode).
            Default: 'sglang.ai/bootstrap-port'
        request_timeout_secs: Request timeout in seconds. Default: 600
        max_concurrent_requests: Maximum number of concurrent requests allowed for rate limiting. Default: 64
        cors_allowed_origins: List of allowed origins for CORS. Empty list allows all origins. Default: []
69
70
71
72
73
74
75
76
    """

    def __init__(
        self,
        worker_urls: List[str],
        policy: PolicyType = PolicyType.RoundRobin,
        host: str = "127.0.0.1",
        port: int = 3001,
77
        worker_startup_timeout_secs: int = 300,
78
        worker_startup_check_interval: int = 10,
79
        cache_threshold: float = 0.50,
80
81
        balance_abs_threshold: int = 32,
        balance_rel_threshold: float = 1.0001,
82
83
        eviction_interval_secs: int = 60,
        max_tree_size: int = 2**24,
84
        max_payload_size: int = 256 * 1024 * 1024,  # 256MB
85
86
        dp_aware: bool = False,
        api_key: Optional[str] = None,
87
        log_dir: Optional[str] = None,
88
        log_level: Optional[str] = None,
89
90
91
92
        service_discovery: bool = False,
        selector: Dict[str, str] = None,
        service_discovery_port: int = 80,
        service_discovery_namespace: Optional[str] = None,
93
94
        prefill_selector: Dict[str, str] = None,
        decode_selector: Dict[str, str] = None,
95
        bootstrap_port_annotation: str = "sglang.ai/bootstrap-port",
96
97
        prometheus_port: Optional[int] = None,
        prometheus_host: Optional[str] = None,
98
99
        request_timeout_secs: int = 600,
        request_id_headers: Optional[List[str]] = None,
100
        pd_disaggregation: bool = False,
101
102
        prefill_urls: Optional[List[tuple]] = None,
        decode_urls: Optional[List[str]] = None,
103
104
        prefill_policy: Optional[PolicyType] = None,
        decode_policy: Optional[PolicyType] = None,
105
106
        max_concurrent_requests: int = 64,
        cors_allowed_origins: List[str] = None,
107
108
109
110
111
112
113
114
115
116
117
        retry_max_retries: int = 3,
        retry_initial_backoff_ms: int = 100,
        retry_max_backoff_ms: int = 10_000,
        retry_backoff_multiplier: float = 2.0,
        retry_jitter_factor: float = 0.1,
        cb_failure_threshold: int = 5,
        cb_success_threshold: int = 2,
        cb_timeout_duration_secs: int = 30,
        cb_window_duration_secs: int = 60,
        disable_retries: bool = False,
        disable_circuit_breaker: bool = False,
118
    ):
119
120
        if selector is None:
            selector = {}
121
122
123
124
        if prefill_selector is None:
            prefill_selector = {}
        if decode_selector is None:
            decode_selector = {}
125
126
        if cors_allowed_origins is None:
            cors_allowed_origins = []
127

128
129
130
131
132
        self._router = _Router(
            worker_urls=worker_urls,
            policy=policy,
            host=host,
            port=port,
133
            worker_startup_timeout_secs=worker_startup_timeout_secs,
134
            worker_startup_check_interval=worker_startup_check_interval,
135
            cache_threshold=cache_threshold,
136
137
            balance_abs_threshold=balance_abs_threshold,
            balance_rel_threshold=balance_rel_threshold,
138
139
            eviction_interval_secs=eviction_interval_secs,
            max_tree_size=max_tree_size,
140
            max_payload_size=max_payload_size,
141
142
            dp_aware=dp_aware,
            api_key=api_key,
143
            log_dir=log_dir,
144
            log_level=log_level,
145
146
147
148
            service_discovery=service_discovery,
            selector=selector,
            service_discovery_port=service_discovery_port,
            service_discovery_namespace=service_discovery_namespace,
149
150
            prefill_selector=prefill_selector,
            decode_selector=decode_selector,
151
            bootstrap_port_annotation=bootstrap_port_annotation,
152
153
            prometheus_port=prometheus_port,
            prometheus_host=prometheus_host,
154
155
            request_timeout_secs=request_timeout_secs,
            request_id_headers=request_id_headers,
156
            pd_disaggregation=pd_disaggregation,
157
158
            prefill_urls=prefill_urls,
            decode_urls=decode_urls,
159
160
            prefill_policy=prefill_policy,
            decode_policy=decode_policy,
161
162
            max_concurrent_requests=max_concurrent_requests,
            cors_allowed_origins=cors_allowed_origins,
163
164
165
166
167
168
169
170
171
172
173
            retry_max_retries=retry_max_retries,
            retry_initial_backoff_ms=retry_initial_backoff_ms,
            retry_max_backoff_ms=retry_max_backoff_ms,
            retry_backoff_multiplier=retry_backoff_multiplier,
            retry_jitter_factor=retry_jitter_factor,
            cb_failure_threshold=cb_failure_threshold,
            cb_success_threshold=cb_success_threshold,
            cb_timeout_duration_secs=cb_timeout_duration_secs,
            cb_window_duration_secs=cb_window_duration_secs,
            disable_retries=disable_retries,
            disable_circuit_breaker=disable_circuit_breaker,
174
175
176
177
178
179
180
181
        )

    def start(self) -> None:
        """Start the router server.

        This method blocks until the server is shut down.
        """
        self._router.start()