Ensure only one daemon can run at a time (#622)

* Substitute SO_REUSEADDR by SO_LINGER.
* Wait for daemon shutdown before testing NodeStrategy

Signed-off-by: Michel Hidalgo <[email protected]>

ros2cli/ros2cli/node/daemon.py

@@ -16,6 +16,7 @@
 import os
 import platform
 import socket
+import struct
 import subprocess
 import sys
 import time
@@ -29,15 +30,14 @@
 
 def is_daemon_running(args):
     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-    if platform.system() != 'Windows':
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
     addr = ('localhost', get_daemon_port())
     try:
         s.bind(addr)
     except socket.error as e:
         if e.errno == errno.EADDRINUSE:
             return True
-    else:
+    finally:
         s.close()
     return False
 
@@ -132,5 +132,33 @@ def __exit__(self, exc_type, exc_value, traceback):
         self._proxy.__exit__(exc_type, exc_value, traceback)
 
 
+def shutdown_daemon(args, wait_duration=None):
+    """
+    Shuts down remote daemon node.
+
+    :param args: DaemonNode arguments namespace.
+    :param wait_duration: optional duration, in seconds, to wait
+      until the daemon node is fully shut down. Non-positive
+      durations will result in an indefinite wait.
+    :return: whether the daemon is shut down already or not.
+    """
+    with DaemonNode(args) as node:
+        node.system.shutdown()
+
+    if wait_duration is None:
+        return not is_daemon_running(args)
+
+    if wait_duration > 0.0:
+        timeout = time.time() + wait_duration
+    else:
+        timeout = None
+
+    while is_daemon_running(args):
+        time.sleep(0.1)
+        if timeout and time.time() >= timeout:
+            return False
+    return True
+
+
 def add_arguments(parser):
     pass

ros2cli/ros2cli/xmlrpc/local_server.py

@@ -12,12 +12,29 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import socket
+import struct
 from xmlrpc.server import SimpleXMLRPCRequestHandler
 from xmlrpc.server import SimpleXMLRPCServer
 
 
 class LocalXMLRPCServer(SimpleXMLRPCServer):
 
+    allow_reuse_address = False
+
+    def server_bind(self):
+        # Prevent listening socket from lingering in TIME_WAIT state after close()
+        self.socket.setsockopt(
+            socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
+        super(LocalXMLRPCServer, self).server_bind()
+
+    def get_request(self):
+        # Prevent accepted socket from lingering in TIME_WAIT state after close()
+        sock, addr = super(LocalXMLRPCServer, self).get_request()
+        sock.setsockopt(
+            socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
+        return sock, addr
+
     def verify_request(self, request, client_address):
         if client_address[0] != '127.0.0.1':
             return False

ros2cli/test/test_strategy.py

@@ -16,17 +16,16 @@
 
 import pytest
 
-from ros2cli.node.daemon import DaemonNode
 from ros2cli.node.daemon import is_daemon_running
+from ros2cli.node.daemon import shutdown_daemon
 from ros2cli.node.daemon import spawn_daemon
 from ros2cli.node.strategy import NodeStrategy
 
 
 @pytest.fixture
 def enforce_no_daemon_is_running():
     if is_daemon_running(args=[]):
-        with DaemonNode(args=[]) as node:
-            node.system.shutdown()
+        assert shutdown_daemon(args=[], wait_duration=5.0)
     yield
 
 
@@ -53,8 +52,6 @@ def test_with_no_daemon_running(enforce_no_daemon_is_running):
 
 
 def test_enforce_no_daemon(enforce_daemon_is_running):
-    if not is_daemon_running(args=[]):
-        assert spawn_daemon(args=[], wait_until_spawned=5.0)
     args = argparse.Namespace(no_daemon=True)
     with NodeStrategy(args=args) as node:
         assert node._daemon_node is None