Add ZMQ IPv6 support, bench_serving sampling params, and reduce routed_dp_rank log noise (#27180)

Co-authored-by: Hanming Lu <69857889+hanming-lu@users.noreply.github.com>
Co-authored-by: Grigory Sizov <grisha.sizov@gmail.com>

python/sglang/bench_serving.py

@@ -616,13 +616,16 @@ async def async_request_sglang_generate(
     prompt = request_func_input.prompt
 
     async with _create_bench_client_session() as session:
+        sampling_params = {
+            "temperature": args.temperature,
+            "max_new_tokens": request_func_input.output_len,
+            "ignore_eos": not args.disable_ignore_eos,
+        }
+        if args.top_p < 1.0:
+            sampling_params["top_p"] = args.top_p
         payload = {
             ("text" if isinstance(prompt, str) else "input_ids"): prompt,
-            "sampling_params": {
-                "temperature": 0.0,
-                "max_new_tokens": request_func_input.output_len,
-                "ignore_eos": not args.disable_ignore_eos,
-            },
+            "sampling_params": sampling_params,
             "stream": not args.disable_stream,
             "lora_path": request_func_input.lora_name,
             "return_logprob": args.return_logprob,
@@ -1717,6 +1720,11 @@ def run_benchmark(args_: argparse.Namespace):
     if not hasattr(args, "return_logprob"):
         args.return_logprob = False
 
+    if not hasattr(args, "temperature"):
+        args.temperature = 0.0
+    if not hasattr(args, "top_p"):
+        args.top_p = 1.0
+
     if not hasattr(args, "use_trace_timestamps"):
         args.use_trace_timestamps = False
     if not hasattr(args, "mooncake_slowdown_factor"):
@@ -2205,6 +2213,18 @@ if __name__ == "__main__":
         action="store_true",
         help="Disable ignoring EOS.",
     )
+    parser.add_argument(
+        "--temperature",
+        type=float,
+        default=0.0,
+        help="Sampling temperature.",
+    )
+    parser.add_argument(
+        "--top-p",
+        type=float,
+        default=1.0,
+        help="Nucleus sampling parameter.",
+    )
     parser.add_argument(
         "--extra-request-body",
         metavar='{"key1": "value1", "key2": "value2"}',

python/sglang/srt/entrypoints/engine.py

@@ -298,7 +298,7 @@ class Engine(EngineScoreMixin, EngineBase):
         if routed_dp_rank is not None:
             dp_size = self.server_args.dp_size
             if dp_size <= 1 and routed_dp_rank == 0:
-                logger.warning(
+                logger.debug(
                     f"routed_dp_rank={routed_dp_rank} is ignored because dp_size={dp_size}"
                 )
                 return None

python/sglang/srt/managers/load_snapshot.py

@@ -51,6 +51,7 @@ import msgspec.msgpack
 import msgspec.structs
 
 from sglang.srt.environ import envs
+from sglang.srt.utils.network import is_zmq_endpoint_ipv6
 
 if TYPE_CHECKING:
     from sglang.srt.managers.io_struct import GetLoadsReqOutput
@@ -439,6 +440,8 @@ class ZmqLoadSnapshotWriter:
         self._zmq = _zmq
         self._ctx = _zmq.Context.instance()
         self._socket = self._ctx.socket(_zmq.PUSH)
+        if is_zmq_endpoint_ipv6(endpoint):
+            self._socket.setsockopt(_zmq.IPV6, 1)
         self._socket.setsockopt(_zmq.LINGER, 0)
         self._socket.setsockopt(_zmq.CONFLATE, 1)
         self._socket.connect(endpoint)
@@ -577,6 +580,8 @@ class ZmqShmLoadSnapshotReader:
         self._zmq = _zmq
         self._ctx = _zmq.Context.instance()
         self._socket = self._ctx.socket(_zmq.PULL)
+        if is_zmq_endpoint_ipv6(endpoint):
+            self._socket.setsockopt(_zmq.IPV6, 1)
         self._socket.setsockopt(_zmq.LINGER, 0)
         self._socket.setsockopt(_zmq.CONFLATE, 1)
         self._socket.bind(endpoint)

python/sglang/srt/utils/network.py

@@ -374,8 +374,7 @@ def get_zmq_socket(
         port = socket.bind_to_random_port("tcp://*")
         return port, socket
     else:
-        # Handle IPv6 if endpoint contains brackets
-        if endpoint.find("[") != -1:
+        if is_zmq_endpoint_ipv6(endpoint):
             socket.setsockopt(zmq.IPV6, 1)
 
         config_socket(socket, socket_type)
@@ -388,6 +387,17 @@ def get_zmq_socket(
         return socket
 
 
+def is_zmq_endpoint_ipv6(endpoint: str) -> bool:
+    """Return whether a ZMQ TCP endpoint contains a bracketed IPv6 host."""
+    prefix = "tcp://["
+    if not endpoint.startswith(prefix):
+        return False
+    end = endpoint.find("]", len(prefix))
+    if end == -1:
+        return False
+    return is_valid_ipv6_address(endpoint[len(prefix) : end])
+
+
 def _is_ipv6(host: str) -> bool:
     """Check whether *host* is a valid IPv6 address (without brackets)."""
     try:

test/registered/utils/test_network_address.py

@@ -3,7 +3,7 @@ import unittest
 from unittest.mock import patch
 
 from sglang.srt.server_args import PortArgs, ServerArgs
-from sglang.srt.utils.network import NetworkAddress
+from sglang.srt.utils.network import NetworkAddress, is_zmq_endpoint_ipv6
 from sglang.test.ci.ci_register import register_cpu_ci
 
 register_cpu_ci(est_time=7, suite="base-a-test-cpu")
@@ -180,6 +180,22 @@ class TestNetworkAddressParseErrors(unittest.TestCase):
             NetworkAddress.parse(":8000")
 
 
+class TestZmqEndpointIPv6(unittest.TestCase):
+    def test_bracketed_ipv6_endpoint(self):
+        self.assertTrue(is_zmq_endpoint_ipv6("tcp://[::1]:30000"))
+        self.assertTrue(is_zmq_endpoint_ipv6("tcp://[2001:db8::1]:30000"))
+
+    def test_non_ipv6_endpoints(self):
+        self.assertFalse(is_zmq_endpoint_ipv6("tcp://127.0.0.1:30000"))
+        self.assertFalse(is_zmq_endpoint_ipv6("tcp://localhost:30000"))
+        self.assertFalse(is_zmq_endpoint_ipv6("ipc:///tmp/sglang.sock"))
+
+    def test_malformed_or_non_tcp_endpoints(self):
+        self.assertFalse(is_zmq_endpoint_ipv6("tcp://[not-ipv6]:30000"))
+        self.assertFalse(is_zmq_endpoint_ipv6("tcp://[::1:30000"))
+        self.assertFalse(is_zmq_endpoint_ipv6("ipc://[::1]:30000"))
+
+
 class TestNetworkAddressBracketStripping(unittest.TestCase):
     def test_strip_brackets(self):
         na = NetworkAddress("[::1]", 8000)