diff --git a/docs/user-guide/training-backend.md b/docs/user-guide/training-backend.md index 87c23c5d1a9..ff7e17db471 100644 --- a/docs/user-guide/training-backend.md +++ b/docs/user-guide/training-backend.md @@ -485,10 +485,6 @@ A router sits in front of the SGLang workers. Router-side flags take a `--router --router-balance-abs-threshold 0 # force uniform distribution (lowers prefix-cache hit rate) ``` -Miles always starts its own router per model; external router mode -(`--sglang-router-ip`/`--sglang-router-port`) was removed and is expected to -return with the k8s-native mode. - --- ## Further reading diff --git a/miles/utils/misc.py b/miles/utils/misc.py index 4abe3dc24c5..766bf7eddc9 100644 --- a/miles/utils/misc.py +++ b/miles/utils/misc.py @@ -93,6 +93,10 @@ def _get_node_ip() -> str: def _get_free_port_block(*, start_port: int, count: int) -> int: return get_free_port(start_port=start_port, consecutive=count) + @staticmethod + def _is_port_available(*, port: int) -> bool: + return is_port_available(port) + @staticmethod def _get_gpu_uuids(gpu_ids: list[int]) -> list[str | None]: return get_gpu_uuids(gpu_ids) diff --git a/miles/utils/test_utils/mock_sglang_engine.py b/miles/utils/test_utils/mock_sglang_engine.py index 8747f9297fa..eac9d87dd50 100644 --- a/miles/utils/test_utils/mock_sglang_engine.py +++ b/miles/utils/test_utils/mock_sglang_engine.py @@ -92,6 +92,10 @@ def _get_node_ip(self): self._record("_get_node_ip", (), {}) return NodeProbeMixin._get_node_ip() + def _is_port_available(self, *, port: int) -> bool: + self._record("_is_port_available", (), {"port": port}) + return True + def _get_gpu_uuids(self, gpu_ids: list[int]): self._record("_get_gpu_uuids", (gpu_ids,), {}) return [None] * len(gpu_ids) diff --git a/miles/utils/workers/ray_worker_manager.py b/miles/utils/workers/ray_worker_manager.py index d6900b24b4d..64edae8a2ca 100644 --- a/miles/utils/workers/ray_worker_manager.py +++ b/miles/utils/workers/ray_worker_manager.py @@ -148,15 +148,22 @@ async def alloc_ports(self) -> None: node_ip = await self.actor_handle._get_node_ip.remote() for port_info in self.spec.port_infos: - port = ( - self.manager_ref.port_allocator.alloc( + if port_info.allow_dynamic: + port = self.manager_ref.port_allocator.alloc( self.actor_handle, node_ip=node_ip, consecutive=port_info.num_consecutive ) - if port_info.allow_dynamic - else port_info.static_port + (self.cell_index if port_info.offset_by_cell else 0) - ) + else: + port = port_info.static_port + (self.cell_index if port_info.offset_by_cell else 0) + await self._assert_static_port_is_free(port=port, port_name=port_info.name, node_ip=node_ip) self.self_addrs[port_info.name] = HostAndPort(host=_wrap_ipv6(node_ip), port=port) + async def _assert_static_port_is_free(self, *, port: int, port_name: str, node_ip: str) -> None: + free = await self.actor_handle._is_port_available.remote(port=port) + assert free, ( + f"Port {port} on {node_ip} is already in use, so {self.name} cannot serve its {port_name!r} " + f"endpoint there; a stale process from an earlier run is the usual cause" + ) + async def post_setup(self) -> None: ctx = LaunchCommandContext( cell_index=self.cell_index, diff --git a/tests/fast/utils/test_utils/test_mock_sglang_engine.py b/tests/fast/utils/test_utils/test_mock_sglang_engine.py index a7b56ca66f2..9ab9aa58cda 100644 --- a/tests/fast/utils/test_utils/test_mock_sglang_engine.py +++ b/tests/fast/utils/test_utils/test_mock_sglang_engine.py @@ -15,7 +15,7 @@ def _public_methods(cls) -> set[str]: """Public methods (and known semi-private helpers) of ``cls``.""" if hasattr(cls, "__ray_actor_class__"): cls = cls.__ray_actor_class__ # unwrap @ray.remote - keep_underscored = {"_get_free_port_block", "_get_node_ip", "_get_gpu_uuids", "_collect_env_report"} + keep_underscored = {name for name in vars(NodeProbeMixin) if not name.startswith("__")} return { name for name, _ in inspect.getmembers(cls, predicate=inspect.isfunction) @@ -95,6 +95,16 @@ def test_signature_compat_for_every_shared_method(self) -> None: ) +class TestNodeProbeHelpersAreCovered: + def test_every_node_probe_helper_takes_part_in_the_contract_comparison(self) -> None: + """A probe helper left out of the comparison set lets the mock silently lose a stub the real actor has.""" + probe_helpers = {name for name in vars(NodeProbeMixin) if not name.startswith("__")} + + assert probe_helpers + assert probe_helpers <= _public_methods(CommandActor) + assert probe_helpers <= _public_methods(MockSGLangEngine) + + # ----------------------------- real Ray smoke tests ----------------------------- @@ -139,6 +149,15 @@ def test_fault_injection_round_trips_through_ray(self, ray_local_mode): ray.kill(actor) +class TestPortProbe: + def test_the_mock_reports_every_port_as_free_and_records_the_probe(self) -> None: + """A mock-driven launch must never be refused a pinned port by the real machine's port state.""" + engine = MockSGLangEngine.__ray_actor_class__() + + assert engine._is_port_available(port=30001) is True + assert engine.get_calls() == [("_is_port_available", (), {"port": 30001})] + + class TestNodeAddress: def test_the_mock_reports_the_node_ip_instead_of_loopback(self) -> None: """A mock engine placed on another node must publish an address its peers can reach.""" diff --git a/tests/fast/utils/workers/fake_ray.py b/tests/fast/utils/workers/fake_ray.py index dbf10365627..8b750651bb8 100644 --- a/tests/fast/utils/workers/fake_ray.py +++ b/tests/fast/utils/workers/fake_ray.py @@ -6,6 +6,7 @@ _ASYNC_METHOD_NODE_IP = "_get_node_ip" _ASYNC_METHOD_FREE_PORT_BLOCK = "_get_free_port_block" +_ASYNC_METHOD_IS_PORT_AVAILABLE = "_is_port_available" EVENT_CREATE = "create" EVENT_KILL = "kill" @@ -171,6 +172,8 @@ def last_event_index(self, event: str) -> int: def _compute_value(self, *, handle: FakeRayActorHandle, method: str, kwargs: dict[str, Any]) -> Any: if method == _ASYNC_METHOD_NODE_IP: return handle.node_ip + if method == _ASYNC_METHOD_IS_PORT_AVAILABLE: + return kwargs["port"] not in self._used_ports.get(handle.node_ip, set()) if method == _ASYNC_METHOD_FREE_PORT_BLOCK: return self._alloc_port_block( node_ip=handle.node_ip, start_port=kwargs["start_port"], count=kwargs["count"] diff --git a/tests/fast/utils/workers/test_ray_worker_manager.py b/tests/fast/utils/workers/test_ray_worker_manager.py index 059c854e2b8..a12d858e9ed 100644 --- a/tests/fast/utils/workers/test_ray_worker_manager.py +++ b/tests/fast/utils/workers/test_ray_worker_manager.py @@ -176,7 +176,7 @@ async def test_a_consecutive_port_block_is_reserved_as_a_whole(self, fake_ray_cl assert [call.kwargs["count"] for call in fake_ray_cluster.calls_of("_get_free_port_block")] == [5, 5] async def test_static_ports_bypass_the_allocator(self, fake_ray_cluster: FakeRayCluster): - """A port the worker cannot choose is taken from the spec verbatim, without probing the node.""" + """A port the worker cannot choose is taken from the spec verbatim, without asking the allocator.""" spec = _make_spec( "router", port_infos=[ @@ -188,6 +188,15 @@ async def test_static_ports_bypass_the_allocator(self, fake_ray_cluster: FakeRay assert manager.get_worker_addr("router-0-0").port == 7777 assert len(fake_ray_cluster.calls_of("_get_free_port_block")) == 1 + assert [call.kwargs["port"] for call in fake_ray_cluster.calls_of("_is_port_available")] == [7777] + + async def test_a_static_port_already_taken_on_the_node_fails_the_launch(self, fake_ray_cluster: FakeRayCluster): + """A stale listener on the pinned port would otherwise be mistaken for our own worker.""" + fake_ray_cluster.occupy_ports("10.0.0.1", 7777) + spec = _make_spec("router", port_infos=[PortInfo(name="primary", static_port=7777, allow_dynamic=False)]) + + with pytest.raises(AssertionError, match="7777 on 10.0.0.1 is already in use"): + await _launch([spec]) async def test_ports_are_tracked_per_node(self, fake_ray_cluster: FakeRayCluster): """Workers on different nodes may reuse the same port number.""" @@ -214,6 +223,119 @@ async def test_ipv6_hosts_are_bracketed(self, fake_ray_cluster: FakeRayCluster): assert manager.get_worker_addr("router-0-0").host == "[2001:db8::7]" +class TestStaticPortsAreProbedBeforeUse: + async def test_the_probe_asks_for_the_cell_offset_port_rather_than_the_spec_base( + self, fake_ray_cluster: FakeRayCluster + ): + """Probing the spec's base port would leave every cell but the first checking an address it never binds.""" + spec = _make_spec( + "session-server", + num_cells=3, + port_infos=[PortInfo(name="primary", static_port=7000, allow_dynamic=False, offset_by_cell=True)], + ) + await _launch([spec]) + + assert [call.kwargs["port"] for call in fake_ray_cluster.calls_of("_is_port_available")] == [7000, 7001, 7002] + + async def test_an_occupied_cell_offset_port_refuses_the_launch(self, fake_ray_cluster: FakeRayCluster): + """A stale listener on a later cell's offset port must be caught too, not only one on the base port.""" + fake_ray_cluster.occupy_ports("10.0.0.1", 7002) + spec = _make_spec( + "session-server", + num_cells=3, + port_infos=[PortInfo(name="primary", static_port=7000, allow_dynamic=False, offset_by_cell=True)], + ) + + with pytest.raises(AssertionError, match="Port 7002 on 10.0.0.1 is already in use"): + await _launch([spec]) + + async def test_the_probe_runs_on_the_node_the_worker_landed_on(self, fake_ray_cluster: FakeRayCluster): + """A port is only contended on the node that binds it, so the probe must go through the worker's own actor.""" + fake_ray_cluster.use_node_ips("10.0.0.1", "10.0.0.2") + fake_ray_cluster.occupy_ports("10.0.0.2", 7777) + spec = _make_spec( + "router", + num_workers_per_cell=2, + port_infos=[PortInfo(name="primary", static_port=7777, allow_dynamic=False)], + ) + + with pytest.raises(AssertionError, match="Port 7777 on 10.0.0.2 is already in use"): + await _launch([spec]) + + assert [call.handle.node_ip for call in fake_ray_cluster.calls_of("_is_port_available")] == [ + "10.0.0.1", + "10.0.0.2", + ] + + async def test_a_port_occupied_on_an_unrelated_node_does_not_block_the_launch( + self, fake_ray_cluster: FakeRayCluster + ): + """Refusing a port because another machine uses it would make every run hostage to unrelated nodes.""" + fake_ray_cluster.occupy_ports("10.0.0.2", 7777) + spec = _make_spec("router", port_infos=[PortInfo(name="primary", static_port=7777, allow_dynamic=False)]) + + manager = await _launch([spec]) + + assert manager.get_worker_addrs("router-0-0")["primary"].port == 7777 + + async def test_a_master_static_port_is_probed_only_by_the_worker_that_reserves_it( + self, fake_ray_cluster: FakeRayCluster + ): + """Peers that never bind the cell's master port would otherwise refuse a port their own worker 0 is serving.""" + spec = _make_spec( + "engine", + num_cells=2, + num_workers_per_cell=3, + port_infos=[ + PortInfo(name="primary", static_port=8000, allow_dynamic=True), + PortInfo(name="dist_init", static_port=9000, mode="master", allow_dynamic=False), + ], + ) + await _launch([spec]) + + assert [call.kwargs["port"] for call in fake_ray_cluster.calls_of("_is_port_available")] == [9000, 9000] + + async def test_dynamic_ports_are_never_probed_by_the_static_check(self, fake_ray_cluster: FakeRayCluster): + """The allocator already hands out a free port, and re-probing its own reservation would refuse it.""" + spec = _make_spec( + "engine", + num_cells=2, + num_workers_per_cell=2, + port_infos=[ + PortInfo(name="primary", static_port=8000, allow_dynamic=True), + PortInfo(name="nccl", static_port=10000, allow_dynamic=True), + ], + ) + await _launch([spec]) + + assert fake_ray_cluster.calls_of("_is_port_available") == [] + + async def test_the_refusal_names_the_worker_and_the_endpoint_it_could_not_serve( + self, fake_ray_cluster: FakeRayCluster + ): + """An operator hunting the stale process needs to know which worker wanted which endpoint.""" + fake_ray_cluster.occupy_ports("10.0.0.1", 7777) + spec = _make_spec("router", port_infos=[PortInfo(name="prometheus", static_port=7777, allow_dynamic=False)]) + + with pytest.raises(AssertionError, match="router-0-0 cannot serve its 'prometheus' endpoint"): + await _launch([spec]) + + async def test_a_refused_port_rolls_the_cell_back_before_any_command_runs(self, fake_ray_cluster: FakeRayCluster): + """A refusal that let the commands start would add our own processes to the stale ones already there.""" + fake_ray_cluster.occupy_ports("10.0.0.1", 7777) + spec = _make_spec( + "router", + num_workers_per_cell=2, + port_infos=[PortInfo(name="primary", static_port=7777, allow_dynamic=False)], + ) + + with pytest.raises(AssertionError): + await _launch([spec]) + + assert fake_ray_cluster.calls_of("run") == [] + assert [handle.killed for handle in fake_ray_cluster.handles] == [True, True] + + class TestPortAllocationDetails: async def test_the_allocator_probes_the_workers_own_actor_on_its_own_node(self, fake_ray_cluster: FakeRayCluster): """Ports must be probed on the node that will bind them, through that worker's own actor."""