v8.2.1: Add a utility for managing file locks to prevent concurrent executions

Introduced `lock_utils.py` to manage file-based locks. Includes functions to acquire and release locks, ensuring the safety of processes by detecting stale or live locks with PID validation.
master
Aidaho 2025-06-08 12:56:41 +03:00
parent f749f7366e
commit a3eb55749d
1 changed files with 58 additions and 0 deletions

View File

@ -0,0 +1,58 @@
import os
import time
import atexit
import psutil
LOCK_FILE = "/tmp/roxy-wi.lock"
def acquire_file_lock() -> bool:
if os.path.exists(LOCK_FILE):
try:
with open(LOCK_FILE, "r") as f:
pid_str, ts_str = f.read().strip().split(",")
old_pid = int(pid_str)
old_time = float(ts_str)
except Exception as e:
print(f"[LOCK] Corrupt lock file. Removing. ({e})")
os.remove(LOCK_FILE)
_create_lock()
return False
if psutil.pid_exists(old_pid):
print( f"[LOCK] Lock held by live process PID {old_pid} (started {time.ctime(old_time)}).")
return True
else:
print( f"[LOCK] Detected dead PID {old_pid}. Lock is stale. Taking ownership.")
os.remove(LOCK_FILE)
_create_lock()
return False
else:
_create_lock()
return False
def _create_lock():
try:
with open(LOCK_FILE, "w") as f:
f.write(f"{os.getpid()},{time.time()}")
print( f"[LOCK] Lock acquired by PID {os.getpid()}")
except Exception as e:
print( f"error: [LOCK] Failed to write lock file: {e}")
raise e
# Register cleanup on normal exit
atexit.register(release_file_lock)
def release_file_lock():
if not os.path.exists(LOCK_FILE):
return
try:
with open(LOCK_FILE, "r") as f:
pid_str, _ = f.read().strip().split(",")
if int(pid_str) == os.getpid():
os.remove(LOCK_FILE)
print( f"[LOCK] Lock released by PID {os.getpid()}")
except Exception as e:
print( f"[LOCK] Error releasing lock: {e}")