launch_router.py 32.5 KB
Newer Older
1
2
import argparse
import dataclasses
3
import logging
4
import sys
5
from typing import Dict, List, Optional
6
7
8
9
10

from sglang_router import Router
from sglang_router_rs import PolicyType


11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def setup_logger():
    logger = logging.getLogger("router")
    logger.setLevel(logging.INFO)

    formatter = logging.Formatter(
        "[Router (Python)] %(asctime)s - %(levelname)s - %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S",
    )

    handler = logging.StreamHandler()
    handler.setFormatter(formatter)
    logger.addHandler(handler)

    return logger


27
28
29
@dataclasses.dataclass
class RouterArgs:
    # Worker configuration
30
    worker_urls: List[str] = dataclasses.field(default_factory=list)
31
32
33
    host: str = "127.0.0.1"
    port: int = 30000

34
    # PD-specific configuration
35
    pd_disaggregation: bool = False  # Enable PD disaggregated mode
36
37
38
39
40
    prefill_urls: List[tuple] = dataclasses.field(
        default_factory=list
    )  # List of (url, bootstrap_port)
    decode_urls: List[str] = dataclasses.field(default_factory=list)

41
42
    # Routing policy
    policy: str = "cache_aware"
43
44
    prefill_policy: Optional[str] = None  # Specific policy for prefill nodes in PD mode
    decode_policy: Optional[str] = None  # Specific policy for decode nodes in PD mode
45
46
47
48
49
50
51
52
    worker_startup_timeout_secs: int = 600
    worker_startup_check_interval: int = 30
    cache_threshold: float = 0.3
    balance_abs_threshold: int = 64
    balance_rel_threshold: float = 1.5
    eviction_interval: int = 120
    max_tree_size: int = 2**26
    max_payload_size: int = 512 * 1024 * 1024  # 512MB default for large batches
53
54
    dp_aware: bool = False
    api_key: Optional[str] = None
55
    log_dir: Optional[str] = None
56
    log_level: Optional[str] = None
57
58
59
60
61
    # Service discovery configuration
    service_discovery: bool = False
    selector: Dict[str, str] = dataclasses.field(default_factory=dict)
    service_discovery_port: int = 80
    service_discovery_namespace: Optional[str] = None
62
63
64
65
    # PD service discovery configuration
    prefill_selector: Dict[str, str] = dataclasses.field(default_factory=dict)
    decode_selector: Dict[str, str] = dataclasses.field(default_factory=dict)
    bootstrap_port_annotation: str = "sglang.ai/bootstrap-port"
66
67
68
    # Prometheus configuration
    prometheus_port: Optional[int] = None
    prometheus_host: Optional[str] = None
69
70
    # Request ID headers configuration
    request_id_headers: Optional[List[str]] = None
71
    # Request timeout in seconds
72
    request_timeout_secs: int = 1800
73
    # Max concurrent requests for rate limiting
74
    max_concurrent_requests: int = 256
75
76
77
78
79
80
    # Queue size for pending requests when max concurrent limit reached
    queue_size: int = 100
    # Maximum time (in seconds) a request can wait in queue before timing out
    queue_timeout_secs: int = 60
    # Token bucket refill rate (tokens per second). If not set, defaults to max_concurrent_requests
    rate_limit_tokens_per_second: Optional[int] = None
81
82
    # CORS allowed origins
    cors_allowed_origins: List[str] = dataclasses.field(default_factory=list)
83
    # Retry configuration
84
85
86
87
88
    retry_max_retries: int = 5
    retry_initial_backoff_ms: int = 50
    retry_max_backoff_ms: int = 30_000
    retry_backoff_multiplier: float = 1.5
    retry_jitter_factor: float = 0.2
89
    disable_retries: bool = False
90
91
92
93
94
95
    # Health check configuration
    health_failure_threshold: int = 3
    health_success_threshold: int = 2
    health_check_timeout_secs: int = 5
    health_check_interval_secs: int = 60
    health_check_endpoint: str = "/health"
96
    # Circuit breaker configuration
97
98
99
100
    cb_failure_threshold: int = 10
    cb_success_threshold: int = 3
    cb_timeout_duration_secs: int = 60
    cb_window_duration_secs: int = 120
101
    disable_circuit_breaker: bool = False
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

    @staticmethod
    def add_cli_args(
        parser: argparse.ArgumentParser,
        use_router_prefix: bool = False,
        exclude_host_port: bool = False,
    ):
        """
        Add router-specific arguments to an argument parser.

        Args:
            parser: The argument parser to add arguments to
            use_router_prefix: If True, prefix all arguments with 'router-' to avoid conflicts
            exclude_host_port: If True, don't add host and port arguments (used when inheriting from server)
        """
        prefix = "router-" if use_router_prefix else ""

        # Worker configuration
        if not exclude_host_port:
            parser.add_argument(
                "--host",
                type=str,
                default=RouterArgs.host,
                help="Host address to bind the router server",
            )
            parser.add_argument(
                "--port",
                type=int,
                default=RouterArgs.port,
                help="Port number to bind the router server",
            )

        parser.add_argument(
            "--worker-urls",
            type=str,
137
138
            nargs="*",
            default=[],
139
140
141
142
143
144
145
146
            help="List of worker URLs (e.g., http://worker1:8000 http://worker2:8000)",
        )

        # Routing policy configuration
        parser.add_argument(
            f"--{prefix}policy",
            type=str,
            default=RouterArgs.policy,
147
            choices=["random", "round_robin", "cache_aware", "power_of_two"],
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
            help="Load balancing policy to use. In PD mode, this is used for both prefill and decode unless overridden",
        )
        parser.add_argument(
            f"--{prefix}prefill-policy",
            type=str,
            default=None,
            choices=["random", "round_robin", "cache_aware", "power_of_two"],
            help="Specific policy for prefill nodes in PD mode. If not specified, uses the main policy",
        )
        parser.add_argument(
            f"--{prefix}decode-policy",
            type=str,
            default=None,
            choices=["random", "round_robin", "cache_aware", "power_of_two"],
            help="Specific policy for decode nodes in PD mode. If not specified, uses the main policy",
163
164
165
166
        )

        # PD-specific arguments
        parser.add_argument(
167
            f"--{prefix}pd-disaggregation",
168
169
170
171
172
            action="store_true",
            help="Enable PD (Prefill-Decode) disaggregated mode",
        )
        parser.add_argument(
            f"--{prefix}prefill",
173
            nargs="+",
174
            action="append",
175
176
177
            help="Prefill server URL and optional bootstrap port. Can be specified multiple times. "
            "Format: --prefill URL [BOOTSTRAP_PORT]. "
            "BOOTSTRAP_PORT can be a port number, 'none', or omitted (defaults to none).",
178
179
180
181
182
183
184
        )
        parser.add_argument(
            f"--{prefix}decode",
            nargs=1,
            action="append",
            metavar=("URL",),
            help="Decode server URL. Can be specified multiple times.",
185
        )
186
187
188
189
190
191
        parser.add_argument(
            f"--{prefix}worker-startup-timeout-secs",
            type=int,
            default=RouterArgs.worker_startup_timeout_secs,
            help="Timeout in seconds for worker startup",
        )
192
193
194
195
196
197
        parser.add_argument(
            f"--{prefix}worker-startup-check-interval",
            type=int,
            default=RouterArgs.worker_startup_check_interval,
            help="Interval in seconds between checks for worker startup",
        )
198
199
200
201
202
203
204
        parser.add_argument(
            f"--{prefix}cache-threshold",
            type=float,
            default=RouterArgs.cache_threshold,
            help="Cache threshold (0.0-1.0) for cache-aware routing",
        )
        parser.add_argument(
205
206
207
208
209
210
211
            f"--{prefix}balance-abs-threshold",
            type=int,
            default=RouterArgs.balance_abs_threshold,
            help="Load balancing is triggered when (max_load - min_load) > abs_threshold AND max_load > min_load * rel_threshold. Otherwise, use cache aware",
        )
        parser.add_argument(
            f"--{prefix}balance-rel-threshold",
212
            type=float,
213
214
            default=RouterArgs.balance_rel_threshold,
            help="Load balancing is triggered when (max_load - min_load) > abs_threshold AND max_load > min_load * rel_threshold. Otherwise, use cache aware",
215
216
217
218
219
220
221
222
223
224
225
226
227
        )
        parser.add_argument(
            f"--{prefix}eviction-interval",
            type=int,
            default=RouterArgs.eviction_interval,
            help="Interval in seconds between cache eviction operations",
        )
        parser.add_argument(
            f"--{prefix}max-tree-size",
            type=int,
            default=RouterArgs.max_tree_size,
            help="Maximum size of the approximation tree for cache-aware routing",
        )
228
229
230
231
232
233
        parser.add_argument(
            f"--{prefix}max-payload-size",
            type=int,
            default=RouterArgs.max_payload_size,
            help="Maximum payload size in bytes",
        )
234
235
236
237
238
239
240
241
242
243
244
        parser.add_argument(
            f"--{prefix}dp-aware",
            action="store_true",
            help="Enable data parallelism aware schedule",
        )
        parser.add_argument(
            f"--{prefix}api-key",
            type=str,
            default=None,
            help="The api key used for the authorization with the worker.  Useful when the dp aware scheduling strategy is enaled.",
        )
245
246
247
248
249
250
        parser.add_argument(
            f"--{prefix}log-dir",
            type=str,
            default=None,
            help="Directory to store log files. If not specified, logs are only output to console.",
        )
251
252
253
254
255
256
257
        parser.add_argument(
            f"--{prefix}log-level",
            type=str,
            default="info",
            choices=["debug", "info", "warning", "error", "critical"],
            help="Set the logging level. If not specified, defaults to INFO.",
        )
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
        parser.add_argument(
            f"--{prefix}service-discovery",
            action="store_true",
            help="Enable Kubernetes service discovery",
        )
        parser.add_argument(
            f"--{prefix}selector",
            type=str,
            nargs="+",
            help="Label selector for Kubernetes service discovery (format: key1=value1 key2=value2)",
        )
        parser.add_argument(
            f"--{prefix}service-discovery-port",
            type=int,
            default=RouterArgs.service_discovery_port,
            help="Port to use for discovered worker pods",
        )
        parser.add_argument(
            f"--{prefix}service-discovery-namespace",
            type=str,
            help="Kubernetes namespace to watch for pods. If not provided, watches all namespaces (requires cluster-wide permissions)",
        )
280
281
282
283
284
285
286
287
288
289
290
291
        parser.add_argument(
            f"--{prefix}prefill-selector",
            type=str,
            nargs="+",
            help="Label selector for prefill server pods in PD mode (format: key1=value1 key2=value2)",
        )
        parser.add_argument(
            f"--{prefix}decode-selector",
            type=str,
            nargs="+",
            help="Label selector for decode server pods in PD mode (format: key1=value1 key2=value2)",
        )
292
293
294
295
296
297
298
299
300
301
302
303
304
        # Prometheus configuration
        parser.add_argument(
            f"--{prefix}prometheus-port",
            type=int,
            default=29000,
            help="Port to expose Prometheus metrics. If not specified, Prometheus metrics are disabled",
        )
        parser.add_argument(
            f"--{prefix}prometheus-host",
            type=str,
            default="127.0.0.1",
            help="Host address to bind the Prometheus metrics server",
        )
305
306
307
308
309
310
        parser.add_argument(
            f"--{prefix}request-id-headers",
            type=str,
            nargs="*",
            help="Custom HTTP headers to check for request IDs (e.g., x-request-id x-trace-id). If not specified, uses common defaults.",
        )
311
312
313
314
315
316
        parser.add_argument(
            f"--{prefix}request-timeout-secs",
            type=int,
            default=RouterArgs.request_timeout_secs,
            help="Request timeout in seconds",
        )
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
        # Retry configuration
        parser.add_argument(
            f"--{prefix}retry-max-retries",
            type=int,
            default=RouterArgs.retry_max_retries,
        )
        parser.add_argument(
            f"--{prefix}retry-initial-backoff-ms",
            type=int,
            default=RouterArgs.retry_initial_backoff_ms,
        )
        parser.add_argument(
            f"--{prefix}retry-max-backoff-ms",
            type=int,
            default=RouterArgs.retry_max_backoff_ms,
        )
        parser.add_argument(
            f"--{prefix}retry-backoff-multiplier",
            type=float,
            default=RouterArgs.retry_backoff_multiplier,
        )
        parser.add_argument(
            f"--{prefix}retry-jitter-factor",
            type=float,
            default=RouterArgs.retry_jitter_factor,
        )
        parser.add_argument(
            f"--{prefix}disable-retries",
            action="store_true",
            help="Disable retries (equivalent to setting retry_max_retries=1)",
        )
        # Circuit breaker configuration
        parser.add_argument(
            f"--{prefix}cb-failure-threshold",
            type=int,
            default=RouterArgs.cb_failure_threshold,
        )
        parser.add_argument(
            f"--{prefix}cb-success-threshold",
            type=int,
            default=RouterArgs.cb_success_threshold,
        )
        parser.add_argument(
            f"--{prefix}cb-timeout-duration-secs",
            type=int,
            default=RouterArgs.cb_timeout_duration_secs,
        )
        parser.add_argument(
            f"--{prefix}cb-window-duration-secs",
            type=int,
            default=RouterArgs.cb_window_duration_secs,
        )
        parser.add_argument(
            f"--{prefix}disable-circuit-breaker",
            action="store_true",
            help="Disable circuit breaker (equivalent to setting cb_failure_threshold to u32::MAX)",
        )
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
        # Health check configuration
        parser.add_argument(
            f"--{prefix}health-failure-threshold",
            type=int,
            default=RouterArgs.health_failure_threshold,
            help="Number of consecutive health check failures before marking worker unhealthy",
        )
        parser.add_argument(
            f"--{prefix}health-success-threshold",
            type=int,
            default=RouterArgs.health_success_threshold,
            help="Number of consecutive health check successes before marking worker healthy",
        )
        parser.add_argument(
            f"--{prefix}health-check-timeout-secs",
            type=int,
            default=RouterArgs.health_check_timeout_secs,
            help="Timeout in seconds for health check requests",
        )
        parser.add_argument(
            f"--{prefix}health-check-interval-secs",
            type=int,
            default=RouterArgs.health_check_interval_secs,
            help="Interval in seconds between runtime health checks",
        )
        parser.add_argument(
            f"--{prefix}health-check-endpoint",
            type=str,
            default=RouterArgs.health_check_endpoint,
            help="Health check endpoint path",
        )
405
406
407
408
409
410
        parser.add_argument(
            f"--{prefix}max-concurrent-requests",
            type=int,
            default=RouterArgs.max_concurrent_requests,
            help="Maximum number of concurrent requests allowed (for rate limiting)",
        )
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
        parser.add_argument(
            f"--{prefix}queue-size",
            type=int,
            default=RouterArgs.queue_size,
            help="Queue size for pending requests when max concurrent limit reached (0 = no queue, return 429 immediately)",
        )
        parser.add_argument(
            f"--{prefix}queue-timeout-secs",
            type=int,
            default=RouterArgs.queue_timeout_secs,
            help="Maximum time (in seconds) a request can wait in queue before timing out",
        )
        parser.add_argument(
            f"--{prefix}rate-limit-tokens-per-second",
            type=int,
            default=RouterArgs.rate_limit_tokens_per_second,
            help="Token bucket refill rate (tokens per second). If not set, defaults to max_concurrent_requests",
        )
429
430
431
432
433
434
435
        parser.add_argument(
            f"--{prefix}cors-allowed-origins",
            type=str,
            nargs="*",
            default=[],
            help="CORS allowed origins (e.g., http://localhost:3000 https://example.com)",
        )
436
437
438
439
440
441
442
443
444
445
446
447
448

    @classmethod
    def from_cli_args(
        cls, args: argparse.Namespace, use_router_prefix: bool = False
    ) -> "RouterArgs":
        """
        Create RouterArgs instance from parsed command line arguments.

        Args:
            args: Parsed command line arguments
            use_router_prefix: If True, look for arguments with 'router-' prefix
        """
        prefix = "router_" if use_router_prefix else ""
449
450
451
452
453
454
        worker_urls = getattr(args, "worker_urls", [])

        # Parse PD URLs
        prefill_urls = cls._parse_prefill_urls(getattr(args, f"{prefix}prefill", None))
        decode_urls = cls._parse_decode_urls(getattr(args, f"{prefix}decode", None))

455
        return cls(
456
            worker_urls=worker_urls,
457
458
            host=args.host,
            port=args.port,
459
            pd_disaggregation=getattr(args, f"{prefix}pd_disaggregation", False),
460
461
            prefill_urls=prefill_urls,
            decode_urls=decode_urls,
462
            policy=getattr(args, f"{prefix}policy"),
463
464
            prefill_policy=getattr(args, f"{prefix}prefill_policy", None),
            decode_policy=getattr(args, f"{prefix}decode_policy", None),
465
466
467
            worker_startup_timeout_secs=getattr(
                args, f"{prefix}worker_startup_timeout_secs"
            ),
468
469
470
            worker_startup_check_interval=getattr(
                args, f"{prefix}worker_startup_check_interval"
            ),
471
            cache_threshold=getattr(args, f"{prefix}cache_threshold"),
472
473
            balance_abs_threshold=getattr(args, f"{prefix}balance_abs_threshold"),
            balance_rel_threshold=getattr(args, f"{prefix}balance_rel_threshold"),
474
475
            eviction_interval=getattr(args, f"{prefix}eviction_interval"),
            max_tree_size=getattr(args, f"{prefix}max_tree_size"),
476
            max_payload_size=getattr(args, f"{prefix}max_payload_size"),
477
478
            dp_aware=getattr(args, f"{prefix}dp_aware", False),
            api_key=getattr(args, f"{prefix}api_key", None),
479
            log_dir=getattr(args, f"{prefix}log_dir", None),
480
            log_level=getattr(args, f"{prefix}log_level", None),
481
482
483
484
485
486
            service_discovery=getattr(args, f"{prefix}service_discovery", False),
            selector=cls._parse_selector(getattr(args, f"{prefix}selector", None)),
            service_discovery_port=getattr(args, f"{prefix}service_discovery_port"),
            service_discovery_namespace=getattr(
                args, f"{prefix}service_discovery_namespace", None
            ),
487
488
489
490
491
492
493
            prefill_selector=cls._parse_selector(
                getattr(args, f"{prefix}prefill_selector", None)
            ),
            decode_selector=cls._parse_selector(
                getattr(args, f"{prefix}decode_selector", None)
            ),
            bootstrap_port_annotation="sglang.ai/bootstrap-port",  # Mooncake-specific annotation
494
495
            prometheus_port=getattr(args, f"{prefix}prometheus_port", None),
            prometheus_host=getattr(args, f"{prefix}prometheus_host", None),
496
            request_id_headers=getattr(args, f"{prefix}request_id_headers", None),
497
498
499
500
501
502
503
504
            request_timeout_secs=getattr(
                args, f"{prefix}request_timeout_secs", RouterArgs.request_timeout_secs
            ),
            max_concurrent_requests=getattr(
                args,
                f"{prefix}max_concurrent_requests",
                RouterArgs.max_concurrent_requests,
            ),
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
            queue_size=getattr(
                args,
                f"{prefix}queue_size",
                RouterArgs.queue_size,
            ),
            queue_timeout_secs=getattr(
                args,
                f"{prefix}queue_timeout_secs",
                RouterArgs.queue_timeout_secs,
            ),
            rate_limit_tokens_per_second=getattr(
                args,
                f"{prefix}rate_limit_tokens_per_second",
                RouterArgs.rate_limit_tokens_per_second,
            ),
520
            cors_allowed_origins=getattr(args, f"{prefix}cors_allowed_origins", []),
521
522
523
524
525
526
527
528
529
530
531
532
533
            retry_max_retries=getattr(args, f"{prefix}retry_max_retries"),
            retry_initial_backoff_ms=getattr(args, f"{prefix}retry_initial_backoff_ms"),
            retry_max_backoff_ms=getattr(args, f"{prefix}retry_max_backoff_ms"),
            retry_backoff_multiplier=getattr(args, f"{prefix}retry_backoff_multiplier"),
            retry_jitter_factor=getattr(args, f"{prefix}retry_jitter_factor"),
            cb_failure_threshold=getattr(args, f"{prefix}cb_failure_threshold"),
            cb_success_threshold=getattr(args, f"{prefix}cb_success_threshold"),
            cb_timeout_duration_secs=getattr(args, f"{prefix}cb_timeout_duration_secs"),
            cb_window_duration_secs=getattr(args, f"{prefix}cb_window_duration_secs"),
            disable_retries=getattr(args, f"{prefix}disable_retries", False),
            disable_circuit_breaker=getattr(
                args, f"{prefix}disable_circuit_breaker", False
            ),
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
            health_failure_threshold=getattr(
                args,
                f"{prefix}health_failure_threshold",
                RouterArgs.health_failure_threshold,
            ),
            health_success_threshold=getattr(
                args,
                f"{prefix}health_success_threshold",
                RouterArgs.health_success_threshold,
            ),
            health_check_timeout_secs=getattr(
                args,
                f"{prefix}health_check_timeout_secs",
                RouterArgs.health_check_timeout_secs,
            ),
            health_check_interval_secs=getattr(
                args,
                f"{prefix}health_check_interval_secs",
                RouterArgs.health_check_interval_secs,
            ),
            health_check_endpoint=getattr(
                args, f"{prefix}health_check_endpoint", RouterArgs.health_check_endpoint
            ),
557
558
        )

559
560
561
562
563
564
565
566
567
568
569
570
    @staticmethod
    def _parse_selector(selector_list):
        if not selector_list:
            return {}

        selector = {}
        for item in selector_list:
            if "=" in item:
                key, value = item.split("=", 1)
                selector[key] = value
        return selector

571
572
573
574
    @staticmethod
    def _parse_prefill_urls(prefill_list):
        """Parse prefill URLs from --prefill arguments.

575
576
577
578
579
        Format: --prefill URL [BOOTSTRAP_PORT]
        Example:
            --prefill http://prefill1:8080 9000  # With bootstrap port
            --prefill http://prefill2:8080 none  # Explicitly no bootstrap port
            --prefill http://prefill3:8080       # Defaults to no bootstrap port
580
581
582
583
584
        """
        if not prefill_list:
            return []

        prefill_urls = []
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
        for prefill_args in prefill_list:

            url = prefill_args[0]

            # Handle optional bootstrap port
            if len(prefill_args) >= 2:
                bootstrap_port_str = prefill_args[1]
                # Handle 'none' as None
                if bootstrap_port_str.lower() == "none":
                    bootstrap_port = None
                else:
                    try:
                        bootstrap_port = int(bootstrap_port_str)
                    except ValueError:
                        raise ValueError(
                            f"Invalid bootstrap port: {bootstrap_port_str}. Must be a number or 'none'"
                        )
602
            else:
603
604
                # No bootstrap port specified, default to None
                bootstrap_port = None
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622

            prefill_urls.append((url, bootstrap_port))

        return prefill_urls

    @staticmethod
    def _parse_decode_urls(decode_list):
        """Parse decode URLs from --decode arguments.

        Format: --decode URL
        Example: --decode http://decode1:8081 --decode http://decode2:8081
        """
        if not decode_list:
            return []

        # decode_list is a list of single-element lists due to nargs=1
        return [url[0] for url in decode_list]

623
624
625
626
627
628
629

def policy_from_str(policy_str: str) -> PolicyType:
    """Convert policy string to PolicyType enum."""
    policy_map = {
        "random": PolicyType.Random,
        "round_robin": PolicyType.RoundRobin,
        "cache_aware": PolicyType.CacheAware,
630
        "power_of_two": PolicyType.PowerOfTwo,
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
    }
    return policy_map[policy_str]


def launch_router(args: argparse.Namespace) -> Optional[Router]:
    """
    Launch the SGLang router with the configuration from parsed arguments.

    Args:
        args: Namespace object containing router configuration
            Can be either raw argparse.Namespace or converted RouterArgs

    Returns:
        Router instance if successful, None if failed
    """
646
    logger = logging.getLogger("router")
647
648
649
650
651
652
653
    try:
        # Convert to RouterArgs if needed
        if not isinstance(args, RouterArgs):
            router_args = RouterArgs.from_cli_args(args)
        else:
            router_args = args

654
        # Validate configuration based on mode
655
656
657
658
659
660
661
        if router_args.pd_disaggregation:
            # Validate PD configuration - skip URL requirements if using service discovery
            if not router_args.service_discovery:
                if not router_args.prefill_urls:
                    raise ValueError("PD disaggregation mode requires --prefill")
                if not router_args.decode_urls:
                    raise ValueError("PD disaggregation mode requires --decode")
662

663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
            # Warn about policy usage in PD mode
            if (
                router_args.prefill_policy
                and router_args.decode_policy
                and router_args.policy
            ):
                logger.warning(
                    "Both --prefill-policy and --decode-policy are specified. "
                    "The main --policy flag will be ignored for PD mode."
                )
            elif (
                router_args.prefill_policy
                and not router_args.decode_policy
                and router_args.policy
            ):
                logger.info(
                    f"Using --prefill-policy '{router_args.prefill_policy}' for prefill nodes "
                    f"and --policy '{router_args.policy}' for decode nodes."
                )
            elif (
                router_args.decode_policy
                and not router_args.prefill_policy
                and router_args.policy
            ):
                logger.info(
                    f"Using --policy '{router_args.policy}' for prefill nodes "
                    f"and --decode-policy '{router_args.decode_policy}' for decode nodes."
                )

692
        # Create router with unified constructor
693
        router = Router(
694
            worker_urls=(
695
696
697
                []
                if router_args.service_discovery or router_args.pd_disaggregation
                else router_args.worker_urls
698
            ),
699
700
            host=router_args.host,
            port=router_args.port,
701
702
            policy=policy_from_str(router_args.policy),
            worker_startup_timeout_secs=router_args.worker_startup_timeout_secs,
703
            worker_startup_check_interval=router_args.worker_startup_check_interval,
704
            cache_threshold=router_args.cache_threshold,
705
706
            balance_abs_threshold=router_args.balance_abs_threshold,
            balance_rel_threshold=router_args.balance_rel_threshold,
707
708
            eviction_interval_secs=router_args.eviction_interval,
            max_tree_size=router_args.max_tree_size,
709
            max_payload_size=router_args.max_payload_size,
710
711
            dp_aware=router_args.dp_aware,
            api_key=router_args.api_key,
712
            log_dir=router_args.log_dir,
713
            log_level=router_args.log_level,
714
715
716
717
            service_discovery=router_args.service_discovery,
            selector=router_args.selector,
            service_discovery_port=router_args.service_discovery_port,
            service_discovery_namespace=router_args.service_discovery_namespace,
718
719
            prefill_selector=router_args.prefill_selector,
            decode_selector=router_args.decode_selector,
720
721
            prometheus_port=router_args.prometheus_port,
            prometheus_host=router_args.prometheus_host,
722
            request_timeout_secs=router_args.request_timeout_secs,
723
            pd_disaggregation=router_args.pd_disaggregation,
724
            prefill_urls=(
725
                router_args.prefill_urls if router_args.pd_disaggregation else None
726
727
            ),
            decode_urls=(
728
                router_args.decode_urls if router_args.pd_disaggregation else None
729
            ),
730
731
732
733
734
735
736
737
738
739
            prefill_policy=(
                policy_from_str(router_args.prefill_policy)
                if router_args.prefill_policy
                else None
            ),
            decode_policy=(
                policy_from_str(router_args.decode_policy)
                if router_args.decode_policy
                else None
            ),
740
            request_id_headers=router_args.request_id_headers,
741
            max_concurrent_requests=router_args.max_concurrent_requests,
742
743
744
            queue_size=router_args.queue_size,
            queue_timeout_secs=router_args.queue_timeout_secs,
            rate_limit_tokens_per_second=router_args.rate_limit_tokens_per_second,
745
            cors_allowed_origins=router_args.cors_allowed_origins,
746
747
748
749
750
751
752
753
754
755
756
            retry_max_retries=router_args.retry_max_retries,
            retry_initial_backoff_ms=router_args.retry_initial_backoff_ms,
            retry_max_backoff_ms=router_args.retry_max_backoff_ms,
            retry_backoff_multiplier=router_args.retry_backoff_multiplier,
            retry_jitter_factor=router_args.retry_jitter_factor,
            cb_failure_threshold=router_args.cb_failure_threshold,
            cb_success_threshold=router_args.cb_success_threshold,
            cb_timeout_duration_secs=router_args.cb_timeout_duration_secs,
            cb_window_duration_secs=router_args.cb_window_duration_secs,
            disable_retries=router_args.disable_retries,
            disable_circuit_breaker=router_args.disable_circuit_breaker,
757
758
759
760
761
            health_failure_threshold=router_args.health_failure_threshold,
            health_success_threshold=router_args.health_success_threshold,
            health_check_timeout_secs=router_args.health_check_timeout_secs,
            health_check_interval_secs=router_args.health_check_interval_secs,
            health_check_endpoint=router_args.health_check_endpoint,
762
763
764
765
766
767
        )

        router.start()
        return router

    except Exception as e:
768
        logger.error(f"Error starting router: {e}")
769
        raise e
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789


class CustomHelpFormatter(
    argparse.RawDescriptionHelpFormatter, argparse.ArgumentDefaultsHelpFormatter
):
    """Custom formatter that preserves both description formatting and shows defaults"""

    pass


def parse_router_args(args: List[str]) -> RouterArgs:
    """Parse command line arguments and return RouterArgs instance."""
    parser = argparse.ArgumentParser(
        description="""SGLang Router - High-performance request distribution across worker nodes

Usage:
This launcher enables starting a router with individual worker instances. It is useful for
multi-node setups or when you want to start workers and router separately.

Examples:
790
  # Regular mode
791
  python -m sglang_router.launch_router --worker-urls http://worker1:8000 http://worker2:8000
792

793
  # PD disaggregated mode with same policy for both
794
  python -m sglang_router.launch_router --pd-disaggregation \\
795
    --prefill http://prefill1:8000 9000 --prefill http://prefill2:8000 \\
796
797
    --decode http://decode1:8001 --decode http://decode2:8001 \\
    --policy cache_aware
798

799
800
801
802
803
804
805
  # PD mode with optional bootstrap ports
  python -m sglang_router.launch_router --pd-disaggregation \\
    --prefill http://prefill1:8000 9000 \\    # With bootstrap port
    --prefill http://prefill2:8000 none \\    # Explicitly no bootstrap port
    --prefill http://prefill3:8000 \\         # Defaults to no bootstrap port
    --decode http://decode1:8001 --decode http://decode2:8001

806
807
  # PD mode with different policies for prefill and decode
  python -m sglang_router.launch_router --pd-disaggregation \\
808
    --prefill http://prefill1:8000 --prefill http://prefill2:8000 \\
809
810
811
    --decode http://decode1:8001 --decode http://decode2:8001 \\
    --prefill-policy cache_aware --decode-policy power_of_two

812
813
814
815
816
817
818
819
820
821
    """,
        formatter_class=CustomHelpFormatter,
    )

    RouterArgs.add_cli_args(parser, use_router_prefix=False)
    return RouterArgs.from_cli_args(parser.parse_args(args), use_router_prefix=False)


def main() -> None:
    router_args = parse_router_args(sys.argv[1:])
822
    launch_router(router_args)
823
824
825
826


if __name__ == "__main__":
    main()