lock

LockManager

File-based locking for thread-safe resource management

The LockManager module provides a robust file-based locking mechanism to ensure thread-safe operations across multiple threads or processes. It allows acquiring and releasing locks on resources identified by scan, fileset, or file IDs, with support for both shared (read) and exclusive (write) locks.

Key Features

• Thread-safe lock acquisition and release using file-based locking mechanisms
• Support for both shared (multiple readers) and exclusive (single writer) locks
• Lock timeout to prevent indefinite waiting for lock acquisition
• Automatic cleanup of stale locks on initialization
• Monitoring and debugging capabilities with lock metadata storage
• Support for multiple lock levels: Scan, Fileset, and File

Usage Examples

from plantdb.commons.fsdb.lock import LockManager, LockType

# Initialize the lock manager with a base path
manager = LockManager('/path/to/database')

# Acquire an exclusive lock for a scan
with manager.acquire_lock('scan123', LockType.EXCLUSIVE, user='user1', level='scan'):
    # Perform critical section operations...

# Acquire a shared lock for a fileset
with manager.acquire_lock('scan123/fileset1', LockType.SHARED, user='user1', level='fileset'):
    # Perform critical section operations...

# Acquire an exclusive lock for a file
with manager.acquire_lock('scan123/fileset1/file1', LockType.EXCLUSIVE, user='user1', level='file'):
    # Perform critical section operations...

LockError

LockError(message)

Bases: Exception

Raised when the lock acquisition fails.

Source code in plantdb/commons/fsdb/lock.py

def __init__(self, message: str):
    self.message = message
    super().__init__(self.message)

LockLevel

Bases: Enum

Enumeration of lock levels supported by the LockManager.

Used by:

• Reference API commons commons fsdb lock  LockManager
  •  acquire_lock
  •  get_lock_status

LockManager

LockManager(base_path, default_timeout=30.0, **kwargs)

Acquires and releases file-based locks for thread-safe resource management.

This class provides functionality for acquiring and releasing file-based locks, ensuring thread-safe operations across multiple threads or processes. Locks are stored in a designated subdirectory within the specified base path, which is created if it does not exist. On initialization, stale lock files are cleaned up to maintain consistency.

The class leverages thread locks to manage access to shared resources and uses file-based locking mechanisms to ensure exclusive access across processes.

Attributes:

Name	Type	Description
base_path	str	The base directory of the database.
default_timeout	float	The default timeout duration in seconds when attempting to acquire a lock. Default is 30.0 seconds.
locks_dir	str	Location of the directory where lock files are stored.
_active_locks	Dict[str, Dict]	Dictionary mapping lock names to dictionaries containing information about the lock including: 'type': the type of the lock (shared or exclusive), 'user': the user who acquired the lock (if available), 'timestamp': the timestamp when the lock was acquired (if available), 'count': int indicating how many locks are active for a given resource and lock type combination,
_lock_files	Dict[str, int]	The file descriptors (int) for each active lock in the dictionary mapping.
_thread_lock	RLock	A thread-safe lock used to synchronize access to the active locks dictionary and file descriptors.

Examples:

>>> from plantdb.commons.fsdb.lock import LockManager
>>> manager = LockManager('/path/to/local/database')
>>> lock = manager.acquire_lock('scan123', LockType.EXCLUSIVE, user='user1', level='scan')

Lock manager constructor.

Parameters:

Name	Type	Description	Default
base_path	str or Path	The base directory of the database.	required
default_timeout	float	The default timeout duration in seconds when attempting to acquire a lock. Default is 30.0 seconds.	30.0

Other Parameters:

Name	Type	Description
log_level	str	The log level (e.g., DEBUG, INFO). Default is INFO.

Source code in plantdb/commons/fsdb/lock.py

def __init__(self, base_path: str, default_timeout: float = 30.0, **kwargs):
    """
    Lock manager constructor.

    Parameters
    ----------
    base_path : str or pathlib.Path
        The base directory of the database.
    default_timeout : float, optional
        The default timeout duration in seconds when attempting to acquire a lock.
        Default is 30.0 seconds.

    Other Parameters
    ----------------
    log_level : str, optional
        The log level (e.g., `DEBUG`, `INFO`). Default is `INFO`.
    """
    self.base_path = base_path
    self.default_timeout = default_timeout
    self.locks_dir = os.path.join(base_path, ".locks")
    self._active_locks: Dict[str, Dict] = {}  # Track active locks
    self._lock_files: Dict[str, int] = {}  # File descriptors for locks
    self._thread_lock = threading.RLock()  # Thread-safe operations

    # Store the last time we emitted a warning per resource
    self._warning_timestamps: Dict[str, float] = {}
    # How long we wait before emitting the same warning again (seconds)
    self._warning_debounce_interval: float = kwargs.get("warning_debounce_interval", 5.0)

    # Test write capability in base_path
    try:
        test_file = os.path.join(base_path, '.write_test.tmp')
        with open(test_file, 'w') as f:
            f.write('test')
        os.remove(test_file)
    except PermissionError as e:
        # Create a logger without the log_file as it will not be writable either
        logger = get_logger(__name__ + '.LockManager', log_level="DEBUG")
        logger.error(f"Cannot write to base_path `{base_path}`.")
        logger.debug(f"Current UID={os.getuid()}, GID={os.getgid()}")
        raise e

    # Ensure locks directory exists
    os.makedirs(self.locks_dir, exist_ok=True)
    # Initialize lock manager logger
    self.logger = get_logger(__name__ + '.LockManager',
                             log_file=kwargs.get("log_file", os.path.join(base_path, '.locks', 'lock_manager.log')),
                             log_level=kwargs.get("log_level", "INFO"))
    self.logger.debug(f"Initialized LockManager with base path: `{base_path}`")
    # Clean up stale locks on initialization
    self._cleanup_stale_locks()

acquire_lock

acquire_lock(resource_id, lock_type, user, level, timeout=None)

Acquire a lock for a specific resource and lock type.

This method attempts to acquire a lock (either shared or exclusive) for a given resource ID. If the lock is successfully acquired, it yields control to the calling code. Upon completion of the calling code, it ensures that the lock is released.

Parameters:

Name	Type	Description	Default
resource_id	str	The unique identifier for the resource (scan_id, scan_id/fileset_id, or scan_id/fileset_id/file_id).	required
lock_type	LockType	The type of lock to acquire (shared or exclusive).	required
user	str	The user requesting the lock.	required
level	LockLevel	The level of the lock (scan, fileset, or file).	required
timeout	Optional[float]	The maximum time to wait for acquiring the lock, in seconds. If not provided, uses a default timeout value.	None

Raises:

Type	Description
LockTimeoutError	If the lock could not be acquired within the specified timeout period or if an exclusive lock is already held by another user.
OSError	If there is an error with file operations while trying to acquire the lock.
IOError	If there is an input/output error while trying to acquire the lock.

Notes

This method uses a context manager and should be used within a 'with' statement. It handles both shared and exclusive locks, allowing multiple acquisitions for shared locks but raising an exception if an exclusive lock is already held.

Examples:

>>> import tempfile
>>> from plantdb.commons.fsdb.lock import LockManager
>>> from plantdb.commons.fsdb.lock import LockType
>>> from plantdb.commons.fsdb.lock import LockLevel
>>> manager = LockManager(tempfile.gettempdir())
>>> with manager.acquire_lock('scan123', LockType.EXCLUSIVE, user='user1', level=LockLevel.SCAN):
...     print("scan123/scan/exclusive" in manager._active_locks)
True
>>> with manager.acquire_lock('scan123/test_fileset', LockType.EXCLUSIVE, user='user1', level=LockLevel.FILESET):
...     print("scan123/test_fileset/fileset/exclusive" in manager._active_locks)
True

Source code in plantdb/commons/fsdb/lock.py

@contextmanager
def acquire_lock(self, resource_id: str, lock_type: LockType, user: str, level: LockLevel, timeout: Optional[float] = None):
    """
    Acquire a lock for a specific resource and lock type.

    This method attempts to acquire a lock (either shared or exclusive) for a given resource ID.
    If the lock is successfully acquired, it yields control to the calling code.
    Upon completion of the calling code, it ensures that the lock is released.

    Parameters
    ----------
    resource_id : str
        The unique identifier for the resource (scan_id, scan_id/fileset_id, or scan_id/fileset_id/file_id).
    lock_type : LockType
        The type of lock to acquire (shared or exclusive).
    user : str
        The user requesting the lock.
    level : LockLevel
        The level of the lock (scan, fileset, or file).
    timeout : Optional[float], optional
        The maximum time to wait for acquiring the lock, in seconds. If not provided,
        uses a default timeout value.

    Raises
    ------
    LockTimeoutError
        If the lock could not be acquired within the specified timeout period or if an exclusive
        lock is already held by another user.
    OSError
        If there is an error with file operations while trying to acquire the lock.
    IOError
        If there is an input/output error while trying to acquire the lock.

    Notes
    -----
    This method uses a context manager and should be used within a 'with' statement. It handles both shared
    and exclusive locks, allowing multiple acquisitions for shared locks but raising an exception if an exclusive
    lock is already held.

    Examples
    --------
    >>> import tempfile
    >>> from plantdb.commons.fsdb.lock import LockManager
    >>> from plantdb.commons.fsdb.lock import LockType
    >>> from plantdb.commons.fsdb.lock import LockLevel
    >>> manager = LockManager(tempfile.gettempdir())
    >>> with manager.acquire_lock('scan123', LockType.EXCLUSIVE, user='user1', level=LockLevel.SCAN):
    ...     print("scan123/scan/exclusive" in manager._active_locks)
    True
    >>> with manager.acquire_lock('scan123/test_fileset', LockType.EXCLUSIVE, user='user1', level=LockLevel.FILESET):
    ...     print("scan123/test_fileset/fileset/exclusive" in manager._active_locks)
    True
    """
    timeout = timeout or self.default_timeout
    # Create a unique lock key based on resource_id, level, and lock type
    lock_key = f"{resource_id}/{level.value}/{lock_type.value}"

    self.logger.debug(f"Attempting to acquire {lock_type.value} lock for {level.value} {resource_id} by user {user}")

    with self._thread_lock:
        # Check if we already have this lock
        if lock_key in self._active_locks:
            # For shared locks, allow multiple acquisitions
            if lock_type == LockType.SHARED:
                self._active_locks[lock_key]['count'] += 1
                self.logger.debug(f"Incrementing shared lock count for {level.value} {resource_id}")
                try:
                    yield
                    return
                finally:
                    self._active_locks[lock_key]['count'] -= 1
                    if self._active_locks[lock_key]['count'] <= 0:
                        self._release_lock(resource_id, lock_type, level)
            else:
                # For exclusive locks, allow reentrant acquisition by the same user/thread
                current_lock_user = self._active_locks[lock_key].get('user')
                if current_lock_user == user:
                    self._active_locks[lock_key]['count'] += 1
                    self.logger.debug(f"Incrementing exclusive lock count for {level.value} {resource_id} by same user {user}")
                    try:
                        yield
                        return
                    finally:
                        self._active_locks[lock_key]['count'] -= 1
                        if self._active_locks[lock_key]['count'] <= 0:
                            self._release_lock(resource_id, lock_type, level)
                else:
                    raise LockError(f"Exclusive lock already held for {level.value} {resource_id}")

    # Acquire new lock
    acquired = False
    start_time = time.time()

    # Helper to check for conflicting active locks of a different type
    def _has_active_lock(other_type: LockType) -> bool:
        other_key = f"{resource_id}/{level.value}/{other_type.value}"
        return other_key in self._active_locks

    try:
        lock_file_path = self._get_lock_file_path(resource_id, level)

        attempt = 0
        while time.time() - start_time < timeout:
            attempt += 1

            # Respect intra‑process lock conflicts before touching the file
            if lock_type == LockType.SHARED and _has_active_lock(LockType.EXCLUSIVE):
                # An exclusive lock is active → treat as unavailable and retry
                self.logger.debug(
                    f"Shared lock request for {level.value} {resource_id} blocked by active exclusive lock"
                )
                time.sleep(0.5)
                continue

            if lock_type == LockType.EXCLUSIVE and _has_active_lock(LockType.SHARED):
                # One or more shared locks are active → wait
                self.logger.debug(
                    f"Exclusive lock request for {level.value} {resource_id} blocked by active shared lock(s)"
                )
                time.sleep(0.5)
                continue

            try:
                # Open lock file
                lock_fd = os.open(lock_file_path, os.O_CREAT | os.O_WRONLY | os.O_TRUNC)

                # Determine fcntl flags based on lock type
                if lock_type == LockType.EXCLUSIVE:
                    lock_flags = fcntl.LOCK_EX | fcntl.LOCK_NB
                else:  # SHARED
                    lock_flags = fcntl.LOCK_SH | fcntl.LOCK_NB

                # Try to acquire the lock (non-blocking)
                fcntl.flock(lock_fd, lock_flags)
                # SUCCESS - clear any stale warning timestamp for this resource
                self._warning_timestamps.pop(resource_id, None)

                # Lock acquired successfully
                with self._thread_lock:
                    self._active_locks[lock_key] = {
                        'type': lock_type,
                        'user': user,
                        'timestamp': time.time(),
                        'count': 1
                    }
                    self._lock_files[lock_key] = lock_fd

                self._write_lock_info(resource_id, lock_type, user, level)
                acquired = True
                self.logger.debug(
                    f"Successfully acquired {lock_type.value} lock for {level.value} {resource_id} "
                    f"(attempt {attempt})"
                )
                break

            except (OSError, IOError) as e:
                # Lock not available, close fd and retry
                try:
                    os.close(lock_fd)
                except:
                    pass

                now = time.time()
                last_warn = self._warning_timestamps.get(resource_id, 0.0)
                if now - last_warn >= self._warning_debounce_interval:
                    self.logger.warning(
                        f"Lock acquisition attempt failed for {level.value} {resource_id} (attempt {attempt}) - "
                        f"retrying... [pid={os.getpid()}, tid={threading.get_ident()}]"
                    )
                    self._warning_timestamps[resource_id] = now
                else:
                    # We’re within the debounce window - log at DEBUG instead of spamming WARN
                    self.logger.debug(
                        f"Retrying lock for {level.value} {resource_id} (attempt {attempt}) - still waiting"
                    )
                time.sleep(0.1)  # Brief pause before retry

        if not acquired:
            raise LockTimeoutError(
                f"Could not acquire {lock_type.value} lock for {level.value} {resource_id} within {timeout} seconds")

        # Yield control to the calling code
        yield

    finally:
        # Always release the lock
        if acquired:
            self._release_lock(resource_id, lock_type, level)

cleanup_all_locks

cleanup_all_locks()

Emergency cleanup of all locks (use with caution)

Source code in plantdb/commons/fsdb/lock.py

def cleanup_all_locks(self):
    """Emergency cleanup of all locks (use with caution)"""
    self.logger.warning("Cleaning up all active locks...")
    with self._thread_lock:
        for lock_key in list(self._active_locks.keys()):
            # Parse the lock key to extract resource_id and level
            parts = lock_key.split('/', 2)
            if len(parts) != 3:
                continue

            resource_id, level_str, lock_type_str = parts
            level = LockLevel(level_str)
            lock_type = LockType(lock_type_str)
            self._release_lock(resource_id, lock_type, level)
    return

get_lock_status

get_lock_status(resource_id, level)

Retrieve the current lock status for a given resource ID.

This method checks all active locks and determines if there is an exclusive or shared lock associated with the specified resource. It returns a dictionary containing information about the exclusive lock (if any) and all shared locks associated with the resource.

Parameters:

Name	Type	Description	Default
resource_id	str	The unique identifier of the resource for which to retrieve the lock status.	required
level	LockLevel	The level of the lock (scan, fileset, or file).	required

Returns:

Name	Type	Description
status	dict	A dictionary with two keys: - 'exclusive': Information about the exclusive lock (if any), or None. This is a nested dictionary containing the user and timestamp. - 'shared': A list of dictionaries, each representing a shared lock. Each dictionary contains the user, timestamp, and count.

Notes

This method iterates through all active locks to find matches with the given resource ID. It distinguishes between exclusive and shared locks based on their type.

Examples:

>>> import tempfile
>>> from plantdb.commons.fsdb.lock import LockManager
>>> from plantdb.commons.fsdb.lock import LockType
>>> from plantdb.commons.fsdb.lock import LockLevel
>>> manager = LockManager(tempfile.gettempdir())
>>> with manager.acquire_lock('scan123', LockType.EXCLUSIVE, user='user1', level=LockLevel.SCAN):
...     status = manager.get_lock_status("scan123", LockLevel.SCAN)
...     print(status)
{'exclusive': {'user': 'user1', 'timestamp': 1775734565.6584513}, 'shared': []}
>>> with manager.acquire_lock('scan123/test_fileset', LockType.EXCLUSIVE, user='user1', level=LockLevel.FILESET):
...     status = manager.get_lock_status("scan123/test_fileset", LockLevel.FILESET)
...     print(status)
{'exclusive': {'user': 'user1', 'timestamp': 1775734690.2060757}, 'shared': []}

Source code in plantdb/commons/fsdb/lock.py

def get_lock_status(self, resource_id: str, level: LockLevel) -> Dict:
    """
    Retrieve the current lock status for a given resource ID.

    This method checks all active locks and determines if there is an
    exclusive or shared lock associated with the specified resource. It returns
    a dictionary containing information about the exclusive lock (if any) and
    all shared locks associated with the resource.

    Parameters
    ----------
    resource_id : str
        The unique identifier of the resource for which to retrieve the lock status.
    level : LockLevel
        The level of the lock (scan, fileset, or file).

    Returns
    -------
    status : dict
        A dictionary with two keys:
            - 'exclusive': Information about the exclusive lock (if any), or None.
              This is a nested dictionary containing the user and timestamp.
            - 'shared': A list of dictionaries, each representing a shared lock.
              Each dictionary contains the user, timestamp, and count.

    Notes
    -----
    This method iterates through all active locks to find matches with the given resource ID.
    It distinguishes between exclusive and shared locks based on their type.

    Examples
    --------
    >>> import tempfile
    >>> from plantdb.commons.fsdb.lock import LockManager
    >>> from plantdb.commons.fsdb.lock import LockType
    >>> from plantdb.commons.fsdb.lock import LockLevel
    >>> manager = LockManager(tempfile.gettempdir())
    >>> with manager.acquire_lock('scan123', LockType.EXCLUSIVE, user='user1', level=LockLevel.SCAN):
    ...     status = manager.get_lock_status("scan123", LockLevel.SCAN)
    ...     print(status)
    {'exclusive': {'user': 'user1', 'timestamp': 1775734565.6584513}, 'shared': []}
    >>> with manager.acquire_lock('scan123/test_fileset', LockType.EXCLUSIVE, user='user1', level=LockLevel.FILESET):
    ...     status = manager.get_lock_status("scan123/test_fileset", LockLevel.FILESET)
    ...     print(status)
    {'exclusive': {'user': 'user1', 'timestamp': 1775734690.2060757}, 'shared': []}
    """
    status = {'exclusive': None, 'shared': []}

    # Look for locks matching this resource and level
    for lock_key, lock_info in self._active_locks.items():
        # Extract resource_id and level from the lock key
        # Format: resource_id/lock_level/lock_type
        parts = lock_key.split('/')
        if len(parts) < 3:
            continue

        lock_type = parts.pop(-1)
        lock_level = parts.pop(-1)
        lock_resource_id = "/".join(parts)
        # Check if this lock belongs to the requested resource and level
        if lock_resource_id == resource_id and lock_level == level.value:
            if lock_info['type'] == LockType.EXCLUSIVE:
                status['exclusive'] = {
                    'user': lock_info['user'],
                    'timestamp': lock_info['timestamp']
                }
            else:  # SHARED
                status['shared'].append({
                    'user': lock_info['user'],
                    'timestamp': lock_info['timestamp'],
                    'count': lock_info['count']
                })

    return status

LockTimeoutError

LockTimeoutError(message='Error: Lock acquisition timed-out!')

Bases: Exception

Raised when the lock acquisition times out

Raised by:

• Reference API commons commons fsdb lock  LockManager  acquire_lock

Source code in plantdb/commons/fsdb/lock.py

def __init__(self, message: str = "Error: Lock acquisition timed-out!"):
    self.message = message
    super().__init__(self.message)