From dba776933884c67bb060b87eb8f3dd1d234a0e99 Mon Sep 17 00:00:00 2001 From: Tom Chen Date: Tue, 11 Aug 2026 21:52:46 +0800 Subject: [PATCH] Start dynamic worker ports above Ray's worker range --- miles/utils/workers/addr_allocator.py | 4 ++- tests/fast/utils/workers/conftest.py | 12 ++++++++ .../fast/utils/workers/test_addr_allocator.py | 29 ++++++++++++++++++- 3 files changed, 43 insertions(+), 2 deletions(-) diff --git a/miles/utils/workers/addr_allocator.py b/miles/utils/workers/addr_allocator.py index 897ce29cd68..dba168caf11 100644 --- a/miles/utils/workers/addr_allocator.py +++ b/miles/utils/workers/addr_allocator.py @@ -5,6 +5,8 @@ logger = logging.getLogger(__name__) +_DYNAMIC_PORT_START = 20000 + @dataclass class PortAllocator: @@ -13,7 +15,7 @@ class PortAllocator: def alloc(self, actor, *, node_ip: str, consecutive: int = 1) -> int: # use small ports to prevent ephemeral port between 32768 and 65536. # also, ray uses port 10002-19999, thus we avoid near-10002 to avoid racing condition - start_port = self._next_port_of_ip.get(node_ip, 15000) + start_port = self._next_port_of_ip.get(node_ip, _DYNAMIC_PORT_START) port: int = ray.get( actor._get_free_port_block.remote( start_port=start_port, diff --git a/tests/fast/utils/workers/conftest.py b/tests/fast/utils/workers/conftest.py index acebb253baf..30964e2e3eb 100644 --- a/tests/fast/utils/workers/conftest.py +++ b/tests/fast/utils/workers/conftest.py @@ -8,3 +8,15 @@ def patch_ray_get(monkeypatch): import miles.utils.workers.addr_allocator as mod monkeypatch.setattr(mod.ray, "get", lambda x: x) + + +@pytest.fixture +def patch_ray_get_failure(monkeypatch): + """Make ``ray.get(...)`` raise, mimicking a probe that is submitted + successfully but fails while its result is retrieved.""" + import miles.utils.workers.addr_allocator as mod + + def _raise(_object_ref): + raise RuntimeError("free port probe failed") + + monkeypatch.setattr(mod.ray, "get", _raise) diff --git a/tests/fast/utils/workers/test_addr_allocator.py b/tests/fast/utils/workers/test_addr_allocator.py index cb8c60713a8..c47e380a2ce 100644 --- a/tests/fast/utils/workers/test_addr_allocator.py +++ b/tests/fast/utils/workers/test_addr_allocator.py @@ -1,5 +1,7 @@ from __future__ import annotations +import pytest + from tests.fast.ray.rollout.conftest import fake_engine from miles.utils.workers.addr_allocator import PortAllocator @@ -24,7 +26,7 @@ def test_alloc_starts_from_the_base_port_on_an_unseen_node(self, patch_ray_get): """A node with no cursor yet starts at the base port, away from ray's range.""" cursors = PortAllocator() engine = fake_engine(host="10.0.0.1", port_seed=0) - assert cursors.alloc(engine, node_ip="10.0.0.1") == 15000 + assert cursors.alloc(engine, node_ip="10.0.0.1") == 20000 def test_alloc_consecutive_reserves_a_whole_block(self, patch_ray_get): """A consecutive=N allocation must move this node's cursor past the entire block.""" @@ -33,6 +35,31 @@ def test_alloc_consecutive_reserves_a_whole_block(self, patch_ray_get): first = cursors.alloc(engine, node_ip="10.0.0.1", consecutive=5) assert cursors._next_port_of_ip["10.0.0.1"] == first + 5 + def test_a_skipped_candidate_range_advances_from_the_returned_block(self, patch_ray_get): + """When the actor skips occupied candidates, the cursor must advance from the returned port plus the block size.""" + cursors = PortAllocator() + engine = fake_engine(host="10.0.0.1", port_seed=25000) + first = cursors.alloc(engine, node_ip="10.0.0.1", consecutive=4) + assert first == 25000 + assert cursors._next_port_of_ip["10.0.0.1"] == 25004 + + def test_a_failed_probe_does_not_advance_the_node_cursor(self, patch_ray_get_failure): + """A probe that fails on result retrieval propagates its error and leaves this node's cursor untouched.""" + cursors = PortAllocator() + cursors._next_port_of_ip["10.0.0.1"] = 20005 + engine = fake_engine(host="10.0.0.1", port_seed=0) + with pytest.raises(RuntimeError, match="free port probe failed"): + cursors.alloc(engine, node_ip="10.0.0.1", consecutive=3) + assert cursors._next_port_of_ip == {"10.0.0.1": 20005} + + def test_a_failed_probe_does_not_create_a_cursor_for_an_unseen_node(self, patch_ray_get_failure): + """A probe that fails on result retrieval leaves a never-seen node without any cursor.""" + cursors = PortAllocator() + engine = fake_engine(host="10.0.0.1", port_seed=0) + with pytest.raises(RuntimeError, match="free port probe failed"): + cursors.alloc(engine, node_ip="10.0.0.1", consecutive=3) + assert cursors._next_port_of_ip == {} + def test_alloc_tracks_nodes_independently(self, patch_ray_get): """Each node ip owns its own cursor.""" cursors = PortAllocator()