lock

plantdb.commons.fsdb.lock

ScanLockManager

File-based locking for thread-safe resource management

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

Usage Examples

from plantdb.commons.fsdb.lock import ScanLockManager

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

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

# Check the status of locks for 'scan123'
status = manager.get_lock_status('scan123')
print(status)

LockError

LockError(message)

Bases: Exception

Raised when lock acquisition fails.

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

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

LockTimeoutError

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

Bases: Exception

Raised when lock acquisition times out

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)

ScanLockManager

ScanLockManager(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 (scan ID + lock type) 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 scan 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 ScanLockManager
>>> manager = ScanLockManager('/path/to/local/database')
>>> lock = manager.acquire_lock('scan123')

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
    # Ensure locks directory exists
    os.makedirs(self.locks_dir, exist_ok=True)
    # Initialize lock manager logger
    self.logger = get_logger(__name__ + '.ScanLockManager',
                             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 ScanLockManager with base path: `{base_path}`")
    # Clean up stale locks on initialization
    self._cleanup_stale_locks()

acquire_lock

acquire_lock(scan_id, lock_type, user, timeout=None)

Acquire a lock for a specific scan and lock type.

This method attempts to acquire a lock (either shared or exclusive) for a given scan 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
scan_id	str	The unique identifier for the scan.	required
lock_type	LockType	The type of lock to acquire (shared or exclusive).	required
user	str	The user requesting the lock.	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.

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

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

    This method attempts to acquire a lock (either shared or exclusive) for a given scan 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
    ----------
    scan_id : str
        The unique identifier for the scan.
    lock_type : LockType
        The type of lock to acquire (shared or exclusive).
    user : str
        The user requesting the lock.
    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.
    """
    timeout = timeout or self.default_timeout
    lock_key = f"{scan_id}_{lock_type.value}"

    self.logger.debug(f"Attempting to acquire {lock_type.value} lock for scan {scan_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 {scan_id}")
                try:
                    yield
                    return
                finally:
                    self._active_locks[lock_key]['count'] -= 1
                    if self._active_locks[lock_key]['count'] <= 0:
                        self._release_lock(scan_id, lock_type)
            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 {scan_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(scan_id, lock_type)
                else:
                    raise LockError(f"Exclusive lock already held for scan {scan_id}")

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

    try:
        lock_file_path = self._get_lock_file_path(scan_id)

        while time.time() - start_time < timeout:
            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)

                # 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(scan_id, lock_type, user)
                acquired = True
                self.logger.info(f"Successfully acquired {lock_type.value} lock for scan {scan_id}")
                break

            except (OSError, IOError) as e:
                # Lock not available, close fd and retry
                try:
                    os.close(lock_fd)
                except:
                    pass
                self.logger.warning(f"Lock acquisition attempt failed for {scan_id}, retrying...")
                time.sleep(0.1)  # Brief pause before retry

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

        # Yield control to the calling code
        yield

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

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()):
            scan_id, lock_type_str = lock_key.split('_', 1)
            lock_type = LockType(lock_type_str)
            self._release_lock(scan_id, lock_type)
    return

get_lock_status

get_lock_status(scan_id)

Retrieve the current lock status for a given scan ID.

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

Parameters:

Name	Type	Description	Default
scan_id	str	The unique identifier of the scan for which to retrieve the lock status.	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 scan ID. It distinguishes between exclusive and shared locks based on their type.

Examples:

>>> from plantdb.commons.fsdb.lock import ScanLockManager
>>> manager = ScanLockManager('/path/to/local/database')
>>> lock = manager.acquire_lock('scan123')
>>> status = manager.get_lock_status("scan123")
>>> print(status)
{'exclusive': None, 'shared': [{'user': 'user1', 'timestamp': '2023-01-01T12:00:00', 'count': 1}]}

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

def get_lock_status(self, scan_id: str) -> Dict:
    """
    Retrieve the current lock status for a given scan ID.

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

    Parameters
    ----------
    scan_id : str
        The unique identifier of the scan for which to retrieve the lock status.

    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 scan ID.
    It distinguishes between exclusive and shared locks based on their type.

    Examples
    --------
    >>> from plantdb.commons.fsdb.lock import ScanLockManager
    >>> manager = ScanLockManager('/path/to/local/database')
    >>> lock = manager.acquire_lock('scan123')
    >>> status = manager.get_lock_status("scan123")
    >>> print(status)
    {'exclusive': None, 'shared': [{'user': 'user1', 'timestamp': '2023-01-01T12:00:00', 'count': 1}]}
    """
    status = {'exclusive': None, 'shared': []}

    for lock_key, lock_info in self._active_locks.items():
        if lock_key.startswith(scan_id):
            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