Mjcarroll/windows socket exhaustion

ros2cli/ros2cli/node/daemon.py

@@ -77,11 +77,10 @@ def is_daemon_running(args):
 
 def _is_daemon_address_free():
     # Mirror LocalXMLRPCServer.allow_reuse_address: SO_REUSEADDR on
-    # non-Windows so TIME_WAIT doesn't make us falsely report busy.
+    # all platforms so TIME_WAIT doesn't make us falsely report busy.
     try:
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-            if os.name != 'nt':
-                s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
             s.bind(daemon.get_address())
         return True
     except socket.error as e:
@@ -140,6 +139,9 @@ def spawn_daemon(args, timeout=None, debug=False):
       `False` if it was already running.
     :raises: if it fails to spawn the daemon.
     """
+    if is_daemon_running(args):
+        return False
+
     # Acquire socket by instantiating XMLRPC server.
     try:
         server = daemon.make_xmlrpc_server()

ros2cli/ros2cli/xmlrpc/local_server.py

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import os
 import socket
 # Import SimpleXMLRPCRequestHandler to re-export it.
 from xmlrpc.server import SimpleXMLRPCRequestHandler  # noqa
@@ -33,11 +32,8 @@ def get_local_ipaddrs():
 class LocalXMLRPCServer(SimpleXMLRPCServer):
 
     # Allow re-binding even if another server instance was recently bound (i.e. we are still in
-    # TCP TIME_WAIT). This is already the default behavior on Windows, and further SO_REUSEADDR can
-    # lead to undefined behavior on Windows; see
-    # https://learn.microsoft.com/en-us/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse.  # noqa
-    # So we don't set the option for Windows.
-    allow_reuse_address = False if os.name == 'nt' else True
+    # TCP TIME_WAIT).
+    allow_reuse_address = True
 
     def verify_request(self, request, client_address):
         if client_address[0] not in get_local_ipaddrs():

ros2cli/test/test_ros2cli_daemon.py

@@ -116,16 +116,20 @@ def noop_execute_callback(goal_handle):
             action_name=TEST_ACTION_NAME,
             execute_callback=noop_execute_callback
         )
-        action_server  # to avoid "assigned by never used" warning
         action_client = rclpy.action.ActionClient(
             node=node,
             action_type=test_msgs.action.Fibonacci,
             action_name=TEST_ACTION_NAME
         )
-        action_client  # to avoid "assigned by never used" warning
 
         yield node
 
+        # Teardown: explicitly destroy the node to make sure that any
+        # lingering middleware resources are freed (specifically sockets)
+        action_client.destroy()
+        action_server.destroy()
+        node.destroy_node()
+
 
 @pytest.fixture(scope='module')
 def daemon_node():

ros2cli/test/test_strategy.py

@@ -24,16 +24,24 @@
 
 @pytest.fixture
 def enforce_no_daemon_is_running():
+    # Setup phase: enforce no daemon running
     if is_daemon_running(args=[]):
         assert shutdown_daemon(args=[], timeout=5.0)
     yield
+    # Teardown phase: enforce no daemon left over
+    if is_daemon_running(args=[]):
+        assert shutdown_daemon(args=[], timeout=5.0)
 
 
 @pytest.fixture
 def enforce_daemon_is_running():
+    # Setup phase: enforce the daemon is running
     if not is_daemon_running(args=[]):
         assert spawn_daemon(args=[], timeout=5.0)
     yield
+    # Teardown phase: enforce no daemon left over
+    if is_daemon_running(args=[]):
+        assert shutdown_daemon(args=[], timeout=5.0)
 
 
 def test_with_daemon_running(enforce_daemon_is_running):

ros2multicast/test/test_api.py

@@ -13,6 +13,8 @@
 # limitations under the License.
 
 import random
+import re
+import subprocess
 import sys
 import threading
 import time
@@ -22,29 +24,86 @@
 from ros2multicast.api import receive
 from ros2multicast.api import send
 
+_MULTICAST_ADDRESS = re.compile(r'\b2(?:2[4-9]|3[0-9])(?:\.\d{1,3}){3}\b')
+_UDP_ENDPOINT = re.compile(r'^\s*(?:UDP|udp)\s+\S', re.MULTILINE)
+
+
+def _capture(argv):
+    try:
+        return subprocess.run(argv, capture_output=True, text=True, timeout=30).stdout
+    except (OSError, subprocess.SubprocessError):
+        return ''
+
+
+def _network_state():
+    """
+    Summarize the machine-wide network resources these tests draw on.
+
+    Both the UDP port pool and the multicast membership table are shared by every process
+    on the machine.  Once either is exhausted, binding and joining fail for everything on
+    the box, which is easy to mistake for a defect in the code under test.
+    See https://github.com/ros2/ros2cli/issues/1141.
+    """
+    if sys.platform.startswith('win'):
+        joins = _capture(['netsh', 'interface', 'ipv4', 'show', 'joins'])
+        endpoints = _capture(['netstat', '-ano', '-p', 'UDP'])
+        port_range = '+'.join(re.findall(
+            r':\s*(\d+)',
+            _capture(['netsh', 'int', 'ipv4', 'show', 'dynamicport', 'udp'])))
+    else:
+        joins = _capture(['netstat', '--groups', '--numeric'])
+        endpoints = _capture(['ss', '-uan'])
+        port_range = _capture(['sysctl', '-n', 'net.ipv4.ip_local_port_range']).strip()
+    return (
+        f'{len(_MULTICAST_ADDRESS.findall(joins))} multicast memberships, '
+        f'{len(_UDP_ENDPOINT.findall(endpoints))} udp endpoints open, '
+        f'dynamic port range: {port_range or "unknown"}'
+    )
+
+
+def _reraise(error):
+    """Re-raise a socket error, first recording the machine state that may have caused it."""
+    print(
+        f'ros2multicast socket operation failed ({error}); machine network state: '
+        f'{_network_state()}',
+        file=sys.stderr)
+    raise error
+
 
 def _send_receive(sent_data, rx_kwargs, tx_kwargs):
     received_data = None
+    rx_error = None
 
     def target():
-        nonlocal received_data
+        nonlocal received_data, rx_error
         try:
             received_data, _ = receive(**rx_kwargs)
         except TimeoutError:
             pass
+        except Exception as e:  # noqa: B902
+            # Stash it for the calling thread. Left unhandled here, pytest turns it into
+            # an unhandled-thread-exception warning, so the test either fails with a
+            # misleading assertion or passes when it should not.
+            rx_error = e
 
     t = threading.Thread(target=target)
     t.start()
     time.sleep(0.1)
-    send(sent_data, **tx_kwargs)
-    t.join()
+    try:
+        send(sent_data, **tx_kwargs)
+    except OSError as e:
+        _reraise(e)
+    finally:
+        t.join()
+    if rx_error is not None:
+        _reraise(rx_error)
     return received_data
 
 
 def test_api():
     sent_data = b'test_api'
 
-    rx_kwargs = {'timeout': 1.0}
+    rx_kwargs = {'timeout': 0.2}
     tx_kwargs = {}
 
     assert sent_data == _send_receive(sent_data, rx_kwargs, tx_kwargs)
@@ -84,13 +143,11 @@ def test_group_mismatch():
     try:
         assert _send_receive(sent_data, rx_kwargs, tx_kwargs) is None
     except OSError as e:
-        if sys.platform.startswith('win'):
-            if 10051 == e.winerror:
-                # TODO(sloretz) understand why this test fails this way in CI
-                # "A socket operation was attempted to an unreachable network"
-                pytest.skip('Unknown why this OSError occurs on Windows')
-        else:
-            raise
+        if sys.platform.startswith('win') and 10051 == e.winerror:
+            # TODO(sloretz) understand why this test fails this way in CI
+            # "A socket operation was attempted to an unreachable network"
+            pytest.skip('Unknown why this OSError occurs on Windows')
+        raise
 
 
 def test_port_mismatch():