shm_broadcast: retry bind on EADDRINUSE (fix dp-attention port race) (#29455)

Co-authored-by: Xingyu Liu <38244988+charlotte12l@users.noreply.github.com>
Co-authored-by: xingyuliu <xingyuliu@fb.com>

python/sglang/srt/distributed/device_communicators/shm_broadcast.py

@@ -18,6 +18,7 @@ from torch.distributed import ProcessGroup
 from zmq import IPV6  # type: ignore
 from zmq import SUB, SUBSCRIBE, XPUB, XPUB_VERBOSE, Context  # type: ignore
 
+from sglang.srt.environ import envs
 from sglang.srt.utils.network import NetworkAddress, get_local_ip_auto, get_open_port
 from sglang.srt.utils.stale_shm_cleanup import make_shm_name
 
@@ -211,10 +212,18 @@ class MessageQueue:
             # message. otherwise, we will only receive the first subscription
             # see http://api.zeromq.org/3-3:zmq-setsockopt for more details
             self.local_socket.setsockopt(XPUB_VERBOSE, True)
-            local_subscribe_port = get_open_port()
-            socket_addr = f"tcp://127.0.0.1:{local_subscribe_port}"
-            logger.debug("Binding to %s", socket_addr)
-            self.local_socket.bind(socket_addr)
+            # Bind atomically to avoid get_open_port()'s check-then-bind race;
+            # search from SGLANG_PORT to keep the existing port range.
+            sglang_port = envs.SGLANG_PORT.get()
+            if sglang_port is not None:
+                local_subscribe_port = self.local_socket.bind_to_random_port(
+                    "tcp://127.0.0.1", min_port=sglang_port, max_port=sglang_port + 8
+                )
+            else:
+                local_subscribe_port = self.local_socket.bind_to_random_port(
+                    "tcp://127.0.0.1"
+                )
+            logger.debug("Bound to tcp://127.0.0.1:%d", local_subscribe_port)
             self.current_idx = 0
 
         else:

python/sglang/srt/environ.py

@@ -394,6 +394,12 @@ class Envs:
     SGLANG_DISTRIBUTED_INIT_METHOD_OVERRIDE = EnvStr(None)
     SGLANG_TCP_STORE_PORT = EnvInt(29600)
 
+    # Base port hint for ephemeral sockets (ZMQ, SHM broadcaster, etc.).
+    # When set, get_open_port() and shm_broadcast search upwards from this
+    # value instead of asking the OS for a random port.  Useful to keep all
+    # SGLang ports in a predictable range behind a firewall.
+    SGLANG_PORT = EnvInt(None)
+
     # Tool Calling
     SGLANG_FORWARD_UNKNOWN_TOOLS = EnvBool(False)
 

python/sglang/srt/utils/network.py

@@ -15,9 +15,10 @@ logger = logging.getLogger(__name__)
 
 
 def get_open_port() -> int:
-    port = os.getenv("SGLANG_PORT")
+    from sglang.srt.environ import envs
+
+    port = envs.SGLANG_PORT.get()
     if port is not None:
-        port = int(port)
         while True:
             if is_port_available(port):
                 return port