From 6e2e7f6d8d881a464b409c4b07c8900c23b05aff Mon Sep 17 00:00:00 2001 From: CAICAIIs <3360776475@qq.com> Date: Tue, 25 Aug 2026 17:30:18 +0800 Subject: [PATCH 1/5] fix(engine): allocate distributed init ports outside the ephemeral range find_free_port() probed bind(0) and closed the reservation immediately; the just-freed ephemeral port could be re-taken as the source port of an outbound connection, leaving it in TIME-WAIT so the later TCPStore server bind failed with EADDRINUSE on busy hosts, and consecutive calls could return the same port for the train and rollout clusters. Ports are now probed with a real bind from a fixed non-ephemeral range (which also excludes listeners and TIME-WAIT sockets), kept distinct per process, with a bind(0) fallback. Fixes #517 --- areno/engine/protocol.py | 32 +++++++++++++++++++++++++++- tests/test_parallel_partition_cpu.py | 19 +++++++++++++++++ 2 files changed, 50 insertions(+), 1 deletion(-) diff --git a/areno/engine/protocol.py b/areno/engine/protocol.py index dc6cbeae..add43b46 100644 --- a/areno/engine/protocol.py +++ b/areno/engine/protocol.py @@ -235,9 +235,39 @@ def result(self, timeout: float | None = None) -> list[Any]: return self._pending.results +_FREE_PORT_EXCLUSIONS: set[int] = set() +# Fixed range outside the kernel ephemeral range (32768-60999 on Linux) so a +# just-released reservation cannot be re-taken as the source port of an +# outbound connection, which would leave the port in TIME-WAIT and fail the +# later TCPStore server bind with EADDRINUSE (see #517). +_FREE_PORT_MIN = 20000 +_FREE_PORT_MAX = 30000 + + def find_free_port() -> int: - """Reserve an available localhost TCP port for torch distributed init.""" + """Reserve an available localhost TCP port for torch distributed init. + + Ports are probed with a real bind from a fixed non-ephemeral range (which + also excludes listeners and TIME-WAIT sockets on that address), and + consecutive calls in one process never return the same port, so the train + and rollout clusters cannot collide at TCPStore bind time. + """ + + import random as _random + for _ in range(128): + port = _random.randint(_FREE_PORT_MIN, _FREE_PORT_MAX) + if port in _FREE_PORT_EXCLUSIONS: + continue + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + try: + sock.bind(("127.0.0.1", port)) + except OSError: + continue # listener or TIME-WAIT socket already on this port + _FREE_PORT_EXCLUSIONS.add(port) + return port + # Fall back to the kernel ephemeral allocator if the fixed range is + # exhausted (unexpected on any real host). with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.bind(("127.0.0.1", 0)) return int(sock.getsockname()[1]) diff --git a/tests/test_parallel_partition_cpu.py b/tests/test_parallel_partition_cpu.py index 26fa4c19..7cc25d89 100644 --- a/tests/test_parallel_partition_cpu.py +++ b/tests/test_parallel_partition_cpu.py @@ -145,3 +145,22 @@ def test_real_gloo_tp_broadcast_uses_partition_global_root() -> None: 2: "rollout-root", 3: "rollout-root", } + + +def test_find_free_port_returns_distinct_bindable_ports() -> None: + """Consecutive calls return distinct, immediately bindable ports. + + The port must come from the fixed non-ephemeral range and be re-bindable + right away; ephemeral-range ports can be stolen by outbound traffic after + the probe closes, which fails the later TCPStore bind with EADDRINUSE + (#517). + """ + + import socket + + ports = [find_free_port() for _ in range(8)] + assert len(set(ports)) == len(ports), f"duplicate ports returned: {ports}" + for port in ports: + assert 20000 <= port < 30000, f"port {port} outside the fixed range" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.bind(("127.0.0.1", port)) From c753bc6a09f34959981840729ec949c822e649cc Mon Sep 17 00:00:00 2001 From: CAICAIIs <3360776475@qq.com> Date: Tue, 25 Aug 2026 19:17:37 +0800 Subject: [PATCH 2/5] fix(engine): half-open port range and thread-safe exclusion set Self-review fixes: - randint could return the inclusive upper bound 30000 while the test asserts port < 30000; make the range half-open [20000, 30000) - guard the process-local exclusion set with a lock so concurrent callers cannot hand out the same port - document the residual caveat for hosts that widen ip_local_port_range --- areno/engine/protocol.py | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/areno/engine/protocol.py b/areno/engine/protocol.py index add43b46..980950a9 100644 --- a/areno/engine/protocol.py +++ b/areno/engine/protocol.py @@ -236,10 +236,14 @@ def result(self, timeout: float | None = None) -> list[Any]: _FREE_PORT_EXCLUSIONS: set[int] = set() +_FREE_PORT_EXCLUSIONS_LOCK = threading.Lock() # Fixed range outside the kernel ephemeral range (32768-60999 on Linux) so a # just-released reservation cannot be re-taken as the source port of an # outbound connection, which would leave the port in TIME-WAIT and fail the -# later TCPStore server bind with EADDRINUSE (see #517). +# later TCPStore server bind with EADDRINUSE (see #517). Half-open [20000, +# 30000); hosts that widen ip_local_port_range beyond this range are not +# covered by the outbound-reuse argument, but the bind probe still rejects +# occupied ports at selection time. _FREE_PORT_MIN = 20000 _FREE_PORT_MAX = 30000 @@ -248,15 +252,16 @@ def find_free_port() -> int: """Reserve an available localhost TCP port for torch distributed init. Ports are probed with a real bind from a fixed non-ephemeral range (which - also excludes listeners and TIME-WAIT sockets on that address), and - consecutive calls in one process never return the same port, so the train - and rollout clusters cannot collide at TCPStore bind time. + also rejects listeners, including wildcard listeners, and TIME-WAIT + sockets on the probe address), and consecutive calls in one process never + return the same port, so the train and rollout clusters cannot collide at + TCPStore bind time. """ import random as _random for _ in range(128): - port = _random.randint(_FREE_PORT_MIN, _FREE_PORT_MAX) + port = _random.randint(_FREE_PORT_MIN, _FREE_PORT_MAX - 1) if port in _FREE_PORT_EXCLUSIONS: continue with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: @@ -264,7 +269,8 @@ def find_free_port() -> int: sock.bind(("127.0.0.1", port)) except OSError: continue # listener or TIME-WAIT socket already on this port - _FREE_PORT_EXCLUSIONS.add(port) + with _FREE_PORT_EXCLUSIONS_LOCK: + _FREE_PORT_EXCLUSIONS.add(port) return port # Fall back to the kernel ephemeral allocator if the fixed range is # exhausted (unexpected on any real host). From ab1225bcd021307becb50f05f34218c7b55713c5 Mon Sep 17 00:00:00 2001 From: CAICAIIs <3360776475@qq.com> Date: Thu, 27 Aug 2026 11:09:42 +0800 Subject: [PATCH 3/5] fix(engine): coordinator-held TCPStore replaces find_free_port Addresses maintainer review (#518): instead of probing a port with bind-close-bind (which raced with outbound traffic and could hand the same port to the train and rollout clusters), the coordinator now creates and retains a TCPStore with port=0 before spawning workers, and every worker joins the group as a TCPStore client on the resolved port. - protocol.py: _create_rendezvous_store() binds port=0 and is retained on the cluster for its lifetime; Cluster.start and start_partitioned_clusters resolve the store port into the shared world_spec before spawning - context.py: init_process_group joins via a client store and init_process_group(store=...), so no worker ever binds the rendezvous port - backend.py: rollout world_spec uses a placeholder port that start_partitioned_clusters fills in - tests: the gloo broadcast test mirrors production with a coordinator store; a new test asserts the resolved port is genuinely held (re-bind fails while the store is alive) and serves client stores find_free_port is removed; no public API change. --- areno/api/backend/cuda/backend.py | 5 ++- areno/engine/parallel/context.py | 6 ++- areno/engine/protocol.py | 64 +++++++++++----------------- tests/test_parallel_partition_cpu.py | 35 +++++++++------ 4 files changed, 54 insertions(+), 56 deletions(-) diff --git a/areno/api/backend/cuda/backend.py b/areno/api/backend/cuda/backend.py index 1b181708..119860af 100644 --- a/areno/api/backend/cuda/backend.py +++ b/areno/api/backend/cuda/backend.py @@ -178,7 +178,6 @@ def initialize(self, ctx: Context): from areno.engine.protocol import ( ClusterPartition, DistributedWorldSpec, - find_free_port, start_partitioned_clusters, ) @@ -204,7 +203,9 @@ def initialize(self, ctx: Context): ) world_spec = DistributedWorldSpec( master_addr="127.0.0.1", - master_port=find_free_port(), + # Placeholder: start_partitioned_clusters creates a coordinator-held + # TCPStore with port=0 and fills in the resolved port before spawn. + master_port=0, global_world_size=world_size + len(rollout_devices), train=train_partition, rollout=rollout_partition, diff --git a/areno/engine/parallel/context.py b/areno/engine/parallel/context.py index 47ab42c4..5c1f7ebe 100644 --- a/areno/engine/parallel/context.py +++ b/areno/engine/parallel/context.py @@ -113,9 +113,13 @@ def init_process_group( resolved_global_rank = rank if global_rank is None else global_rank resolved_global_world_size = world_size if global_world_size is None else global_world_size + # The coordinator holds the server-side rendezvous store on `master_port` + # (created with port=0 in protocol.py), so every worker joins as a TCPStore + # client instead of racing to bind the port itself (#517). + store = dist.TCPStore(master_addr, master_port, world_size=resolved_global_world_size, is_master=False) dist.init_process_group( backend=backend, - init_method=f"tcp://{master_addr}:{master_port}", + store=store, rank=resolved_global_rank, world_size=resolved_global_world_size, ) diff --git a/areno/engine/protocol.py b/areno/engine/protocol.py index 980950a9..b96228b6 100644 --- a/areno/engine/protocol.py +++ b/areno/engine/protocol.py @@ -13,7 +13,6 @@ import asyncio import multiprocessing as mp import queue -import socket import threading import traceback from dataclasses import dataclass @@ -235,48 +234,20 @@ def result(self, timeout: float | None = None) -> list[Any]: return self._pending.results -_FREE_PORT_EXCLUSIONS: set[int] = set() -_FREE_PORT_EXCLUSIONS_LOCK = threading.Lock() -# Fixed range outside the kernel ephemeral range (32768-60999 on Linux) so a -# just-released reservation cannot be re-taken as the source port of an -# outbound connection, which would leave the port in TIME-WAIT and fail the -# later TCPStore server bind with EADDRINUSE (see #517). Half-open [20000, -# 30000); hosts that widen ip_local_port_range beyond this range are not -# covered by the outbound-reuse argument, but the bind probe still rejects -# occupied ports at selection time. -_FREE_PORT_MIN = 20000 -_FREE_PORT_MAX = 30000 +def _create_rendezvous_store(master_addr: str, world_size: int): + """Create and retain a coordinator-side TCPStore with an OS-assigned port. - -def find_free_port() -> int: - """Reserve an available localhost TCP port for torch distributed init. - - Ports are probed with a real bind from a fixed non-ephemeral range (which - also rejects listeners, including wildcard listeners, and TIME-WAIT - sockets on the probe address), and consecutive calls in one process never - return the same port, so the train and rollout clusters cannot collide at - TCPStore bind time. + The store binds ``port=0`` and keeps the listening socket open for the + coordinator's lifetime, so no other process (including outbound traffic + reusing ephemeral ports) can steal the resolved port before the workers + connect as client stores. This replaces the bind-close-bind probe of the + old ``find_free_port``, which raced with outbound connections on busy + hosts and failed worker startup with EADDRINUSE (#517). """ - import random as _random + import torch.distributed as dist - for _ in range(128): - port = _random.randint(_FREE_PORT_MIN, _FREE_PORT_MAX - 1) - if port in _FREE_PORT_EXCLUSIONS: - continue - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - try: - sock.bind(("127.0.0.1", port)) - except OSError: - continue # listener or TIME-WAIT socket already on this port - with _FREE_PORT_EXCLUSIONS_LOCK: - _FREE_PORT_EXCLUSIONS.add(port) - return port - # Fall back to the kernel ephemeral allocator if the fixed range is - # exhausted (unexpected on any real host). - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: - sock.bind(("127.0.0.1", 0)) - return int(sock.getsockname()[1]) + return dist.TCPStore(master_addr, 0, world_size=world_size, is_master=True, wait_for_workers=False) def _rollout_payload_count(payload: RolloutPayload) -> int: @@ -328,6 +299,9 @@ def __init__( raise ValueError("world_spec and partition must be provided together") self.world_spec = world_spec self.partition = partition + # Coordinator-side rendezvous store; retained for the cluster lifetime + # so the resolved master port stays reserved for the workers (#517). + self._rendezvous_store = None # `spawn` start method is required by CUDA-aware workers; do not # inherit fds/CUDA state from the parent. self.ctx = mp.get_context("spawn") @@ -350,9 +324,12 @@ def start(self) -> None: if self.world_spec is not None: raise RuntimeError("partitioned clusters must be started with start_partitioned_clusters()") world_size = self.config.tp_size * int(self.config.dp_size) + # The coordinator holds the rendezvous store open so the resolved port + # is genuinely reserved until the workers connect (see #517). + self._rendezvous_store = _create_rendezvous_store("127.0.0.1", world_size) world_spec = DistributedWorldSpec( master_addr="127.0.0.1", - master_port=find_free_port(), + master_port=int(self._rendezvous_store.port), global_world_size=world_size, train=ClusterPartition( role="train", @@ -746,6 +723,13 @@ def start_partitioned_clusters( clusters = (train_cluster, rollout_cluster) if train_cluster.world_spec != world_spec or rollout_cluster.world_spec != world_spec: raise ValueError("both clusters must use the supplied world_spec") + # One coordinator-held store for the combined train + rollout world; the + # resolved port replaces the placeholder in the shared world_spec before + # any worker spawns (see #517). + store = _create_rendezvous_store(world_spec.master_addr, world_spec.global_world_size) + world_spec.master_port = int(store.port) + for cluster in clusters: + cluster._rendezvous_store = store try: for cluster in clusters: cluster._spawn_workers() diff --git a/tests/test_parallel_partition_cpu.py b/tests/test_parallel_partition_cpu.py index 7cc25d89..24ad9caf 100644 --- a/tests/test_parallel_partition_cpu.py +++ b/tests/test_parallel_partition_cpu.py @@ -2,11 +2,12 @@ import multiprocessing as mp +import pytest import torch from areno.engine.parallel import context from areno.engine.parallel.collectives import broadcast_object -from areno.engine.protocol import find_free_port +from areno.engine.protocol import _create_rendezvous_store def _run_offset_tp_broadcast(global_rank: int, port: int, output_queue) -> None: @@ -127,7 +128,10 @@ def test_single_engine_context_creates_no_policy_publisher_group(monkeypatch) -> def test_real_gloo_tp_broadcast_uses_partition_global_root() -> None: spawn = mp.get_context("spawn") output_queue = spawn.Queue() - port = find_free_port() + # The coordinator holds the server store (port=0) so the resolved port is + # genuinely reserved before the workers join as client stores. + store = _create_rendezvous_store("127.0.0.1", 4) + port = int(store.port) processes = [ spawn.Process(target=_run_offset_tp_broadcast, args=(global_rank, port, output_queue)) for global_rank in range(4) @@ -147,20 +151,25 @@ def test_real_gloo_tp_broadcast_uses_partition_global_root() -> None: } -def test_find_free_port_returns_distinct_bindable_ports() -> None: - """Consecutive calls return distinct, immediately bindable ports. +def test_rendezvous_store_resolves_and_holds_its_port() -> None: + """The coordinator store resolves a real port and keeps it bound. - The port must come from the fixed non-ephemeral range and be re-bindable - right away; ephemeral-range ports can be stolen by outbound traffic after - the probe closes, which fails the later TCPStore bind with EADDRINUSE - (#517). + The resolved port must be immediately usable by client stores, and binding + the same port again must fail while the server store is alive (the socket + is genuinely reserved, unlike the old bind-close-bind probe, #517). """ import socket - ports = [find_free_port() for _ in range(8)] - assert len(set(ports)) == len(ports), f"duplicate ports returned: {ports}" - for port in ports: - assert 20000 <= port < 30000, f"port {port} outside the fixed range" - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + import torch.distributed as dist + + store = _create_rendezvous_store("127.0.0.1", 2) + port = int(store.port) + assert port > 0 + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + with pytest.raises(OSError): sock.bind(("127.0.0.1", port)) + # A client store connects to the retained server. + client = dist.TCPStore("127.0.0.1", port, world_size=2, is_master=False) + store.set("k", "v") + assert client.get("k") == b"v" From 0fcfea8a978fe6e5418b553e8dbf2f1ce2ddb2f2 Mon Sep 17 00:00:00 2001 From: CAICAIIs <3360776475@qq.com> Date: Thu, 27 Aug 2026 11:51:53 +0800 Subject: [PATCH 4/5] test: use coordinator-held store in policy tensor sync gloo tests CI caught that test_policy_tensor_sync_cpu.py still imported the removed find_free_port; both real-gloo reshuffle tests now create a coordinator-held rendezvous store and pass its resolved port to the spawned workers, mirroring the production rendezvous path. --- tests/test_policy_tensor_sync_cpu.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/tests/test_policy_tensor_sync_cpu.py b/tests/test_policy_tensor_sync_cpu.py index 12b26abb..82d2a7c9 100644 --- a/tests/test_policy_tensor_sync_cpu.py +++ b/tests/test_policy_tensor_sync_cpu.py @@ -22,7 +22,7 @@ build_adapter_policy_plan, transfer_policy_weights, ) -from areno.engine.protocol import PolicySyncPayload, find_free_port +from areno.engine.protocol import PolicySyncPayload, _create_rendezvous_store def _set_rank(rank: int, world_size: int) -> None: @@ -353,7 +353,9 @@ def _gloo_policy_sync_worker(global_rank: int, port: int, output_queue) -> None: def test_real_gloo_collectives_reshard_train_tp2_to_rollout_tp1() -> None: ctx = mp.get_context("spawn") output_queue = ctx.Queue() - port = find_free_port() + # Coordinator-held store mirrors production: workers join as client stores. + store = _create_rendezvous_store("127.0.0.1", 3) + port = int(store.port) processes = [ctx.Process(target=_gloo_policy_sync_worker, args=(rank, port, output_queue)) for rank in range(3)] for process in processes: process.start() @@ -402,7 +404,9 @@ def _gloo_policy_sync_reverse_worker(global_rank: int, port: int, output_queue) def test_real_gloo_collectives_reshard_train_tp1_to_rollout_tp2() -> None: ctx = mp.get_context("spawn") output_queue = ctx.Queue() - port = find_free_port() + # Coordinator-held store mirrors production: workers join as client stores. + store = _create_rendezvous_store("127.0.0.1", 3) + port = int(store.port) processes = [ ctx.Process(target=_gloo_policy_sync_reverse_worker, args=(rank, port, output_queue)) for rank in range(3) ] From 9ce7fed8e367858513e4ef659f71cb278294ffdf Mon Sep 17 00:00:00 2001 From: CAICAIIs <3360776475@qq.com> Date: Thu, 27 Aug 2026 14:01:55 +0800 Subject: [PATCH 5/5] test: mock the client TCPStore in group-construction tests The store-based rendezvous now creates a real client TCPStore before calling init_process_group; the two mocked tests patched only init_process_group, so the client store tried to connect to the placeholder port 12345 (no server) and blocked for the connection timeout. Patch TCPStore alongside init_process_group so the tests keep exercising group construction only. --- tests/test_parallel_partition_cpu.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/test_parallel_partition_cpu.py b/tests/test_parallel_partition_cpu.py index 24ad9caf..83d25eaf 100644 --- a/tests/test_parallel_partition_cpu.py +++ b/tests/test_parallel_partition_cpu.py @@ -38,6 +38,9 @@ def _run_offset_tp_broadcast(global_rank: int, port: int, output_queue) -> None: def _mock_distributed(monkeypatch): calls = [] monkeypatch.setattr(torch.cuda, "is_available", lambda: False) + # These tests exercise group construction only, not the rendezvous, so + # the client TCPStore and the process-group init are both faked. + monkeypatch.setattr(context.dist, "TCPStore", lambda *args, **kwargs: object()) monkeypatch.setattr(context.dist, "init_process_group", lambda **kwargs: calls.append(("init", kwargs))) def new_group(*, ranks):