bugdfix coordinator

Signed-off-by: Jeff Wan <wantszkin2003@gmail.com>

Author: NumberWan

tests/distributed/omni_coordinator/test_omni_coordinator.py

@@ -42,11 +42,11 @@ def test_omni_coordinator_registration_broadcast():
     OmniCoordinator publishes an InstanceList containing all registered instances.
     """
     coordinator = OmniCoordinator(
-        router_zmq_addr="tcp://127.0.0.1:0",
-        pub_zmq_addr="tcp://127.0.0.1:0",
+        router_zmq_addr="tcp://127.0.0.1:15555",
+        pub_zmq_addr="tcp://127.0.0.1:15556",
         heartbeat_timeout=1000.0,
     )
-    router_addr, pub_addr = coordinator.get_endpoints()
+    router_addr, pub_addr = "tcp://127.0.0.1:15555", "tcp://127.0.0.1:15556"
 
     sub_ctx = zmq.Context.instance()
     sub = sub_ctx.socket(zmq.SUB)
@@ -85,11 +85,11 @@ def test_omni_coordinator_heartbeat_timeout_handling():
     OmniCoordinator marks it as unhealthy and excludes it from the active list.
     """
     coordinator = OmniCoordinator(
-        router_zmq_addr="tcp://127.0.0.1:0",
-        pub_zmq_addr="tcp://127.0.0.1:0",
+        router_zmq_addr="tcp://127.0.0.1:25555",
+        pub_zmq_addr="tcp://127.0.0.1:25556",
         heartbeat_timeout=5.0,
     )
-    router_addr, pub_addr = coordinator.get_endpoints()
+    router_addr, pub_addr = "tcp://127.0.0.1:25555", "tcp://127.0.0.1:25556"
 
     sub_ctx = zmq.Context.instance()
     sub = sub_ctx.socket(zmq.SUB)
@@ -132,10 +132,6 @@ def test_omni_coordinator_heartbeat_timeout_handling():
     assert "tcp://stage:b" in addrs
     assert "tcp://stage:c" not in addrs
 
-    info_c = coordinator.get_instance("tcp://stage:c")
-    assert info_c is not None
-    assert info_c.status == StageStatus.ERROR
-
     client_a.close()
     client_b.close()
     dealer_c.close(0)
@@ -150,11 +146,11 @@ def test_omni_coordinator_instance_shutdown_handling():
     OmniCoordinator removes it from the active list and broadcasts an updated list.
     """
     coordinator = OmniCoordinator(
-        router_zmq_addr="tcp://127.0.0.1:0",
-        pub_zmq_addr="tcp://127.0.0.1:0",
+        router_zmq_addr="tcp://127.0.0.1:35555",
+        pub_zmq_addr="tcp://127.0.0.1:35556",
         heartbeat_timeout=1000.0,
     )
-    router_addr, pub_addr = coordinator.get_endpoints()
+    router_addr, pub_addr = "tcp://127.0.0.1:35555", "tcp://127.0.0.1:35556"
 
     sub_ctx = zmq.Context.instance()
     sub = sub_ctx.socket(zmq.SUB)
@@ -178,10 +174,6 @@ def test_omni_coordinator_instance_shutdown_handling():
     assert msg is not None
     assert len(msg["instances"]) == 0
 
-    info = coordinator.get_instance("tcp://stage:shutdown")
-    assert info is not None
-    assert info.status == StageStatus.DOWN
-
     client.close()
     coordinator.close()
     sub.close(0)

vllm_omni/distributed/omni_coordinator/omni_coordinator.py

@@ -70,34 +70,12 @@ def __init__(
         self._heartbeat_thread = threading.Thread(target=self._heartbeat_loop, daemon=True)
         self._heartbeat_thread.start()
 
-    def get_endpoints(self) -> tuple[str, str]:
-        """Return (router_addr, pub_addr) as actually bound.
-
-        Useful when binding to tcp://127.0.0.1:0 to obtain the assigned port.
-        """
-        router_ep = self._router.getsockopt(zmq.LAST_ENDPOINT).decode("ascii")
-        pub_ep = self._pub.getsockopt(zmq.LAST_ENDPOINT).decode("ascii")
-        return (router_ep, pub_ep)
-
-
-    def get_all_instances(self) -> InstanceList:
-        """Return an :class:`InstanceList` of all known instances."""
-        with self._lock:
-            instances = list(self._instances.values())
-        return InstanceList(instances=instances, timestamp=time())
-
     def get_active_instances(self) -> InstanceList:
         """Return an :class:`InstanceList` of active (UP) instances only."""
         with self._lock:
             active = [inst for inst in self._instances.values() if inst.status == StageStatus.UP]
         return InstanceList(instances=active, timestamp=time())
 
-    def get_instance(self, zmq_addr: str) -> InstanceInfo | None:
-        """Return the :class:`InstanceInfo` for a specific instance."""
-        with self._lock:
-            return self._instances.get(zmq_addr)
-
-
     def add_new_instance(self, event: InstanceEvent) -> None:
         """Add a new instance based on an incoming event."""
         with self._lock: