From 8e261bb2c8bd6b1e215b87348ec171c988e985f9 Mon Sep 17 00:00:00 2001 From: hpoeche Date: Sun, 28 Jun 2026 22:13:17 +0200 Subject: [PATCH 1/3] sdk: fix local_file backend multi-instance safety Previously the LocalFileIdentifiableStore allowed multiple instances (potentially across multiple processes) to acces the same directory. This can lead to undetected invalid cache entries. Additionally it would increase the overhead for adding thread-safety to the R/W operations as synchronization across multiple instances would be necessary. In order to overcome these limitations, these changes introduce a directory lock mechanism, that only allows one instance of the class to hold a lock on a `.lock` file in the directory. A contex manager is used to secure operations on the directory and prevent concurrent release of the lock. For now this only work on POSIX machines, as it relies on `flock` functionatily which is not provided by Windows. --- sdk/basyx/aas/backend/local_file.py | 219 +++++++++++++++++++++++----- sdk/test/backend/test_local_file.py | 139 +++++++++++++++++- 2 files changed, 319 insertions(+), 39 deletions(-) diff --git a/sdk/basyx/aas/backend/local_file.py b/sdk/basyx/aas/backend/local_file.py index df21ddfe..9c8ddadd 100644 --- a/sdk/basyx/aas/backend/local_file.py +++ b/sdk/basyx/aas/backend/local_file.py @@ -11,7 +11,8 @@ The :class:`~LocalFileIdentifiableStore` handles adding, deleting and otherwise managing the AAS objects in a specific Directory. """ -from typing import Iterator +import contextlib +from typing import Iterator, Generator, TextIO, Optional import logging import json import os @@ -20,6 +21,7 @@ import threading import warnings import weakref +from pathlib import Path from ..adapter.json import json_serialization, json_deserialization from basyx.aas import model @@ -28,23 +30,138 @@ logger = logging.getLogger(__name__) +class DirectoryLock: + """ + Implementation of a Lock on a directory. Uses POSIX ``flock`` on ``/.lock`` to control which instance + holds the lock. Even if the process which holds the lock dies, the lock is released. + + The methods :meth:`acquire` and :meth:`release` are thread-safe and can be called in any state. If :meth:`acquire` + is called when the directory is already locked no errors occur. The same holds for calling :meth:`release` in an + unlocked state. + + .. note:: + Directory locking relies on ``fcntl.flock``, which is only available on POSIX platforms. On Windows + a warning is logged and the lock is skipped. + """ + def __init__(self, directory_path: str): + self.directory_path = Path(directory_path) + self.lock_path = self.directory_path / ".lock" + self._dir_lock_file: Optional[TextIO] = None + + # We count the number of active accesses to the directory to ensure that + # release() is only releasing if all are done + self._locking_lock = threading.Condition() + self._is_locked_flag = False + self._active_accesses = 0 + self._is_releasing = False + + def _lazy_import_fcntl(self): + try: + import fcntl as _fcntl + except ImportError: + _fcntl = None # Windows: directory locking is unavailable + return _fcntl + + def acquire(self) -> None: + """ + Acquires the lock on the directory non-blocking. If the ``/.lock`` file is already locked + it raises a :class:`RuntimeError`. + + .. note:: + Directory locking relies on ``fcntl.flock``, which is only available on POSIX platforms. On Windows + a warning is logged and the lock is skipped. + + :raises RuntimeError: If the directory is already locked by another instance + """ + with self._locking_lock: # Ensure only one dir_lock is acquired at a time + if self._is_locked_flag: + return # Idempotent + + self._dir_lock_file = open(self.lock_path, "a") # use "a" to not swap the inode as in "w" + + _fcntl = self._lazy_import_fcntl() + if _fcntl is None: + # Windows does not support locking files. For now, we raise a warning + # and continue to assure backwards compatibility + + logger.warning("fcntl unavailable; directory locking is disabled (Windows?)") + self._is_locked_flag = True + return + + try: + _fcntl.flock(self._dir_lock_file.fileno(), _fcntl.LOCK_EX | _fcntl.LOCK_NB) + except OSError: + # dir_lock already taken by other process + self._dir_lock_file.close() + self._dir_lock_file = None + raise RuntimeError(f"Directory {self.directory_path} is already locked by another DirectoryLock") + else: + self._is_locked_flag = True + + def release(self) -> None: + """ + If the directory is currently locked, this method waits for all current accesses in :meth:`ensure_locked` + to finish and releases the lock afterwards. + """ + with self._locking_lock: + if not self._is_locked_flag: + return # Idempotent + self._is_releasing = True # Restrict new ensure_locked() entries + while self._active_accesses > 0: # Wait for all active accesses to finish + self._locking_lock.wait() + + # Release flock + self._is_locked_flag = False + if self._dir_lock_file is not None: + self._dir_lock_file.close() # lock is released by closing fd + self._dir_lock_file = None + + @contextlib.contextmanager + def ensure_locked(self) -> Generator: + """Context Manager for access to the locked directory + + Use this as a context manager when accessing files in the locked directory. This ensures that the directory + is locked during the whole execution of the ``with`` block. Fails with a :class:`RuntimeError` if the directory + is not locked when entering the ``with`` block. + + :raises RuntimeError: If the directory is not locked on enter + """ + with self._locking_lock: + if (not self._is_locked_flag) and (not self._is_releasing): + raise RuntimeError(f"Directory {self.directory_path} is not locked!") + self._active_accesses += 1 + try: + yield + finally: + with self._locking_lock: + self._active_accesses -= 1 + self._locking_lock.notify_all() + + def close(self) -> None: + self.release() + + class LocalFileIdentifiableStore(model.AbstractObjectStore[model.Identifier, model.Identifiable]): """ An ObjectStore implementation for :class:`~basyx.aas.model.base.Identifiable` BaSyx Python SDK objects backed - by a local file based local backend - - .. warning:: - This backend is intended for development and testing only. It provides no - concurrency control across processes: concurrent writes to the same object - (e.g. under a multi-worker WSGI server) will silently overwrite each other, - with the last writer winning and no error raised. Use a dedicated database - backend for any production deployment. + by a local file based local backend. + + At most one instance of this class can manage a directory. To ensure this a + :class:`~basyx.aas.backend.local_file.DirectoryLock` is used (with all it limitations on non-POSIX systems). + The lock is applied eager (as soon as the directory exists) either in the constructor or after calling + :meth:`check_directory`. Unless this is done all further method calls will fail with :class:`RuntimeError`. + + Call :meth:`close` to release the directory lock when done. + + .. note:: + Directory locking relies on ``fcntl.flock``, which is only available on POSIX platforms. On Windows + a warning is logged and the lock is skipped. """ def __init__(self, directory_path: str): """ Initializer of class LocalFileIdentifiableStore - :param directory_path: Path to the local file backend (the path where you want to store your AAS JSON files) + :param directory_path: Path to the local file backend (the path where you want to store your AAS JSON files). """ self.directory_path: str = directory_path.rstrip("/") @@ -57,6 +174,23 @@ def __init__(self, directory_path: str): = weakref.WeakValueDictionary() self._object_cache_lock = threading.Lock() + # We need to prevent multiple instances of LocalFileIdentifiableStore performing R/W operations on the same + # directory in order to ensure cache validity. The directory is locked as soon as it exists. + self._dir_lock = DirectoryLock(self.directory_path) + if os.path.exists(self.directory_path): + self._acquire_dir_lock() + + def _acquire_dir_lock(self) -> None: + try: + self._dir_lock.acquire() + except RuntimeError as ex: + raise RuntimeError(f"Directory {self.directory_path} is already in use" + f" by another LocalFileIdentifiableStore instance.") from ex + + def close(self) -> None: + """Release the directory lock. Call this when the store is no longer needed.""" + self._dir_lock.close() + def check_directory(self, create=False): """ Check if the directory exists and created it if not (and requested to do so) @@ -69,6 +203,7 @@ def check_directory(self, create=False): # Create directory os.mkdir(self.directory_path) logger.info("Creating directory {}".format(self.directory_path)) + self._acquire_dir_lock() def get_identifiable_by_hash(self, hash_: str) -> model.Identifiable: """ @@ -76,12 +211,13 @@ def get_identifiable_by_hash(self, hash_: str) -> model.Identifiable: :raises KeyError: If the respective file could not be found """ - try: - with open("{}/{}.json".format(self.directory_path, hash_), "r") as file: - data = json.load(file, cls=json_deserialization.AASFromJsonDecoder) - obj = data["data"] - except FileNotFoundError as e: - raise KeyError("No Identifiable with hash {} found in local file database".format(hash_)) from e + with self._dir_lock.ensure_locked(): + try: + with open("{}/{}.json".format(self.directory_path, hash_), "r") as file: + data = json.load(file, cls=json_deserialization.AASFromJsonDecoder) + obj = data["data"] + except FileNotFoundError as e: + raise KeyError("No Identifiable with hash {} found in local file database".format(hash_)) from e with self._object_cache_lock: if obj.id in self._object_cache: return self._object_cache[obj.id] @@ -94,13 +230,14 @@ def get_item(self, identifier: model.Identifier) -> model.Identifiable: :raises KeyError: If the respective file could not be found """ - with self._object_cache_lock: - if identifier in self._object_cache: - return self._object_cache[identifier] - try: - return self.get_identifiable_by_hash(self._transform_id(identifier)) - except KeyError as e: - raise KeyError("No Identifiable with id {} found in local file database".format(identifier)) from e + with self._dir_lock.ensure_locked(): + with self._object_cache_lock: + if identifier in self._object_cache: + return self._object_cache[identifier] + try: + return self.get_identifiable_by_hash(self._transform_id(identifier)) + except KeyError as e: + raise KeyError("No Identifiable with id {} found in local file database".format(identifier)) from e def _write_atomic(self, x: model.Identifiable) -> None: """ @@ -128,9 +265,10 @@ def add(self, x: model.Identifiable) -> None: :raises KeyError: If an object with the same id exists already in the object store """ logger.debug("Adding object %s to Local File Store ...", repr(x)) - if os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))): - raise KeyError("Identifiable with id {} already exists in local file database".format(x.id)) - self._write_atomic(x) + with self._dir_lock.ensure_locked(): + if os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))): + raise KeyError("Identifiable with id {} already exists in local file database".format(x.id)) + self._write_atomic(x) with self._object_cache_lock: self._object_cache[x.id] = x @@ -141,9 +279,10 @@ def commit(self, x: model.Identifiable) -> None: :param x: The object to persist :raises KeyError: If the object is not present in the store """ - if not os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))): - raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) - self._write_atomic(x) + with self._dir_lock.ensure_locked(): + if not os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))): + raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) + self._write_atomic(x) def discard(self, x: model.Identifiable) -> None: """ @@ -153,10 +292,11 @@ def discard(self, x: model.Identifiable) -> None: :raises KeyError: If the object does not exist in the database """ logger.debug("Deleting object %s from Local File Store database ...", repr(x)) - try: - os.remove("{}/{}.json".format(self.directory_path, self._transform_id(x.id))) - except FileNotFoundError as e: - raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) from e + with self._dir_lock.ensure_locked(): + try: + os.remove("{}/{}.json".format(self.directory_path, self._transform_id(x.id))) + except FileNotFoundError as e: + raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) from e with self._object_cache_lock: self._object_cache.pop(x.id, None) @@ -176,7 +316,8 @@ def __contains__(self, x: object) -> bool: else: return False logger.debug("Checking existence of object with id %s in database ...", repr(x)) - return os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(identifier))) + with self._dir_lock.ensure_locked(): + return os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(identifier))) def __len__(self) -> int: """ @@ -185,7 +326,8 @@ def __len__(self) -> int: :return: The number of objects (determined from the number of documents) """ logger.debug("Fetching number of documents from database ...") - return sum(1 for f in os.listdir(self.directory_path) if f.lower().endswith(".json")) + with self._dir_lock.ensure_locked(): + return sum(1 for f in os.listdir(self.directory_path) if f.lower().endswith(".json")) def __iter__(self) -> Iterator[model.Identifiable]: """ @@ -195,9 +337,10 @@ def __iter__(self) -> Iterator[model.Identifiable]: the identifiable objects on the fly. """ logger.debug("Iterating over objects in database ...") - for name in os.listdir(self.directory_path): - if name.lower().endswith(".json"): - yield self.get_identifiable_by_hash(name[:-5]) + with self._dir_lock.ensure_locked(): + for name in os.listdir(self.directory_path): + if name.lower().endswith(".json"): + yield self.get_identifiable_by_hash(name[:-5]) @staticmethod def _transform_id(identifier: model.Identifier) -> str: diff --git a/sdk/test/backend/test_local_file.py b/sdk/test/backend/test_local_file.py index 71447f61..008eab15 100644 --- a/sdk/test/backend/test_local_file.py +++ b/sdk/test/backend/test_local_file.py @@ -7,6 +7,10 @@ import gc import os.path import shutil +import tempfile +import threading +import concurrent.futures +from typing import Callable from unittest import TestCase @@ -18,6 +22,106 @@ source_core: str = "file://localhost/{}/".format(store_path) +def run_threads(fns: list[Callable]): + fn_futures: list[concurrent.futures.Future] = [] + with concurrent.futures.ThreadPoolExecutor() as executor: + for fn in fns: + fn_futures.append(executor.submit(fn)) + + concurrent.futures.wait(fn_futures) + for future in fn_futures: + ex = future.exception() + if ex is not None: + raise ex + + +class DirectoryLockTest(TestCase): + + def test_double_locking(self): + with tempfile.TemporaryDirectory() as tmpdir: + lock = local_file.DirectoryLock(tmpdir) + lock.acquire() + lock.acquire() + + with self.assertRaises(RuntimeError): + lock2 = local_file.DirectoryLock(tmpdir) + lock2.acquire() + + def test_releasing(self): + with tempfile.TemporaryDirectory() as tmpdir: + lock = local_file.DirectoryLock(tmpdir) + lock.acquire() + self.assertTrue(lock._is_locked_flag) + lock.release() + self.assertFalse(lock._is_locked_flag) + + def test_context_manager_fail(self): + with tempfile.TemporaryDirectory() as tmpdir: + lock = local_file.DirectoryLock(tmpdir) + + with self.assertRaises(RuntimeError) as cm: + with lock.ensure_locked(): + pass + self.assertIn("is not locked", cm.exception.args[0]) + + def test_context_manager_concurrency(self): + with tempfile.TemporaryDirectory() as tmpdir: + lock = local_file.DirectoryLock(tmpdir) + lock.acquire() + + barrier = threading.Barrier(parties=3, timeout=5) + + def first(): + with lock.ensure_locked(): + barrier.wait() + barrier.wait() + + def second(): + with lock.ensure_locked(): + barrier.wait() + barrier.wait() + + def asserts(): + barrier.wait() + self.assertEqual(2, lock._active_accesses) + barrier.wait() + lock.release() + self.assertEqual(0, lock._active_accesses) + + run_threads([first, second, asserts]) + + def test_context_manager_finish_before_release(self): + with tempfile.TemporaryDirectory() as tmpdir: + lock = local_file.DirectoryLock(tmpdir) + lock.acquire() + + access_barrier = threading.Barrier(parties=2, timeout=5) + release_barrier = threading.Barrier(parties=2, timeout=5) + + def access(): + with lock.ensure_locked(): + access_barrier.wait() + access_barrier.wait() + access_barrier.wait() + + def release(): + release_barrier.wait() + lock.release() + release_barrier.wait() + + def asserts(): + access_barrier.wait() + self.assertTrue(lock._is_locked_flag) # in ensure_locked() + release_barrier.wait() + self.assertTrue(lock._is_locked_flag) # in ensure_locked(), release() started + access_barrier.wait() + access_barrier.wait() + release_barrier.wait() + self.assertFalse(lock._is_locked_flag) # out of ensure_locked(), release() finished + + run_threads([access, release, asserts]) + + class LocalFileBackendTest(TestCase): def setUp(self) -> None: self.identifiable_store = local_file.LocalFileIdentifiableStore(store_path) @@ -27,8 +131,40 @@ def tearDown(self) -> None: try: self.identifiable_store.clear() finally: + self.identifiable_store.close() shutil.rmtree(store_path) + def test_multi_instance_fail_on_init(self): + # Create second store for same path and expect it to fail + with self.assertRaises(RuntimeError) as cm: + local_file.LocalFileIdentifiableStore(store_path) + + self.assertIn("is already in use", cm.exception.args[0]) + + def test_multi_instance_fail_on_check(self): + with tempfile.TemporaryDirectory() as tmpdir: + path = os.path.join(tmpdir, "localdb") + + store1 = local_file.LocalFileIdentifiableStore(path) + store2 = local_file.LocalFileIdentifiableStore(path) + + store1.check_directory(create=True) + with self.assertRaises(RuntimeError) as cm: + store2.check_directory() + + self.assertIn("is already in use", cm.exception.args[0]) + + def test_dir_lock_fail_add(self): + with tempfile.TemporaryDirectory() as tmpdir: + path = os.path.join(tmpdir, "localdb") + store = local_file.LocalFileIdentifiableStore(path) + # store did not acquire dir_lock as path does not exist yet + + with self.assertRaises(RuntimeError) as cm: + store.add(create_example_submodel()) + + self.assertIn("is not locked", cm.exception.args[0]) + def test_identifiable_store_add(self): test_object = create_example_submodel() self.identifiable_store.add(test_object) @@ -166,7 +302,8 @@ def test_reload_discard(self) -> None: example_submodel = create_example_submodel() self.identifiable_store.add(example_submodel) - # Reload the DictIdentifiableStore and discard the example submodel + # Reload the store: release the existing lock before opening a new instance + self.identifiable_store.close() self.identifiable_store = local_file.LocalFileIdentifiableStore(store_path) self.identifiable_store.discard(example_submodel) self.assertNotIn(example_submodel, self.identifiable_store) From f61b5899a9da33ebe142de73f85c0a5fa05bc109 Mon Sep 17 00:00:00 2001 From: hpoeche Date: Mon, 29 Jun 2026 17:09:59 +0200 Subject: [PATCH 2/3] fix small bugs --- sdk/basyx/aas/backend/local_file.py | 3 ++- sdk/test/backend/test_local_file.py | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/sdk/basyx/aas/backend/local_file.py b/sdk/basyx/aas/backend/local_file.py index 9c8ddadd..a1270b1d 100644 --- a/sdk/basyx/aas/backend/local_file.py +++ b/sdk/basyx/aas/backend/local_file.py @@ -112,6 +112,7 @@ def release(self) -> None: # Release flock self._is_locked_flag = False + self._is_releasing = False if self._dir_lock_file is not None: self._dir_lock_file.close() # lock is released by closing fd self._dir_lock_file = None @@ -127,7 +128,7 @@ def ensure_locked(self) -> Generator: :raises RuntimeError: If the directory is not locked on enter """ with self._locking_lock: - if (not self._is_locked_flag) and (not self._is_releasing): + if (not self._is_locked_flag) or self._is_releasing: raise RuntimeError(f"Directory {self.directory_path} is not locked!") self._active_accesses += 1 try: diff --git a/sdk/test/backend/test_local_file.py b/sdk/test/backend/test_local_file.py index 008eab15..7240c021 100644 --- a/sdk/test/backend/test_local_file.py +++ b/sdk/test/backend/test_local_file.py @@ -24,7 +24,7 @@ def run_threads(fns: list[Callable]): fn_futures: list[concurrent.futures.Future] = [] - with concurrent.futures.ThreadPoolExecutor() as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=len(fns)) as executor: for fn in fns: fn_futures.append(executor.submit(fn)) From d620b37718f8c6142addb7e7badb30e11d9fbe55 Mon Sep 17 00:00:00 2001 From: hpoeche Date: Mon, 13 Jul 2026 18:20:49 +0200 Subject: [PATCH 3/3] sdk: add thread-safety to local_file backend Until now the `LocalFileIdentifiableStore` had a TOCTOU race condition in the `add()` and `commit()` methods. After a file was checked for existance a concurrent running `add()` or `discard()` could still create or remove the file before the original invocation accesses it. To fix this, all write accesses to a file need to be mutual exclusive. In this implementation I chose to use a global `_writing_lock` to lock write operations to all files of the store, as locking individual files has no performance benefit, due to serialization of multiple threads through Pythons GIL. This implementation prevents the following concurrency issues: 1. Two concurrent `add()`: Instead of silent overwriting, first call will succeed, second failes with `KeyError`. 2. `commit()` + `discard()`: Instead of `commit()` creating file after `discard()` deleted it, the file is now certainly discarded. --- sdk/basyx/aas/backend/local_file.py | 32 +++++--- sdk/test/backend/test_local_file.py | 118 +++++++++++++++++++++++++++- 2 files changed, 136 insertions(+), 14 deletions(-) diff --git a/sdk/basyx/aas/backend/local_file.py b/sdk/basyx/aas/backend/local_file.py index a1270b1d..c00bb8da 100644 --- a/sdk/basyx/aas/backend/local_file.py +++ b/sdk/basyx/aas/backend/local_file.py @@ -175,6 +175,9 @@ def __init__(self, directory_path: str): = weakref.WeakValueDictionary() self._object_cache_lock = threading.Lock() + # Prevent concurrent write operations to avoid TOCTOU problems + self._write_lock = threading.Lock() + # We need to prevent multiple instances of LocalFileIdentifiableStore performing R/W operations on the same # directory in order to ensure cache validity. The directory is locked as soon as it exists. self._dir_lock = DirectoryLock(self.directory_path) @@ -266,10 +269,11 @@ def add(self, x: model.Identifiable) -> None: :raises KeyError: If an object with the same id exists already in the object store """ logger.debug("Adding object %s to Local File Store ...", repr(x)) - with self._dir_lock.ensure_locked(): - if os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))): - raise KeyError("Identifiable with id {} already exists in local file database".format(x.id)) - self._write_atomic(x) + with self._write_lock: + with self._dir_lock.ensure_locked(): + if os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))): + raise KeyError("Identifiable with id {} already exists in local file database".format(x.id)) + self._write_atomic(x) with self._object_cache_lock: self._object_cache[x.id] = x @@ -280,10 +284,11 @@ def commit(self, x: model.Identifiable) -> None: :param x: The object to persist :raises KeyError: If the object is not present in the store """ - with self._dir_lock.ensure_locked(): - if not os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))): - raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) - self._write_atomic(x) + with self._write_lock: + with self._dir_lock.ensure_locked(): + if not os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))): + raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) + self._write_atomic(x) def discard(self, x: model.Identifiable) -> None: """ @@ -293,11 +298,12 @@ def discard(self, x: model.Identifiable) -> None: :raises KeyError: If the object does not exist in the database """ logger.debug("Deleting object %s from Local File Store database ...", repr(x)) - with self._dir_lock.ensure_locked(): - try: - os.remove("{}/{}.json".format(self.directory_path, self._transform_id(x.id))) - except FileNotFoundError as e: - raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) from e + with self._write_lock: + with self._dir_lock.ensure_locked(): + try: + os.remove("{}/{}.json".format(self.directory_path, self._transform_id(x.id))) + except FileNotFoundError as e: + raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) from e with self._object_cache_lock: self._object_cache.pop(x.id, None) diff --git a/sdk/test/backend/test_local_file.py b/sdk/test/backend/test_local_file.py index 7240c021..fee769d8 100644 --- a/sdk/test/backend/test_local_file.py +++ b/sdk/test/backend/test_local_file.py @@ -10,7 +10,7 @@ import tempfile import threading import concurrent.futures -from typing import Callable +from typing import Callable, cast from unittest import TestCase @@ -307,3 +307,119 @@ def test_reload_discard(self) -> None: self.identifiable_store = local_file.LocalFileIdentifiableStore(store_path) self.identifiable_store.discard(example_submodel) self.assertNotIn(example_submodel, self.identifiable_store) + + +class _GatedLocalFileStore(local_file.LocalFileIdentifiableStore): + """ + Patch :meth:`_write_atomic()` to control and enforce concurrent writes. + """ + + def __init__(self, directory_path: str, *args, **kwargs): + super().__init__(directory_path) + self._gate = threading.Event() # gate to run _write_atomic() + self._gate.set() + self._inside = threading.Event() # signal _write_atomic() entry + + def _write_atomic(self, x: model.Identifiable) -> None: + self._inside.set() + self._gate.wait(timeout=5) + super()._write_atomic(x) + + +class LocalFileBackendConcurrencyTest(TestCase): + def setUp(self): + self.store = _GatedLocalFileStore(store_path) + self.store.check_directory(create=True) + + def tearDown(self): + try: + self.store.clear() + finally: + self.store.close() + shutil.rmtree(store_path) + + def _example_submodels(self) -> tuple[model.Submodel, model.Submodel]: + first_submodel = model.Submodel( + id_='https://example.org/BackendTest', + submodel_element={ + model.Property(id_short='Prop', value_type=model.datatypes.String, value='first') + } + ) + + second_submodel = model.Submodel( + id_='https://example.org/BackendTest', + submodel_element={ + model.Property(id_short='Prop', value_type=model.datatypes.String, value='second') + } + ) + + return first_submodel, second_submodel + + def test_concurrent_add(self): + """Checks that second add for same ID fails""" + first_submodel, second_submodel = self._example_submodels() + + self.store._gate.clear() + self.store._inside.clear() + barrier = threading.Barrier(2, timeout=5) + all_done = threading.Barrier(3, timeout=5) + + def first(): + self.store.add(first_submodel) + all_done.wait() + + def second(): + self.store._inside.wait() + barrier.wait() + with self.assertRaises(KeyError) as ex: + self.store.add(second_submodel) + + all_done.wait() + self.assertIn("already exists", ex.exception.args[0]) + + def control(): + self.store._inside.wait() + barrier.wait() + self.store._gate.set() + + all_done.wait() + submodel = self.store.get_item("https://example.org/BackendTest") + self.assertIsInstance(submodel, model.Submodel) + sm_property = submodel.get_referable("Prop") + self.assertIsInstance(sm_property, model.Property) + self.assertEqual(sm_property.value, "first") + + run_threads([first, second, control]) + + def test_concurrent_commit_discard(self): + """Checks that discard is not overwritten by concurrent commit""" + submodel, altered_submodel = self._example_submodels() + self.store.add(submodel) + + self.store._gate.clear() + self.store._inside.clear() + barrier = threading.Barrier(2, timeout=5) + all_done = threading.Barrier(3, timeout=5) + + def first(): + self.store.commit(submodel) + all_done.wait() + + def second(): + self.store._inside.wait() + barrier.wait() + self.store.discard(submodel) + all_done.wait() + + def control(): + self.store._inside.wait() + barrier.wait() + self.store._gate.set() + + all_done.wait() + with self.assertRaises(KeyError) as ex: + self.store.get_item("https://example.org/BackendTest") + + self.assertIn("No Identifiable", ex.exception.args[0]) + + run_threads([first, second, control])