#!/usr/bin/env python3
+# © Copyright 2021-2022, Scott Gasch
+
"""File-based locking helper."""
from __future__ import annotations
-
import contextlib
import datetime
import json
"""The contents we'll write to each lock file."""
pid: int
+ """The pid of the process that holds the lock"""
+
commandline: str
+ """The commandline of the process that holds the lock"""
+
expiration_timestamp: Optional[float]
+ """When this lock will expire as seconds since Epoch"""
class LockFile(contextlib.AbstractContextManager):
"""A file locking mechanism that has context-manager support so you
- can use it in a with statement. e.g.
-
- with LockFile('./foo.lock'):
- # do a bunch of stuff... if the process dies we have a signal
- # handler to do cleanup. Other code (in this process or another)
- # that tries to take the same lockfile will block. There is also
- # some logic for detecting stale locks.
+ can use it in a with statement. e.g.::
+ with LockFile('./foo.lock'):
+ # do a bunch of stuff... if the process dies we have a signal
+ # handler to do cleanup. Other code (in this process or another)
+ # that tries to take the same lockfile will block. There is also
+ # some logic for detecting stale locks.
"""
def __init__(
expiration_timestamp: Optional[float] = None,
override_command: Optional[str] = None,
) -> None:
+ """C'tor.
+
+ Args:
+ lockfile_path: path of the lockfile to acquire
+ do_signal_cleanup: handle SIGINT and SIGTERM events by
+ releasing the lock before exiting
+ expiration_timestamp: when our lease on the lock should
+ expire (as seconds since the Epoch). None means the
+ lock will not expire until we explicltly release it.
+ override_command: don't use argv to determine our commandline
+ rather use this instead if provided.
+ """
self.is_locked: bool = False
self.lockfile: str = lockfile_path
self.locktime: Optional[int] = None
self.expiration_timestamp = expiration_timestamp
def locked(self):
+ """Is it locked currently?"""
return self.is_locked
def available(self):
+ """Is it available currently?"""
return not os.path.exists(self.lockfile)
def try_acquire_lock_once(self) -> bool:
+ """Attempt to acquire the lock with no blocking.
+
+ Returns:
+ True if the lock was acquired and False otherwise.
+ """
logger.debug("Trying to acquire %s.", self.lockfile)
try:
# Attempt to create the lockfile. These flags cause
backoff_factor: float = 2.0,
max_attempts=5,
) -> bool:
+ """Attempt to acquire the lock repeatedly with retries and backoffs.
+
+ Args:
+ initial_delay: how long to wait before retrying the first time
+ backoff_factor: a float >= 1.0 the multiples the current retry
+ delay each subsequent time we attempt to acquire and fail
+ to do so.
+ max_attempts: maximum number of times to try before giving up
+ and failing.
+
+ Returns:
+ True if the lock was acquired and False otherwise.
+ """
+
@decorator_utils.retry_if_false(
tries=max_attempts, delay_sec=initial_delay, backoff=backoff_factor
)
return _try_acquire_lock_with_retries()
def release(self):
+ """Release the lock"""
try:
os.unlink(self.lockfile)
except Exception as e: