diff --git a/gns3server/utils/notification_queue.py b/gns3server/utils/notification_queue.py index e198ffc9a..b7cf93759 100644 --- a/gns3server/utils/notification_queue.py +++ b/gns3server/utils/notification_queue.py @@ -17,6 +17,7 @@ import asyncio import json +import time import psutil from gns3server.utils.cpu_percent import CpuPercent @@ -35,22 +36,39 @@ class NotificationQueue(asyncio.Queue): def __init__(self): super().__init__() self._first = True + self._last_ping = None async def get(self, timeout): """ - When timeout is expire we send a ping notification with server information + Return a notification, or a ping notification with server information + at least every `timeout` seconds. The ping used to be generated only + when the queue was idle for the full timeout, which starved it under + sustained event load (e.g. high marker.match rates): clients stopped + receiving compute statistics until the event flow paused. """ # At first get we return a ping so the client immediately receives data if self._first: self._first = False - return ("ping", self._getPing(), {}) + return self._ping() - try: - (action, msg, kwargs) = await asyncio.wait_for(super().get(), timeout) - except asyncio.TimeoutError: - return ("ping", self._getPing(), {}) - return (action, msg, kwargs) + while True: + now = time.monotonic() + if self._last_ping is None or now - self._last_ping >= timeout: + return self._ping() + try: + (action, msg, kwargs) = await asyncio.wait_for(super().get(), timeout - (now - self._last_ping)) + return (action, msg, kwargs) + except asyncio.TimeoutError: + continue # the ping deadline has been reached + + def _ping(self): + """ + Build a ping notification and stamp the ping deadline. + """ + + self._last_ping = time.monotonic() + return ("ping", self._getPing(), {}) def _getPing(self): """ diff --git a/tests/utils/test_notification_queue.py b/tests/utils/test_notification_queue.py new file mode 100644 index 000000000..769a93e98 --- /dev/null +++ b/tests/utils/test_notification_queue.py @@ -0,0 +1,89 @@ +# +# Copyright (C) 2026 GNS3 Technologies Inc. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import time +import asyncio + +import pytest + +from gns3server.utils.notification_queue import NotificationQueue + + +async def _feed(queue, until, interval=0.02): + """ + Continuously put dummy events on the queue to simulate sustained load + (e.g. high marker.match rates). + """ + + seq = 0 + while time.monotonic() < until: + queue.put_nowait(("dummy", {"seq": seq}, {})) + seq += 1 + await asyncio.sleep(interval) + + +@pytest.mark.asyncio +async def test_first_get_returns_ping(): + + queue = NotificationQueue() + action, event, _ = await asyncio.wait_for(queue.get(1), 1) + assert action == "ping" + assert "cpu_usage_percent" in event + + +@pytest.mark.asyncio +async def test_idle_queue_pings_after_timeout(): + + queue = NotificationQueue() + await queue.get(0.3) # consume the first immediate ping + + start = time.monotonic() + action, _, _ = await asyncio.wait_for(queue.get(0.3), 1) + assert action == "ping" + assert time.monotonic() - start >= 0.25 # had to wait for the idle timeout + + +@pytest.mark.asyncio +async def test_ping_not_starved_under_sustained_load(): + """ + Regression test: a continuously-fed queue must still emit a ping at least + every `timeout` seconds. The old idle-timeout-only ping never fired under + sustained event load, so clients stopped receiving compute statistics + (no more compute.updated events) until the event flow paused. + """ + + queue = NotificationQueue() + action, _, _ = await queue.get(0.5) # consume the first immediate ping + assert action == "ping" + + until = time.monotonic() + 2.0 + producer = asyncio.create_task(_feed(queue, until)) + try: + pings = 0 + events = 0 + deadline = time.monotonic() + 2.5 + while time.monotonic() < deadline: + action, _, _ = await asyncio.wait_for(queue.get(0.5), 1) + if action == "ping": + pings += 1 + else: + events += 1 + # real events still flow... + assert events > 0 + # ...and pings interleave roughly every 0.5s instead of starving + assert pings >= 2 + finally: + producer.cancel()