Distributed Locking in Python

When making use of Python's asyncio library, synchronizing code within a single process is a solved problem. A simple asyncio.Lock() ensures that coroutines play nice together. But as soon as your application scales to multiple containers, servers, or microservices, that local lock becomes invisible to the rest of your fleet.

You suddenly face the challenge of Distributed Mutual Exclusion: ensuring that a critical section of code is only executed by one worker at a time, regardless of which machine it is running on.

In this post, we’ll explore a robust distributed lock implementation using ThingsDB. This solution provides the familiar syntax of an asyncio context manager but leverages a globally synchronized backend to manage state across your entire infrastructure.

Why Distributed Locking is a "Hard" Problem

On the surface, it sounds easy: just set a flag in a database. But in a distributed environment, the stakes are higher. If a worker acquires a lock and then crashes or loses network connectivity, that lock could remain "held" in the database forever, creating a permanent deadlock.

ThingsDB solves this by introducing a server-side timeout. If the client holding the lock fails to release it or disappears, ThingsDB automatically releases the lock for the next person in line.

Unlike traditional distributed locks that require "polling" the database, ThingsDB locks are event-driven. They leverage ThingsDB’s emit event system, meaning there is zero wasted overhead. The moment a lock is released, an event triggers the next waiter in line for immediate execution. This ensures maximum throughput while strictly guaranteeing that only one worker is active at a time.

Part 1: The Infrastructure Setup

Before your application can start locking resources, ThingsDB needs to prepare the underlying collection logic. This is an idempotent operation—it only does work the first time it is called.

In a production environment, you should perform this setup during your application's bootstrap phase.

from thingsdb.misc import lock

async def initialize_infrastructure(client):
    # Connect and authenticate
    await client.connect('localhost')
    await client.authenticate('admin', 'pass')

    # This prepares the lock collection in ThingsDB.
    # It’s safe to call every time the service starts.
    await lock.setup(client)

Part 2: The "Initialize Once" Pattern

In a real-world project, you want to avoid creating new lock objects inside your high-frequency loops. Instead, initialize your lock configuration once and reuse that object throughout the lifecycle of your application.

By using functools.partial, we can create a "Lock Factory" that is pre-configured with the correct client, resource name, and timeout settings.

from functools import partial
from thingsdb.misc import lock

# This will hold our pre-configured lock factory
invoice_lock = None

async def setup_application_locks(client):
    global invoice_lock

    # Perform the one-time backend setup
    await lock.setup(client)

    # Initialize the lock object once.
    # If a worker crashes, the lock is freed after 30 seconds.
    invoice_lock = partial(lock.lock,
                           client=client,
                           name='process-invoices',
                           timeout=30)

Note: You can create as many different locks as your application needs. Just ensure that each unique resource or critical section is given its own unique name to avoid unintended interference between different parts of your system.

Part 3: Using the Lock in Your Business Logic

Because the implementation uses the async withsyntax, it integrates seamlessly into any asyncio application. It feels exactly like a local lock, but it’s protecting your data across the globe.

async def process_invoices():
    # Use the pre-configured lock we created during startup
    async with invoice_lock():
        print('Distributed lock acquired! '
              'No other service can enter this block.')

        # Simulate critical work like hitting a payment API
        await asyncio.sleep(5.0)

        print('Work finished. Lock is automatically released.')

Monitoring and Safety

Sometimes you need to know if a resource is busy without actually waiting in line. You can use the locked() function for a non-blocking check:

is_busy = await lock.locked(client, 'process-invoices')
if is_busy:
    print("Another worker is currently processing invoices.")

Key Takeaways

• Queue Management: The queue_size and timeout parameters allow you to control exactly how long a request should wait before giving up, preventing "zombie" requests from clogging your system.
• Resilience: The server-side timeout ensures that your system is self-healing even in the face of hardware failures or process crashes.
• Developer Experience: By mimicking the asyncio.Lock() interface, the learning curve for your team is practically zero.

Conclusion

Distributing your Python application shouldn't mean sacrificing the safety of mutual exclusion. By using ThingsDB to back your locks, you get a high-performance, self-healing synchronization layer that scales with your infrastructure.