From c503dae183f9e3accfb0a6d12c5414a379a228ac Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 18 Aug 2026 11:53:37 +0100 Subject: [PATCH 1/3] Simplify save restore logic --- src/daq_queuing_service/task_queue/queue.py | 15 ++++----------- tests/unit_tests/test_queue.py | 15 ++++++++++++--- 2 files changed, 16 insertions(+), 14 deletions(-) diff --git a/src/daq_queuing_service/task_queue/queue.py b/src/daq_queuing_service/task_queue/queue.py index 20d6b74..01c9f6e 100644 --- a/src/daq_queuing_service/task_queue/queue.py +++ b/src/daq_queuing_service/task_queue/queue.py @@ -70,20 +70,13 @@ class QueueContents(TypedDict): class Modifying(asyncio.Condition): def __init__( self, - on_enter: Callable[[], None], on_exit: Callable[[], None], on_error: Callable[[], None], ): super().__init__() - self._on_enter = on_enter self._on_exit = on_exit self._on_error = on_error - async def __aenter__(self): - result = await super().__aenter__() - self._on_enter() - return result - async def __aexit__( self, exc_type: type[BaseException] | None, @@ -130,7 +123,6 @@ def __init__(self, converter: Converter, broadcaster: Broadcaster[QUEUE_EVENTS]) self._converter = converter self._broadcaster = broadcaster self._modifying = Modifying( - on_enter=self._save_contents, on_exit=self._sync, on_error=self._restore_latest_good_contents, ) @@ -217,6 +209,7 @@ def _sync(self): if not self._call_queue: self._pause_queue(PauseReason.EMPTY_QUEUE) + self._save_contents() self._broadcast_changes() self._modifying.notify_all() @@ -231,17 +224,17 @@ def _copy_contents(self) -> QueueContents: } ) - def _save_contents(self) -> None: + def _save_contents(self): self._last_good_contents = self._copy_contents() - def _restore_from_contents(self, contents: QueueContents) -> None: + def _restore_from_contents(self, contents: QueueContents): self._tasks = TaskRegistry(contents["tasks"]) self._queue = contents["queue"] self._history = contents["history"] self._call_queue = contents["call_queue"] self._call_history = contents["call_history"] - def _restore_latest_good_contents(self) -> None: + def _restore_latest_good_contents(self): self._restore_from_contents(self._last_good_contents) def _broadcast_changes(self): diff --git a/tests/unit_tests/test_queue.py b/tests/unit_tests/test_queue.py index c78da08..5358792 100644 --- a/tests/unit_tests/test_queue.py +++ b/tests/unit_tests/test_queue.py @@ -1099,14 +1099,23 @@ def test__restore_from_contents_replaces_queue_contents(task_queue: TaskQueue): assert task_queue._call_history == new_call_history -async def test__last_good_contents_updated_when_modifying_lock_entered( +async def test__last_good_contents_updated_when_modifying_lock_exited( task_queue: TaskQueue, ): - task_queue._queue = ["should be copied"] + task_queue._queue = [] async with task_queue._modifying: - task_queue._queue = [] + task_queue._add_tasks( + [ + Task( + id="should be copied", + experiment=TaskRequest(name="", instrument_session=""), + ) + ], + position=0, + ) + task_queue._queue = [] assert task_queue._last_good_contents["queue"] == ["should be copied"] assert task_queue._queue == [] From 69332ec1ed82107700980c98740f996ce78df1e5 Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 18 Aug 2026 12:06:13 +0100 Subject: [PATCH 2/3] Don't sync after read-only operations --- src/daq_queuing_service/task_queue/queue.py | 19 +++++++++++-------- tests/unit_tests/test_queue.py | 4 ++-- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/src/daq_queuing_service/task_queue/queue.py b/src/daq_queuing_service/task_queue/queue.py index 01c9f6e..b9b26e5 100644 --- a/src/daq_queuing_service/task_queue/queue.py +++ b/src/daq_queuing_service/task_queue/queue.py @@ -70,10 +70,11 @@ class QueueContents(TypedDict): class Modifying(asyncio.Condition): def __init__( self, + lock: asyncio.Lock, on_exit: Callable[[], None], on_error: Callable[[], None], ): - super().__init__() + super().__init__(lock=lock) self._on_exit = on_exit self._on_error = on_error @@ -122,7 +123,9 @@ def __init__(self, converter: Converter, broadcaster: Broadcaster[QUEUE_EVENTS]) ) self._converter = converter self._broadcaster = broadcaster + self._lock = asyncio.Lock() self._modifying = Modifying( + lock=self._lock, on_exit=self._sync, on_error=self._restore_latest_good_contents, ) @@ -343,7 +346,7 @@ async def get_task_by_id(self, task_id: str) -> TaskWithPosition: TaskNotFoundError: Raised if the no task exists with the requested task ID. """ # Returns copy so don't have to be worried about caller modifying task. - async with self._modifying: + async with self._lock: return self._get_task_by_id(task_id) def _get_task_by_id(self, task_id: str) -> TaskWithPosition: @@ -362,7 +365,7 @@ async def get_task_by_position(self, position: int) -> TaskWithPosition | None: if no task exists at the requested position. """ # Returns copy so don't have to be worried about caller modifying task. - async with self._modifying: + async with self._lock: if position < -self.length or position >= self.length: return None return self._get_task_by_id(self._queue[position]) @@ -375,7 +378,7 @@ async def get_queue(self) -> list[TaskWithPosition]: will be run in. """ # Returns copies so don't have to be worried about caller modifying tasks. - async with self._modifying: + async with self._lock: return self._get_queue() async def get_history(self) -> list[TaskWithPosition]: @@ -386,7 +389,7 @@ async def get_history(self) -> list[TaskWithPosition]: chronological order. """ # Returns copies so don't have to be worried about caller modifying tasks. - async with self._modifying: + async with self._lock: return self._get_history() async def get_tasks(self) -> list[TaskWithPosition]: @@ -397,7 +400,7 @@ async def get_tasks(self) -> list[TaskWithPosition]: with the history. """ # Returns copies so don't have to be worried about caller modifying tasks. - async with self._modifying: + async with self._lock: return self._get_history() + self._get_queue() async def add_tasks(self, tasks: list[Task], position: int | None = None) -> None: @@ -633,14 +636,14 @@ def _get_history(self) -> list[TaskWithPosition]: ] async def get_call_queue(self) -> list[BlueapiCallResponse]: - async with self._modifying: + async with self._lock: return self._get_call_queue() def _get_call_queue(self) -> list[BlueapiCallResponse]: return [call.to_response() for call in self._call_queue] async def get_call_history(self) -> list[BlueapiCallResponse]: - async with self._modifying: + async with self._lock: return self._get_call_history() def _get_call_history(self) -> list[BlueapiCallResponse]: diff --git a/tests/unit_tests/test_queue.py b/tests/unit_tests/test_queue.py index 5358792..cd0e046 100644 --- a/tests/unit_tests/test_queue.py +++ b/tests/unit_tests/test_queue.py @@ -1137,7 +1137,7 @@ def convert( task_queue._converter.construct_blueapi_calls = convert with pytest.raises(ConverterError): - await task_queue.get_queue() + await task_queue.move_task("0", 0) assert task_queue._queue == ["0", "1", "2", "3", "4"] assert list(task_queue._tasks.keys()) == ["0", "1", "2", "3", "4"] @@ -1158,6 +1158,6 @@ def convert( task_queue.__init__(task_queue._converter, task_queue._broadcaster) with pytest.raises(ConverterError): - await task_queue.get_queue() + await task_queue.add_tasks(MagicMock()) task_queue._restore_latest_good_contents.assert_called_once() From db6dfc59a7857ab58caba2cf6a3973f95374185a Mon Sep 17 00:00:00 2001 From: Jacob Williamson Date: Tue, 18 Aug 2026 13:15:33 +0100 Subject: [PATCH 3/3] Add test --- tests/unit_tests/test_queue.py | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/tests/unit_tests/test_queue.py b/tests/unit_tests/test_queue.py index cd0e046..91af02e 100644 --- a/tests/unit_tests/test_queue.py +++ b/tests/unit_tests/test_queue.py @@ -1,5 +1,6 @@ import asyncio import copy +from typing import Any from unittest.mock import MagicMock import pytest @@ -1161,3 +1162,30 @@ def convert( await task_queue.add_tasks(MagicMock()) task_queue._restore_latest_good_contents.assert_called_once() + + +@pytest.mark.parametrize( + "method_name, args", + [ + ("get_queue", []), + ("get_tasks", []), + ("get_history", []), + ("get_task_by_id", ["0"]), + ("get_task_by_position", [0]), + ("get_call_queue", []), + ("get_call_history", []), + ], +) +async def test__sync_not_called_for_read_only_methods( + task_queue: TaskQueue, method_name: str, args: list[Any] +): + task_queue._sync = MagicMock() + + # Need to reinitialise so that mocked _sync is injected into Modifying object + contents = copy.copy(task_queue._last_good_contents) + task_queue.__init__(task_queue._converter, task_queue._broadcaster) + task_queue._restore_from_contents(contents) + + await getattr(task_queue, method_name)(*args) + + task_queue._sync.assert_not_called()