Use file existence for registry mutex (#959)

Part of [Service provider auctions for web deployments](https://www.notion.so/Service-provider-auctions-for-web-deployments-104a6b22d47280dbad51d28aa3a91d75)

Reviewed-on: cerc-io/stack-orchestrator#959
Reviewed-by: ashwin <ashwin@noreply.git.vdb.to>
Co-authored-by: Prathamesh Musale <prathamesh.musale0@gmail.com>
Co-committed-by: Prathamesh Musale <prathamesh.musale0@gmail.com>
This commit is contained in:
Prathamesh Musale 2024-10-29 04:05:35 +00:00 committed by ashwin
parent 5f8e809b2d
commit f64ef5d128
2 changed files with 62 additions and 14 deletions

View File

@ -1,8 +1,59 @@
import fcntl
from functools import wraps from functools import wraps
import os
import time
# Define default file path for the lock # Define default file path for the lock
DEFAULT_LOCK_FILE_PATH = "/tmp/registry_mutex_lock_file" DEFAULT_LOCK_FILE_PATH = "/tmp/registry_mutex_lock_file"
LOCK_TIMEOUT = 30
LOCK_RETRY_INTERVAL = 3
def acquire_lock(client, lock_file_path, timeout):
# Lock alreay acquired by the current client
if client.mutex_lock_acquired:
return
while True:
try:
# Check if lock file exists and is potentially stale
if os.path.exists(lock_file_path):
with open(lock_file_path, 'r') as lock_file:
timestamp = float(lock_file.read().strip())
# If lock is stale, remove the lock file
if time.time() - timestamp > timeout:
print(f"Stale lock detected, removing lock file {lock_file_path}")
os.remove(lock_file_path)
else:
print(f"Lock file {lock_file_path} exists and is recent, waiting...")
time.sleep(LOCK_RETRY_INTERVAL)
continue
# Try to create a new lock file with the current timestamp
fd = os.open(lock_file_path, os.O_CREAT | os.O_EXCL | os.O_RDWR)
with os.fdopen(fd, 'w') as lock_file:
lock_file.write(str(time.time()))
client.mutex_lock_acquired = True
print(f"Registry lock acquired, {lock_file_path}")
# Lock successfully acquired
return
except FileExistsError:
print(f"Lock file {lock_file_path} exists, waiting...")
time.sleep(LOCK_RETRY_INTERVAL)
def release_lock(client, lock_file_path):
try:
os.remove(lock_file_path)
client.mutex_lock_acquired = False
print(f"Registry lock released, {lock_file_path}")
except FileNotFoundError:
# Lock file already removed
pass
def registry_mutex(): def registry_mutex():
@ -13,18 +64,13 @@ def registry_mutex():
if self.mutex_lock_file: if self.mutex_lock_file:
lock_file_path = self.mutex_lock_file lock_file_path = self.mutex_lock_file
with open(lock_file_path, 'w') as lock_file: # Acquire the lock before running the function
acquire_lock(self, lock_file_path, LOCK_TIMEOUT)
try: try:
# Try to acquire the lock return func(self, *args, **kwargs)
fcntl.flock(lock_file, fcntl.LOCK_EX)
# Call the actual function
result = func(self, *args, **kwargs)
finally: finally:
# Always release the lock # Release the lock after the function completes
fcntl.flock(lock_file, fcntl.LOCK_UN) release_lock(self, lock_file_path)
return result
return wrapper return wrapper

View File

@ -117,7 +117,6 @@ class LaconicRegistryClient:
def __init__(self, config_file, log_file=None, mutex_lock_file=None): def __init__(self, config_file, log_file=None, mutex_lock_file=None):
self.config_file = config_file self.config_file = config_file
self.log_file = log_file self.log_file = log_file
self.mutex_lock_file = mutex_lock_file
self.cache = AttrDict( self.cache = AttrDict(
{ {
"name_or_id": {}, "name_or_id": {},
@ -126,6 +125,9 @@ class LaconicRegistryClient:
} }
) )
self.mutex_lock_file = mutex_lock_file
self.mutex_lock_acquired = False
def whoami(self, refresh=False): def whoami(self, refresh=False):
if not refresh and "whoami" in self.cache: if not refresh and "whoami" in self.cache:
return self.cache["whoami"] return self.cache["whoami"]