diff --git a/gns3server/api/routes/controller/projects.py b/gns3server/api/routes/controller/projects.py index f9c838a34..d10de587a 100644 --- a/gns3server/api/routes/controller/projects.py +++ b/gns3server/api/routes/controller/projects.py @@ -485,6 +485,38 @@ async def project_ws_notifications( await project.close() +@router.websocket("/{project_id}/notifications/markers/ws") +async def project_marker_ws_notifications( + project_id: UUID, + websocket: WebSocket, + current_user: schemas.User = Depends(has_privilege_on_websocket("Project.Audit")) +) -> None: + """ + Receive marker notifications (e.g. marker.match) for a project on a + dedicated WebSocket, separate from the main project stream so high-frequency + marker.matches do not block topology events (node.*/link.*). + + Required privilege: Project.Audit + """ + + if current_user is None: + return + + controller = Controller.instance() + project = controller.get_project(str(project_id)) + + log.info(f"New client has connected to the marker notification stream for project ID '{project.id}' (WebSocket method)") + try: + with controller.notification.project_marker_queue(project.id) as queue: + while True: + notification = await queue.get_json(5) + await websocket.send_text(notification) + except (ConnectionClosed, WebSocketDisconnect): + log.info(f"Client has disconnected from the marker notification stream for project ID '{project.id}' (WebSocket method)") + except WebSocketException as e: + log.warning(f"Error while sending marker event to WebSocket client: {e}") + + @router.get("/{project_id}/export", dependencies=[Depends(has_privilege("Project.Audit"))]) async def export_project( project: Project = Depends(dep_project), diff --git a/gns3server/controller/notification.py b/gns3server/controller/notification.py index 3a63d9b87..d3540a746 100644 --- a/gns3server/controller/notification.py +++ b/gns3server/controller/notification.py @@ -31,6 +31,7 @@ class Notification: self._controller = controller self._project_listeners = {} + self._project_marker_listeners = {} self._controller_listeners = set() @contextmanager @@ -49,6 +50,26 @@ class Notification: finally: self._project_listeners[project_id].remove(queue) + @contextmanager + def project_marker_queue(self, project_id): + """ + Get a queue of marker notifications (marker.match etc.) for a project. + + Marker events are delivered on this dedicated channel instead of the + main project queue, so high-frequency marker.matches do not cause + head-of-line blocking for topology events (node.*/link.*). + + Use it with Python with + """ + + queue = NotificationQueue() + self._project_marker_listeners.setdefault(project_id, set()) + self._project_marker_listeners[project_id].add(queue) + try: + yield queue + finally: + self._project_marker_listeners[project_id].remove(queue) + @contextmanager def controller_queue(self): """ @@ -104,6 +125,8 @@ class Notification: elif action == "ping": event["compute_id"] = compute_id self.project_emit(action, event) + elif action.startswith("marker."): + self.marker_emit(action, event, project_id) else: self.project_emit(action, event, project_id) @@ -120,6 +143,25 @@ class Notification: else: self._send_event_to_all_projects(action, event) + def marker_emit(self, action, event, project_id): + """ + Send a marker notification (e.g. marker.match) to clients listening on + the dedicated marker channel for this project. Marker events are kept + off the main project queue on purpose, to avoid head-of-line blocking + from high-frequency matches. + + :param action: Action name + :param event: Event to send + :param project_id: Project id the marker belongs to + """ + + try: + marker_listeners = self._project_marker_listeners[project_id] + except KeyError: + return + for listener in marker_listeners: + asyncio.get_running_loop().call_soon_threadsafe(listener.put_nowait, (action, event, {})) + def _send_event_to_project(self, project_id, action, event): """ Send an event to all the client listening for notifications for diff --git a/tests/controller/test_notification.py b/tests/controller/test_notification.py index 9204acea3..754559154 100644 --- a/tests/controller/test_notification.py +++ b/tests/controller/test_notification.py @@ -120,6 +120,33 @@ async def test_dispatch_node_updated(controller, node, project): assert event["properties"]["startup_config"] == "ip 192" +@pytest.mark.asyncio +async def test_dispatch_marker_routed_to_marker_channel(controller, project): + """ + marker.* events are dispatched to the dedicated marker channel, not the + main project queue, so high-frequency matches cannot block topology events. + """ + + notif = controller.notification + with notif.project_queue(project.id) as project_q, \ + notif.project_marker_queue(project.id) as marker_q: + assert len(notif._project_marker_listeners[project.id]) == 1 + await project_q.get(0.1) # consume initial ping + await marker_q.get(0.1) # consume initial ping + + await notif.dispatch("marker.match", {"link_id": "abc"}, + project_id=project.id, compute_id=1) + + # marker.match lands on the marker channel... + msg = await marker_q.get(5) + assert msg == ('marker.match', {"link_id": "abc"}, {}) + # ...and does NOT land on the main project queue (times out -> ping) + msg = await project_q.get(0.1) + assert msg[0] == "ping" + + assert len(notif._project_marker_listeners[project.id]) == 0 + + def test_various_notification(controller, node): notif = controller.notification