Merge branch 'thread_safe_scan_directory_manager' into 'v3'

Make scan directory manager thread safe

Closes #506

See merge request openflexure/openflexure-microscope-server!346

src/openflexure_microscope_server/scan_directories.py

@@ -5,9 +5,12 @@ import os
 import re
 import shutil
 import zipfile
+import threading
 
 from pydantic import BaseModel
 
+from openflexure_microscope_server.utilities import requires_lock
+
 IMG_DIR_NAME = "images"
 SCAN_ZERO_PAD_DIGITS = 4
 
@@ -43,6 +46,7 @@ class ScanDirectoryManager:
         self._base_scan_dir = base_scan_dir
         if not os.path.exists(self._base_scan_dir):
             os.makedirs(self._base_scan_dir)
+        self._lock = threading.RLock()
 
     @property
     def base_dir(self) -> str:
@@ -67,6 +71,7 @@ class ScanDirectoryManager:
         """
         return os.path.join(self._base_scan_dir, scan_name, IMG_DIR_NAME)
 
+    @requires_lock
     def get_file_path_from(
         self, scan_name: str, filename: str, check_exists: bool = False
     ) -> Optional[str]:
@@ -81,6 +86,7 @@ class ScanDirectoryManager:
                 return None
         return file_path
 
+    @requires_lock
     def get_file_path_from_img_dir(
         self, scan_name: str, filename: str, check_exists: bool = False
     ) -> Optional[str]:
@@ -106,10 +112,12 @@ class ScanDirectoryManager:
         return self.get_file_path_from_img_dir(scan_name, stitch_fname)
 
     @property
+    @requires_lock
     def all_scans(self) -> list[str]:
         """Return a list of the scan names in the base directory."""
         return [f.name for f in os.scandir(self._base_scan_dir) if f.is_dir()]
 
+    @requires_lock
     def all_scans_info(self) -> list[ScanInfo]:
         """Return a lists of ScanInfo objects for each scan."""
         all_info: list[ScanInfo] = []
@@ -117,6 +125,7 @@ class ScanDirectoryManager:
             all_info.append(ScanDirectory(scan_name, self.base_dir).scan_info())
         return all_info
 
+    @requires_lock
     def _unique_scan_name(self, scan_name: str) -> str:
         """Get the next unique scan name starting with the given name.
 
@@ -146,6 +155,7 @@ class ScanDirectoryManager:
 
         return f"{scan_name}_{scan_num:0{SCAN_ZERO_PAD_DIGITS}d}"
 
+    @requires_lock
     def new_scan_dir(self, scan_name: str) -> "ScanDirectory":
         """Get a unique name for this scan and create a directory for it.
 
@@ -168,10 +178,12 @@ class ScanDirectoryManager:
         os.makedirs(self.img_dir_for(full_scan_name))
         return ScanDirectory(full_scan_name, self.base_dir)
 
+    @requires_lock
     def delete_scan(self, scan_name: str) -> None:
         """Delete a scan."""
         shutil.rmtree(self.path_for(scan_name))
 
+    @requires_lock
     def zip_scan(self, scan_name: str, final_version: bool = False) -> "ScanDirectory":
         """Zips any images from the scan not yet zipped, return full path to zip.
 
@@ -181,6 +193,7 @@ class ScanDirectoryManager:
         """
         return ScanDirectory(scan_name, self.base_dir).zip_files(final_version)
 
+    @requires_lock
     def check_free_disk_space(self, min_space: int = 500000000) -> None:
         """Raise an exception if there is not enough free disk space to continue scanning.
 

src/openflexure_microscope_server/things/smart_scan.py

@@ -717,6 +717,9 @@ class SmartScanThing(lt.Thing):
         This is a wrapper around scan manager's delete_scan that logs to the
         invocation logger id there is a problem.
         """
+        if self._ongoing_scan is not None and scan_name == self._ongoing_scan.name:
+            logger.error("Attempted to delete ongoing scan.")
+            return False
         try:
             self._scan_dir_manager.delete_scan(scan_name)
             return True

src/openflexure_microscope_server/utilities.py

@@ -1,14 +1,19 @@
 """Utility functions and classes."""
 
+from typing import TypeVar, Callable, ParamSpec
 import os
 import re
 from threading import Thread
 import logging
 from importlib.metadata import version
 import tomllib
+from functools import wraps
 
 from pydantic import BaseModel
 
+T = TypeVar("T")
+P = ParamSpec("P")
+
 LOGGER = logging.getLogger(__name__)
 
 REPO_DIR = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
@@ -18,6 +23,36 @@ COMMIT_REGEX = re.compile(r"[0-9a-f]{40}")
 REF_REGEX = re.compile(r"^ref:\s(.*)$")
 
 
+def _is_lock_like(obj):
+    """Check if an object is a lock.
+
+    Cannot use ``isinstance(obj, threading.RLock)``, may be possible to use
+    ``isinstance(obj, threading._RLock)``. But this uses a private method and may
+    break. Instead making the check that both "acquire" and "release" exist.
+    """
+    return hasattr(obj, "acquire") and hasattr(obj, "release")
+
+
+def requires_lock(method: Callable[P, T]) -> Callable[P, T]:
+    """Decorate a class method so that it requires the class lock to run.
+
+    The class should have a reentrant lock with the name ``self._lock``.
+    """
+
+    @wraps(method)
+    def wrapper(self, *args: P.args, **kwargs: P.kwargs) -> T:
+        # Confirm the object the method is attached to has a self._lock property
+        if not hasattr(self, "_lock"):
+            raise AttributeError(f"{self.__class__.__name__} has no '_lock' attribute")
+        # And that it is a reentrant lock.
+        if not _is_lock_like(self._lock):
+            raise TypeError("requires_lock requires self._lock to be a lock")
+        with self._lock:
+            return method(self, *args, **kwargs)
+
+    return wrapper
+
+
 class ErrorCapturingThread(Thread):
     """Subclass of Thread that captures exceptions from the target function.
 

tests/test_lock_decorator.py

@@ -0,0 +1,134 @@
+"""Tests for the ``requires_lock`` decorator in ``utilities.py``."""
+
+import threading
+from time import sleep, time
+
+import pytest
+
+from openflexure_microscope_server.utilities import requires_lock
+
+
+class LockableClass:
+    """A class with methods that can be locked with the ``requires_lock`` decorator."""
+
+    def __init__(self):
+        """Initialise the LockableClass with a rentrant lock."""
+        self._lock = threading.RLock()
+
+    @property
+    @requires_lock
+    def test_property(self) -> int:
+        """A property that always returns 1."""
+        return 1
+
+    @requires_lock
+    def fast_task(self) -> int:
+        """Instantly return the number 2."""
+        return 2
+
+    @requires_lock
+    def slow_task(self) -> int:
+        """Take 0.5s to return the number 3."""
+        sleep(0.5)
+        return 3
+
+    @requires_lock
+    def task_that_calls_fast_task(self) -> int:
+        """Require a lock and call a fast task that requires it too."""
+        return self.fast_task()
+
+    @requires_lock
+    def task_that_calls_property(self) -> int:
+        """Require a lock and call a property that requires it too."""
+        return self.test_property
+
+
+@pytest.fixture
+def lockable_obj() -> LockableClass:
+    """Return an instance of LockableClass."""
+    return LockableClass()
+
+
+def test_error_if_no_lock(lockable_obj):
+    """Test there is an AttributeError if the object has no _lock attribute."""
+    # Delete the lock!
+    del lockable_obj._lock
+    with pytest.raises(AttributeError) as exec_info:
+        lockable_obj.test_property
+    assert str(exec_info.value) == "LockableClass has no '_lock' attribute"
+
+
+def test_error_if_lock_of_wrong_type(lockable_obj):
+    """Test there is an TypeError if the object's _lock attribute isn't a lock."""
+    # Delete the lock!
+    lockable_obj._lock = "I'm a mock, not a lock!"
+    with pytest.raises(TypeError) as exec_info:
+        lockable_obj.test_property
+    expected_msg = "requires_lock requires self._lock to be a lock"
+    assert str(exec_info.value) == expected_msg
+
+
+def test_functionality(lockable_obj):
+    """Check that this doesn't affect the most basic functionality of the class."""
+    assert lockable_obj.test_property == 1
+    assert lockable_obj.fast_task() == 2
+    # Don't check slow task here as it is slow and functionally no different from fast
+    # task.
+    # Check tasks that call other tasks locked to check that they are rentrant
+    assert lockable_obj.task_that_calls_property() == 1
+    assert lockable_obj.task_that_calls_fast_task() == 2
+
+
+def check_fast_target_waits(lockable_obj, fast_target):
+    """Start threads with a slow target and the supplied fast target, check lock works.
+
+    :param lockable_obj: An instance of LockableClass
+    :param fast_target: The callable to test with the fast thread.
+
+    The program runs by:
+
+    * First checking that fast_target when run on its own completes in less than
+        0.1s
+    * Starts the thread with the slow_target.
+    * Starts the thread with the fast_target straight afterwards.
+    * Wait for the fast target to complete.
+    * Check the slow target has completeted once returned (showing that the fast
+        target had to wait because of the lock)
+    * Check that the time elapsed is over 0.4s.
+    """
+    # On its own the fast thread is fast.
+    fast_thread = threading.Thread(target=fast_target)
+    t_start = time()
+    fast_thread.start()
+    fast_thread.join()
+    assert time() - t_start < 0.1
+
+    # If slow thread is running it has to wait.
+    fast_thread = threading.Thread(target=fast_target)
+    slow_thread = threading.Thread(target=lockable_obj.slow_task)
+
+    t_start = time()
+    slow_thread.start()
+    fast_thread.start()
+    fast_thread.join()
+    # slow task must have finished
+    assert not slow_thread.is_alive()
+    assert time() - t_start > 0.4
+
+
+def test_locking(lockable_obj):
+    """Test the locking works when calling decorated methods in two threads."""
+    check_fast_target_waits(lockable_obj, lockable_obj.fast_task)
+
+
+def test_property_locking(lockable_obj):
+    """Test that the locking works when calling a property not a function.
+
+    The target is a method with no lock, that calls the property that requires
+    the lock.
+    """
+
+    def property_caller() -> int:
+        return lockable_obj.test_property
+
+    check_fast_target_waits(lockable_obj, property_caller)