From 2d47f87dcb2f70f2b118eb91a055c1e8656cb00c Mon Sep 17 00:00:00 2001 From: catborise Date: Mon, 2 Nov 2026 21:30:56 +0300 Subject: [PATCH] feat(vrtManager): a host that does not answer holds two requests of a worker With threaded workers a request waiting on a stuck host holds one thread, but each new request to that host (a VM list's poll every 5 s, an F5) held another until the threads ran out. In a process at most two web requests now use one host at a time, counted by its name whatever the login. Another waits up to the connect timeout (5 s) for one of them to end, and is then refused as the host not answering: VM lists show the host as not reachable, pages show the connection error. A request counts on the host it connected to last, until it ends, so one that reads several hosts in turn holds none it has left. Threads outside a request (the job worker's, the consoles') are not counted: a job holds its host for as long as it runs. --- vrtManager/connection.py | 52 +++++++++++++++++ vrtManager/tests/test_connection.py | 86 +++++++++++++++++++++++++++++ 2 files changed, 138 insertions(+) diff --git a/vrtManager/connection.py b/vrtManager/connection.py index d3a7c6b5..ccd2191f 100644 --- a/vrtManager/connection.py +++ b/vrtManager/connection.py @@ -10,6 +10,7 @@ from urllib.parse import quote import libvirt from django.conf import settings +from django.core.signals import request_finished, request_started from django.utils.functional import cached_property from libvirt import libvirtError from lxml import etree @@ -33,6 +34,16 @@ TCP_PORT = 16509 # not tried again for RETRY_AFTER seconds. CONNECT_TIMEOUT = getattr(settings, "LIBVIRT_CONNECT_TIMEOUT", 5) RETRY_AFTER = getattr(settings, "LIBVIRT_RETRY_AFTER", 30) +# A call to a host that is up but stuck (a hung NFS mount, for one) waits as +# long as the host does: libvirt has no call timeout, and keepalive does not +# end it, since the daemon still answers keepalives. So in a process at most +# HOST_REQUESTS web requests use one host (by its name, whatever the login) +# at a time; another waits up to +# CONNECT_TIMEOUT for one of them to end, and is then refused. A request +# counts on the host it connected to last, until it ends. Threads outside a +# request (the job worker's, the consoles') are not counted: a job holds its +# host for as long as it runs. +HOST_REQUESTS = getattr(settings, "LIBVIRT_HOST_REQUESTS", 2) CLOSE_REASONS = {0: "error", 1: "end of file", 2: "keepalive timeout", 3: "client closed"} # ssh with a connect timeout for qemu+ssh (libvirt runs the "command" binary # in place of ssh and never passes one itself) @@ -92,6 +103,41 @@ def _after_fork_in_child(): os.register_at_fork(after_in_child=_after_fork_in_child) +_request = threading.local() # in a web request: the connection it counts on (HOST_REQUESTS) + + +def _request_started(**kwargs): + _leave_host() # a request that ended without request_finished (an error outside Django's handling) + _request.active = True + + +def _request_finished(**kwargs): + _leave_host() + _request.active = False + + +def _leave_host(): + held = getattr(_request, "held", None) + if held is not None: + _request.held = None + held.requests.release() + + +def _count_on(connection): + """A web request uses the host of connection from now on (HOST_REQUESTS).""" + if not getattr(_request, "active", False) or getattr(_request, "held", None) is connection: + return + _leave_host() + if not connection.requests.acquire(timeout=CONNECT_TIMEOUT): + raise util.ConnectionFailed( + f"Connection Failed: {connection.host} does not answer: {HOST_REQUESTS} requests still wait for it") + _request.held = connection + + +request_started.connect(_request_started, dispatch_uid="vrtManager.connection.request_started") +request_finished.connect(_request_finished, dispatch_uid="vrtManager.connection.request_finished") + + # libvirt's per-host directory of the VMs' UEFI variables (NVRAM) NVRAM_DIR = "/var/lib/libvirt/qemu/nvram" @@ -112,6 +158,7 @@ class wvmConnection(object): self.last_error = None self._opening = None # threading.Event of the open in progress self._retry_at = 0.0 # monotonic time before which no new open is tried + self.requests = None # the web requests using the host (HOST_REQUESTS), set by the manager # credentials self.host = host @@ -293,6 +340,7 @@ class wvmConnectionManager(object): self._connections = dict() self._connections_lock = ReadWriteLock() self._inherited = [] + self._host_requests = {} # host: the web requests using it, whatever the login (HOST_REQUESTS) def forget_connections(self): """For a forked child: stop using the inherited connections (the @@ -301,6 +349,7 @@ class wvmConnectionManager(object): self._inherited.append(self._connections) self._connections = dict() self._connections_lock = ReadWriteLock() + self._host_requests = {} def _search_connection(self, host, login, passwd, conn): """ @@ -343,10 +392,13 @@ class wvmConnectionManager(object): # Only register it here: connecting under the global lock # would make every other host wait for a slow one. connection = wvmConnection(host, login, passwd, conn) + connection.requests = self._host_requests.setdefault( + host, threading.BoundedSemaphore(HOST_REQUESTS)) self._connections.setdefault(host, []).append(connection) finally: self._connections_lock.release() + _count_on(connection) live = connection.alive_connection() if live is None: # (re-)connect within CONNECT_TIMEOUT; one open per connection at a time diff --git a/vrtManager/tests/test_connection.py b/vrtManager/tests/test_connection.py index ad3115a1..54940e49 100644 --- a/vrtManager/tests/test_connection.py +++ b/vrtManager/tests/test_connection.py @@ -8,6 +8,7 @@ import unittest from unittest.mock import MagicMock, patch from django.conf import settings +from django.core.signals import request_finished, request_started if not settings.configured: settings.configure(MAC_OUI="52:54:10") @@ -214,6 +215,91 @@ class BoundedOpenTestCase(unittest.TestCase): self.assertIn("ConnectTimeout", script.read()) +class HostRequestsTestCase(unittest.TestCase): + """At most HOST_REQUESTS web requests of a process use one host at a + time; another waits CONNECT_TIMEOUT and is refused.""" + + def setUp(self): + for name, value in (("wvmEventLoop", MagicMock()), ("CONNECT_TIMEOUT", 0.2)): + patcher = patch(f"vrtManager.connection.{name}", value) + patcher.start() + self.addCleanup(patcher.stop) + patcher = patch("vrtManager.connection.libvirt.openAuth", side_effect=lambda *args: alive_connection()) + patcher.start() + self.addCleanup(patcher.stop) + self.manager = wvmConnectionManager() + self.done = threading.Event() + self.threads = [] + self.addCleanup(self.end_requests) + + def end_requests(self): + self.done.set() + for thread in self.threads: + thread.join(5) + self.threads, self.done = [], threading.Event() + + def request(self, *hosts, login="u"): + """A web request on its own thread: it connects to hosts in turn and + ends when self.done is set. Returns its errors, filled once connected.""" + errors, connected, done = [], threading.Event(), self.done + + def run(): + request_started.send(sender=None) + try: + for host in hosts: + self.manager.get_connection(host, login, "p", CONN_TCP) + except libvirt.libvirtError as err: + errors.append(str(err)) + connected.set() + done.wait(5) + request_finished.send(sender=None) + + thread = threading.Thread(target=run) + thread.start() + self.threads.append(thread) + connected.wait(5) + return errors + + def test_a_third_request_to_a_host_is_refused_and_other_hosts_answer(self): + self.assertEqual(self.request("h") + self.request("h"), []) + started = time.monotonic() + self.assertEqual(self.request("h"), ["Connection Failed: h does not answer: 2 requests still wait for it"]) + self.assertGreaterEqual(time.monotonic() - started, 0.2) + self.assertEqual(self.request("g"), []) + + def test_a_host_counts_its_requests_whatever_the_login(self): + # two records of one host (another login, or credentials changed meanwhile) + self.assertEqual(self.request("h") + self.request("h", login="other"), []) + self.assertEqual(len(self.request("h", login="third")), 1) + + def test_a_request_without_request_finished_leaves_its_host_at_the_next_one(self): + self.assertEqual(self.request("h"), []) + request_started.send(sender=None) # this thread: a request that never got request_finished + try: + self.manager.get_connection("h", "u", "p", CONN_TCP) + request_started.send(sender=None) # its next request on the thread + self.assertEqual(self.request("h"), []) + finally: + request_finished.send(sender=None) + + def test_a_request_counts_once_on_a_host(self): + self.assertEqual(self.request("h", "h", "h") + self.request("h"), []) + self.assertEqual(len(self.request("h")), 1) + + def test_a_request_counts_on_the_host_it_connected_to_last(self): + self.assertEqual(self.request("h", "g") + self.request("h") + self.request("h"), []) + self.assertEqual(len(self.request("g", "h")), 1) + + def test_an_ended_request_lets_the_next_one_in(self): + self.assertEqual(self.request("h") + self.request("h"), []) + self.end_requests() + self.assertEqual(self.request("h") + self.request("h"), []) + + def test_threads_outside_a_request_are_not_counted(self): + self.assertEqual(self.request("h") + self.request("h"), []) + self.manager.get_connection("h", "u", "p", CONN_TCP) + + 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