Skip to content

LocalFileIdentifiableStore: Fix concurrency issues #591

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
hpoeche wants to merge 5 commits into
base: develop
Choose a base branch
from
Open

LocalFileIdentifiableStore: Fix concurrency issues #591

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8e261bb
sdk: fix local_file backend multi-instance safety
hpoeche Jun 28, 2026
f61b589
fix small bugs
hpoeche Jun 29, 2026
d620b37
sdk: add thread-safety to local_file backend
hpoeche Jul 13, 2026
bceccfc
Merge branch 'develop' into fix/554
hpoeche Jul 28, 2026
d6cbb41
Fix: regorganize imports
hpoeche Jul 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
260 changes: 205 additions & 55 deletions sdk/basyx/aas/backend/local_file.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
The :class:`~LocalFileIdentifiableStore` handles adding, deleting and otherwise managing
the AAS objects in a specific Directory.
"""

import contextlib
import hashlib
import json
import logging
Expand All @@ -20,7 +20,8 @@
import threading
import warnings
import weakref
from typing import Iterator
from pathlib import Path
from typing import Generator, Iterator, Optional, TextIO

from basyx.aas import model

Expand All @@ -29,26 +30,143 @@
logger = logging.getLogger(__name__)


class DirectoryLock:
"""
Implementation of a Lock on a directory. Uses POSIX ``flock`` on ``<directory>/.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
Comment on lines +59 to +64

@s-heppner s-heppner Aug 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I remember we did this to deal with typing issues, but I don't like how it would re-import on every acquire().
Would something like this on the module level?

try:
    import fcntl as _fcntl
except ImportError:
    _fcntl = None  # type: ignore  # Windows does not have directory locking


def acquire(self) -> None:
"""
Acquires the lock on the directory non-blocking. If the ``<directory>/.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
Comment on lines +94 to +95

@s-heppner s-heppner Aug 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Isn't OSError a bit generic? According to my AI, contention should raise BlockingIOError specifically.

Catching this would guard against a filesystem without flock support, producing a misleading "already in use by another instance" that sends the user hunting for a nonexistent second process.

Can you verify please?

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
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

@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) or 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.
Comment on lines -39 to -44

@s-heppner s-heppner Aug 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This still holds true for Windows, as on Windows the whole locking mechanism is simply no-op right? Maybe we should keep and adapt the warning? Or maybe even add a flag that a Windows user needs to set in order to acknowledge that the store is not concurrency safe?

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("/")

Expand All @@ -62,6 +180,26 @@ 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)
if os.path.exists(self.directory_path):
self._acquire_dir_lock()
Comment thread
s-heppner marked this conversation as resolved.

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)
Expand All @@ -78,19 +216,21 @@ 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:
"""
Retrieve an AAS object from the local file by its identifier hash

: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(
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_
)
Expand All @@ -107,13 +247,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(
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
)
Expand Down Expand Up @@ -150,15 +291,17 @@ 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._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
Comment on lines +294 to 306

@s-heppner s-heppner Aug 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this could lead to the object_cache diverging from the actual file's content. Shouldn't the object cache be updated within the write lock?


Expand All @@ -169,13 +312,15 @@ 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(
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)
self._write_atomic(x)

def discard(self, x: model.Identifiable) -> None:
"""
Expand All @@ -185,14 +330,16 @@ 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._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)
Comment on lines +333 to 344

@s-heppner s-heppner Aug 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My comment at add() is a bit easier to understand here.
Let's say the object gets deleted by thread A.
It finishes deleting the file, gives up the _write_lock.
Then while it's between line 342 and 343, thread B reads the object from the cache, which is now effectivly a ghost.

That's why I think the object_cache update should happen inside the _write_lock.


Expand All @@ -212,9 +359,10 @@ 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:
"""
Expand All @@ -223,9 +371,10 @@ 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]:
"""
Expand All @@ -235,9 +384,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:
Expand Down
Loading
Loading