Distributed Locking for Background Jobs
In this tutorial, you will learn about Distributed Locking for Background Jobs. We cover key concepts, practical examples, and best practices to help you master this topic.
Implement distributed locking with Redis and etcd to prevent duplicate job execution, coordinate worker access, and ensure exactly-once processing across nodes.
What You Learn
You will learn how to implement distributed locks with Redis, etcd, and database advisory locks, handle lock expiration and renewal, and use locks for job coordination.
Why It Matters
Without distributed locks, multiple workers can Process the same job simultaneously, causing duplicates and race conditions. Locks ensure only one worker handles a job at a time.
Real-World Use
DodaTech uses Redis Redlock for distributed locking of recurring jobs. When a job runs on a schedule, the first worker to acquire the lock processes it. Other workers skip if they cannot acquire the lock.
Redis Distributed Lock
import redis
import time
import uuid
import threading
r = redis.Redis()
class RedisLock:
def __init__(self, lock_name, ttl=30):
self.lock_name = f'lock:{lock_name}'
self.ttl = ttl
self.lock_value = str(uuid.uuid4())
def acquire(self, blocking=False, timeout=10):
if blocking:
deadline = time.time() + timeout
while time.time() < deadline:
if r.setnx(self.lock_name, self.lock_value):
r.expire(self.lock_name, self.ttl)
return True
time.sleep(0.1)
return False
else:
acquired = r.setnx(self.lock_name, self.lock_value)
if acquired:
r.expire(self.lock_name, self.ttl)
return acquired
def release(self):
lua_script = """
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end
"""
r.eval(lua_script, 1, self.lock_name, self.lock_value)
def __enter__(self):
self.acquire(blocking=True)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
def critical_section(worker_id):
with RedisLock('job-processor', ttl=10) as lock:
print(f"[{worker_id}] Acquired lock, processing...")
time.sleep(2)
print(f"[{worker_id}] Releasing lock")
print(f"[{worker_id}] Lock released")
t1 = threading.Thread(target=critical_section, args=('worker-1',), daemon=True)
t2 = threading.Thread(target=critical_section, args=('worker-2',), daemon=True)
t1.start()
t2.start()
t1.join(timeout=5)
t2.join(timeout=5)
Expected output:
[worker-1] Acquired lock, processing...
[worker-2] Acquired lock, processing...
[worker-1] Releasing lock
[worker-1] Lock released
[worker-2] Releasing lock
[worker-2] Lock released
Lock with Auto-Renewal
import redis
import time
import uuid
import threading
r = redis.Redis()
class AutoRenewingLock:
def __init__(self, name, ttl=10, renew_interval=5):
self.name = f'arlock:{name}'
self.ttl = ttl
self.renew_interval = renew_interval
self.value = str(uuid.uuid4())
self._renewing = False
self._owned = False
def acquire(self, timeout=10):
deadline = time.time() + timeout
while time.time() < deadline:
if r.setnx(self.name, self.value):
r.expire(self.name, self.ttl)
self._owned = True
self._start_renewal()
return True
time.sleep(0.1)
return False
def _start_renewal(self):
self._renewing = True
t = threading.Thread(target=self._renew_loop, daemon=True)
t.start()
def _renew_loop(self):
while self._renewing:
time.sleep(self.renew_interval)
if self._owned:
lua = """
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('expire', KEYS[1], ARGV[2])
end
return 0
"""
r.eval(lua, 1, self.name, self.value, self.ttl)
def release(self):
self._renewing = False
self._owned = False
lua = """
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
end
return 0
"""
r.eval(lua, 1, self.name, self.value)
def long_job(worker):
lock = AutoRenewingLock('long-job', ttl=10, renew_interval=5)
if lock.acquire(timeout=5):
print(f"[{worker}] Locked, working...")
time.sleep(15)
lock.release()
print(f"[{worker}] Done")
else:
print(f"[{worker}] Could not acquire lock")
t1 = threading.Thread(target=long_job, args=('w1',), daemon=True)
t2 = threading.Thread(target=long_job, args=('w2',), daemon=True)
t1.start()
t2.start()
t1.join(timeout=20)
t2.join(timeout=20)
Expected output:
[w1] Locked, working...
[w2] Could not acquire lock
[w1] Done
Database Advisory Lock
import sqlite3
import time
import threading
class AdvisoryLock:
def __init__(self, db_path=':memory:'):
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.conn.execute('''
CREATE TABLE IF NOT EXISTS advisory_locks (
lock_name TEXT PRIMARY KEY,
holder TEXT,
acquired_at REAL,
ttl REAL
)
''')
self.conn.commit()
self._lock = threading.Lock()
def acquire(self, lock_name, holder, ttl=30):
with self._lock:
now = time.time()
cursor = self.conn.execute(
'SELECT holder, acquired_at, ttl FROM advisory_locks WHERE lock_name = ?',
(lock_name,)
)
row = cursor.fetchone()
if row:
_, acquired_at, lock_ttl = row
if now - acquired_at < lock_ttl:
return False
self.conn.execute(
'DELETE FROM advisory_locks WHERE lock_name = ?',
(lock_name,)
)
self.conn.execute(
'INSERT OR REPLACE INTO advisory_locks (lock_name, holder, acquired_at, ttl) VALUES (?, ?, ?, ?)',
(lock_name, holder, now, ttl)
)
self.conn.commit()
return True
def release(self, lock_name, holder):
with self._lock:
self.conn.execute(
'DELETE FROM advisory_locks WHERE lock_name = ? AND holder = ?',
(lock_name, holder)
)
self.conn.commit()
def is_locked(self, lock_name):
cursor = self.conn.execute(
'SELECT holder, acquired_at, ttl FROM advisory_locks WHERE lock_name = ?',
(lock_name,)
)
row = cursor.fetchone()
if row:
_, acquired_at, ttl = row
if time.time() - acquired_at < ttl:
return True
return False
db_lock = AdvisoryLock()
def worker_task(wid, lock_name):
if db_lock.acquire(lock_name, wid, ttl=10):
print(f"[{wid}] Acquired {lock_name}")
time.sleep(2)
db_lock.release(lock_name, wid)
print(f"[{wid}] Released {lock_name}")
else:
print(f"[{wid}] Could not acquire {lock_name}")
threads = [
threading.Thread(target=worker_task, args=('w1', 'cleanup'), daemon=True),
threading.Thread(target=worker_task, args=('w2', 'cleanup'), daemon=True),
]
for t in threads:
t.start()
for t in threads:
t.join(timeout=5)
Expected output:
[w1] Acquired cleanup
[w2] Could not acquire cleanup
[w1] Released cleanup
Lock for Scheduled Jobs
import time
import redis
import uuid
r = redis.Redis()
class ScheduledJobLock:
def __init__(self):
self.lock_prefix = 'scheduled_lock'
def try_run(self, job_name, func, ttl=60):
lock_key = f'{self.lock_prefix}:{job_name}'
lock_value = str(uuid.uuid4())
acquired = r.setnx(lock_key, lock_value)
if not acquired:
print(f"[{job_name}] Skipped, another worker holds lock")
return False
r.expire(lock_key, ttl)
try:
print(f"[{job_name}] Executing")
func()
return True
finally:
lua = """
if redis.call('get', KEYS[1]) == ARGV[1] then
redis.call('del', KEYS[1])
end
"""
r.eval(lua, 1, lock_key, lock_value)
def daily_backup():
time.sleep(0.5)
print(" Backup completed")
lock = ScheduledJobLock()
lock.try_run('daily_backup', daily_backup)
lock.try_run('daily_backup', daily_backup)
Expected output:
[daily_backup] Executing
Backup completed
[daily_backup] Skipped, another worker holds lock
Common Mistakes
1. Not Using Unique Lock Values
Without unique values, a lock acquired by worker A can be released by worker B. Always use unique values for safe release.
2. No Lock Expiration
If a worker crashes while holding a lock, the lock is held forever. Always set TTL on locks.
3. Holding Locks During I/O
Long-held locks block other workers. Hold locks only for critical sections, not during slow I/O operations.
4. Clock Drift in Lock Expiry
Relying on local time for lock expiration causes issues with clock drift. Use Redis TTL which is server-side.
5. Single Point of Failure
A single Redis instance for locking is a SPOF. Use Redis Sentinel or Redlock with multiple nodes for HA.
Practice Questions
1. Why use distributed locks in job processing?
Prevent multiple workers from processing the same job, coordinate access to shared resources, and ensure exactly-once execution.
2. How does Redis SETNX implement locking?
SETNX sets a key only if it does not exist. If successful, the lock is acquired. If the key exists, another worker holds the lock.
3. What is lock renewal and why is it needed?
Jobs that run longer than the lock TTL need to renew the lock periodically. Without renewal, the lock expires and another worker acquires it.
4. How do you safely release a distributed lock?
Use a Lua script that checks the lock value before deleting. This prevents releasing a lock held by another worker.
Challenge
Build a distributed locking system for job coordination: Redis SETNX with unique values, TTL with auto-renewal, Lua script for safe release, fallback to database advisory locks, and monitoring for lock contention.
FAQ
Mini Project: Distributed Lock
import redis
import time
import uuid
r = redis.Redis()
class DistributedLock:
def __init__(self, name, ttl=30):
self.name = f'dlock:{name}'
self.ttl = ttl
self.val = str(uuid.uuid4())
def acquire(self, timeout=5):
deadline = time.time() + timeout
while time.time() < deadline:
if r.setnx(self.name, self.val):
r.expire(self.name, self.ttl)
return True
time.sleep(0.05)
return False
def release(self):
lua = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end"
r.eval(lua, 1, self.name, self.val)
def __enter__(self):
self.acquire()
return self
def __exit__(self, *args):
self.release()
lock = DistributedLock('backup', ttl=10)
with lock:
print("Critical section (protected by distributed lock)")
print("Lock released")
Expected output:
Critical section (protected by distributed lock)
Lock released
What's Next
Now that you understand distributed locking, explore database-based locking for RDBMS-based coordination, then learn about cron timezone handling.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro