oslo-concurrency

import os from oslo_concurrency import lockutils # For inter-process locking, a lock_path must be configured. # For this example, we'll use a temporary directory. # In a real application, this should be a secure, dedicated directory. lock_dir = os.environ.get('OSLO_LOCK_PATH', '/tmp/oslo_locks') os.makedirs(lock_dir, exist_ok=True) # Configure oslo.concurrency to use the lock path # This is typically done via oslo.config in a real OpenStack project, # but we can set it directly for a quick example. lockutils.set_defaults(lock_path=lock_dir) @lockutils.synchronized('my-resource', external=True) def my_locked_function(worker_id): print(f"Worker {worker_id} acquired the lock.") # Simulate some work import time time.sleep(0.5) print(f"Worker {worker_id} released the lock.") if __name__ == '__main__': # Example of calling the locked function from multiple 'processes' # In a real scenario, these would be separate processes. print("Attempting to call my_locked_function from multiple 'workers'...") for i in range(3): my_locked_function(f"SimulatedWorker-{i}") # Clean up the lock directory (optional for demo) try: os.rmdir(lock_dir) except OSError: # Directory might not be empty if locks were created pass