diff --git a/vrtManager/connection.py b/vrtManager/connection.py index c4933a88..b777d18c 100644 --- a/vrtManager/connection.py +++ b/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) diff --git a/vrtManager/test_connection.py b/vrtManager/test_connection.py index dcdafeb7..e2e912a3 100644 --- a/vrtManager/test_connection.py +++ b/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()