fix(vrtManager): the libvirt event loop starts per process, on first use

The connection manager registered libvirt's event implementation and
started the event loop thread when vrtManager.connection was imported.
novncd imports it and then forks a child per console (websockify's
multiprocessing), so every child was forked from a process with a
running libvirt thread, which libvirt does not support, and the child
had no loop thread of its own: keepalive never ran there. Measured in
the test setup: in such a child, a call on a host that went silent
still waited after 70 s.

- The loop starts with the first get_connection of a process. The
  state is per process and shared by all managers; libvirt's event
  implementation is registered once and a child inherits it.
- After a fork, the child renews the lock and starts its own loop on
  first use. It stops using the inherited connections, which are the
  parent's sockets, but keeps them: releasing them would run libvirt
  destructors in the child. No process connects before forking today
  (novncd's parent does not connect, gunicorn does not preload); this
  keeps such a case safe.

With the change, novncd's parent has no libvirt thread, the console
child starts its own, and the same call fails after 29 s with a
keepalive timeout. A test forks a real child; it fails without the
fork handling.

vrtManager/connection.py

@@ -38,11 +38,6 @@ class wvmEventLoop(threading.Thread):
     """ Event Loop Class"""
 
     def __init__(self, group=None, target=None, name=None, args=(), kwargs={}):
-        # register the default event implementation
-        # of libvirt, as we do not have an existing
-        # event loop.
-        libvirt.virEventRegisterDefaultImpl()
-
         if name is None:
             name = "libvirt event loop"
 
@@ -60,6 +55,38 @@ class wvmEventLoop(threading.Thread):
             libvirt.virEventRunDefaultImpl()
 
 
+# The event loop (keepalive, close callbacks) starts with the first connection
+# of a process, not at import: novncd imports this module and then forks a child
+# per console, and libvirt must not be used after a fork of a process with
+# running libvirt threads. Threads do not survive a fork, so a child starts its
+# own loop. This is process state, shared by all connection managers.
+_event_loop = {"lock": threading.Lock(), "running": False, "registered": False}
+
+
+def start_event_loop():
+    if _event_loop["running"]:
+        return
+    with _event_loop["lock"]:
+        if _event_loop["running"]:
+            return
+        if not _event_loop["registered"]:
+            # the registration is libvirt's process state and a child inherits it
+            libvirt.virEventRegisterDefaultImpl()
+            _event_loop["registered"] = True
+        wvmEventLoop().start()
+        _event_loop["running"] = True
+
+
+def _after_fork_in_child():
+    # the loop thread and any thread holding these locks are gone
+    _event_loop["lock"] = threading.Lock()
+    _event_loop["running"] = False
+    connection_manager.forget_connections()
+
+
+os.register_at_fork(after_in_child=_after_fork_in_child)
+
+
 class wvmConnection(object):
     """
     class representing a single connection stored in the Connection Manager
@@ -251,10 +278,15 @@ class wvmConnectionManager(object):
         #     http://wiki.libvirt.org/page/FAQ#Is_libvirt_thread_safe.3F
         self._connections = dict()
         self._connections_lock = ReadWriteLock()
+        self._inherited = []
 
-        # start event loop to handle keepalive requests and other events
-        self._event_loop = wvmEventLoop()
-        self._event_loop.start()
+    def forget_connections(self):
+        """For a forked child: stop using the inherited connections (the
+        parent's sockets). They are kept, not released: their destructors
+        would call libvirt in the child."""
+        self._inherited.append(self._connections)
+        self._connections = dict()
+        self._connections_lock = ReadWriteLock()
 
     def _search_connection(self, host, login, passwd, conn):
         """
@@ -279,6 +311,7 @@ class wvmConnectionManager(object):
         returns a connection object (as returned by the libvirt.open* methods) for the given host and credentials
         raises libvirtError if (re)connecting fails
         """
+        start_event_loop()
         # force all string values to unicode
         host = str(host)
         login = str(login)

vrtManager/test_connection.py

@@ -13,6 +13,7 @@ if not settings.configured:
     settings.configure(MAC_OUI="52:54:10")
 
 import libvirt
+from vrtManager import connection
 from vrtManager.connection import CONN_SSH, CONN_TCP, SSH_COMMAND, wvmConnectionManager
 
 
@@ -206,5 +207,63 @@ class BoundedOpenTestCase(unittest.TestCase):
             self.assertIn("ConnectTimeout", script.read())
 
 
+class EventLoopTestCase(unittest.TestCase):
+    """The libvirt event loop thread starts with the first connection of a
+    process, not at import: novncd forks a child per console after importing
+    this module, and libvirt is not safe to use after a fork of a process
+    with running libvirt threads. A child starts its own loop, so its
+    keepalive works."""
+
+    def setUp(self):
+        for name in ("wvmEventLoop", "libvirt.virEventRegisterDefaultImpl", "libvirt.openAuth"):
+            patcher = patch(f"vrtManager.connection.{name}")
+            setattr(self, name.split(".")[-1], patcher.start())
+            self.addCleanup(patcher.stop)
+        self.openAuth.side_effect = lambda *args: alive_connection()
+        state = patch.dict(connection._event_loop, {"lock": threading.Lock(), "running": False, "registered": False})
+        state.start()
+        self.addCleanup(state.stop)
+
+    def test_no_thread_until_the_first_connection(self):
+        manager = wvmConnectionManager()
+        self.wvmEventLoop.assert_not_called()
+        self.virEventRegisterDefaultImpl.assert_not_called()
+        manager.get_connection("h", "u", "p", CONN_TCP)
+        wvmConnectionManager().get_connection("h2", "u", "p", CONN_TCP)  # one loop per process
+        self.wvmEventLoop.return_value.start.assert_called_once()
+        self.virEventRegisterDefaultImpl.assert_called_once()
+
+    def test_a_forked_child_starts_its_own_loop_with_its_own_connections(self):
+        manager = connection.connection_manager
+        parent_conn = manager.get_connection("fork-host", "u", "p", CONN_TCP)
+        read_end, write_end = os.pipe()
+        pid = os.fork()
+        if pid == 0:  # child: report and leave without running the test runner
+            try:
+                child_conn = manager.get_connection("fork-host", "u", "p", CONN_TCP)
+                kept = any(
+                    entry.connection is parent_conn
+                    for connections in manager._inherited
+                    for entries in connections.values()
+                    for entry in entries
+                )
+                report = (
+                    f"{self.wvmEventLoop.return_value.start.call_count} "
+                    f"{self.virEventRegisterDefaultImpl.call_count} {child_conn is parent_conn} {kept}"
+                )
+            except BaseException as e:
+                report = repr(e)
+            os.write(write_end, report.encode())
+            os._exit(0)
+        os.close(write_end)
+        os.waitpid(pid, 0)
+        report = os.read(read_end, 200).decode()
+        os.close(read_end)
+        # a second loop started in the child, on the inherited registration;
+        # the child opened its own connection instead of the parent's socket,
+        # and kept the parent's (releasing it would run libvirt destructors)
+        self.assertEqual(report, "2 1 False True")
+
+
 if __name__ == "__main__":
     unittest.main()