cultmesh-py: bind the TCP+UDP pair for port 0 without the race
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 4m46s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 46s
Some checks failed
CultMesh QUIC native / Build win32-x64 (pull_request) Has been skipped
CultMesh QUIC native / Build linux-x64 (pull_request) Has been skipped
CultMesh QUIC native / Release native binaries (pull_request) Has been skipped
CultMesh portability and performance / ubuntu-latest reactive semantics and leases (pull_request) Failing after 4m46s
CultMesh QUIC native / Verify committed binaries match SHA256SUMS (pull_request) Successful in 46s
CultMeshLocalServer.start bound TCP to port 0 and then bound UDP to whatever number the kernel picked, failing with EADDRINUSE whenever another process held that number for UDP (this host has hundreds of UDP listeners), and left the accept thread running on failure. Port 0 now retries with a fresh pair (bounded); both sockets bind before any thread starts; an explicit port still fails honestly. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
e2dc04e724
commit
e0f5e13a1b
2 changed files with 97 additions and 8 deletions
|
|
@ -1,5 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
import logging
|
||||
import socket
|
||||
import threading
|
||||
|
|
@ -57,6 +58,7 @@ from cultnet_py.cultmesh_contracts import (
|
|||
)
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
_EPHEMERAL_PAIR_ATTEMPTS = 64
|
||||
|
||||
|
||||
class _ReplyUnqueueable(Exception):
|
||||
|
|
@ -168,26 +170,49 @@ class CultMeshLocalServer:
|
|||
def start(self) -> "CultMeshLocalServer":
|
||||
if self._socket is not None:
|
||||
return self
|
||||
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
server.bind((self.host, self.port))
|
||||
server.listen()
|
||||
server, rudp_socket = self._bind_sockets()
|
||||
server.settimeout(0.2)
|
||||
self._socket = server
|
||||
self.port = int(server.getsockname()[1])
|
||||
self._stop.clear()
|
||||
self._thread = threading.Thread(target=self._accept_loop, daemon=True)
|
||||
self._thread.start()
|
||||
if self.enable_rudp:
|
||||
rudp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
rudp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
rudp_socket.bind((self.host, self.port))
|
||||
if rudp_socket is not None:
|
||||
rudp_socket.settimeout(0.02)
|
||||
self._rudp_socket = rudp_socket
|
||||
self._rudp_thread = threading.Thread(target=self._rudp_loop, daemon=True)
|
||||
self._rudp_thread.start()
|
||||
return self
|
||||
|
||||
def _bind_sockets(self) -> tuple[socket.socket, socket.socket | None]:
|
||||
"""The listening TCP socket and, when RUDP is on, the UDP socket on the same port number.
|
||||
|
||||
An explicit port is bound as asked and a failure is the caller's. Port 0 asks for any free port, and the
|
||||
kernel picks the TCP one without knowing whether another process holds that number for UDP, so a
|
||||
UDP collision retries with a fresh TCP port instead of failing the start. Both sockets are bound
|
||||
before any thread starts, so a failed start leaves nothing running or open.
|
||||
"""
|
||||
attempts = _EPHEMERAL_PAIR_ATTEMPTS if self.port == 0 and self.enable_rudp else 1
|
||||
for attempt in range(attempts):
|
||||
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
rudp_socket = None
|
||||
try:
|
||||
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
server.bind((self.host, self.port))
|
||||
server.listen()
|
||||
if self.enable_rudp:
|
||||
rudp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
rudp_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
rudp_socket.bind((self.host, int(server.getsockname()[1])))
|
||||
return server, rudp_socket
|
||||
except OSError as error:
|
||||
server.close()
|
||||
if rudp_socket is not None:
|
||||
rudp_socket.close()
|
||||
if error.errno != errno.EADDRINUSE or attempt == attempts - 1:
|
||||
raise
|
||||
raise AssertionError("unreachable")
|
||||
|
||||
def stop(self) -> None:
|
||||
self._stop.set()
|
||||
if self._rudp_socket is not None:
|
||||
|
|
|
|||
64
packages/cultmesh-py/tests/test_server_bind.py
Normal file
64
packages/cultmesh-py/tests/test_server_bind.py
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
"""A server asked for any port finds a TCP+UDP pair; one asked for a port gets that port or an honest failure."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import errno
|
||||
import socket
|
||||
import threading
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
from cultmesh_py import create_node
|
||||
from cultmesh_py.server import CultMeshLocalServer
|
||||
|
||||
|
||||
def server(**kwargs) -> CultMeshLocalServer:
|
||||
return CultMeshLocalServer(node=create_node(runtime_id="bind-test"), host="127.0.0.1", **kwargs)
|
||||
|
||||
|
||||
class ServerBindTests(unittest.TestCase):
|
||||
def test_port_zero_retries_with_a_fresh_port_when_another_process_holds_the_udp_number(self) -> None:
|
||||
real_bind = socket.socket.bind
|
||||
refused: list[int] = []
|
||||
|
||||
def bind(sock: socket.socket, address) -> None:
|
||||
# The first two UDP binds find the number taken, as a UDP listener of another process would.
|
||||
if sock.type == socket.SOCK_DGRAM and len(refused) < 2:
|
||||
refused.append(address[1])
|
||||
raise OSError(errno.EADDRINUSE, "Address already in use")
|
||||
real_bind(sock, address)
|
||||
|
||||
with patch.object(socket.socket, "bind", bind):
|
||||
started = server().start()
|
||||
try:
|
||||
self.assertEqual(len(refused), 2)
|
||||
self.assertNotIn(started.port, refused[:1], "the retry used a fresh TCP port")
|
||||
probe = socket.create_connection(("127.0.0.1", started.port), timeout=2.0)
|
||||
probe.close()
|
||||
finally:
|
||||
started.stop()
|
||||
|
||||
def test_an_explicit_port_whose_udp_number_is_taken_fails_and_leaves_nothing_running(self) -> None:
|
||||
holder = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
holder.bind(("127.0.0.1", 0))
|
||||
port = holder.getsockname()[1]
|
||||
threads_before = threading.active_count()
|
||||
try:
|
||||
failing = server(port=port)
|
||||
with self.assertRaises(OSError) as raised:
|
||||
failing.start()
|
||||
self.assertEqual(raised.exception.errno, errno.EADDRINUSE)
|
||||
self.assertIsNone(failing._socket)
|
||||
self.assertEqual(threading.active_count(), threads_before, "no accept thread was left running")
|
||||
# The TCP number was released, not left listening.
|
||||
free = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
try:
|
||||
free.bind(("127.0.0.1", port))
|
||||
finally:
|
||||
free.close()
|
||||
finally:
|
||||
holder.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in a new issue