The Challenge: Memory Leaks vs. Concurrent Race Conditions

When building asynchronous Python applications (such as API rate limiters, payment processors, or chat services), you often need to serialize operations for a specific entity—such as a user_id. A common pattern is using an in-memory dictionary of asyncio.Lock() instances to prevent concurrent operations for the same key.

However, if your user base grows over time, storing locks indefinitely in a dictionary causes an in-memory leak. A tempting shortcut is deleting the lock from the dictionary as soon as it is acquired:

async with lock:
    if user_id in locks:
        del locks[user_id]  # Danger zone!
    await send_request(user_id)

Why Naive Lock Deletion Breaks Concurrency

Deleting the lock from the dictionary while tasks are still using it introduces a critical race condition:

  1. Task A acquires the lock for user_id = 42 and immediately deletes the lock entry from locks.
  2. While Task A is still running await send_request(42), Task B arrives for the same user.
  3. Task B executes locks.setdefault(42, asyncio.Lock()). Since locks[42] was deleted, a brand new lock is created and stored.
  4. Task B acquires this new lock immediately because no other task is waiting on the new lock object.
  5. Both tasks now run concurrently for the same user, completely defeating the purpose of synchronization.

The Solution: Reference-Counted Keyed Locks

Because Python's asyncio runs on a single-threaded event loop, all code without an await expression executes synchronously and atomically. We can leverage this behavior by tracking active references (both tasks currently holding the lock and tasks waiting in line for it).

A lock should only be deleted from memory when its reference count drops back to zero.

Implementing an Async Lock Manager

Using contextlib.asynccontextmanager, you can build an elegant, self-cleaning lock manager:

import asyncio
from contextlib import asynccontextmanager

class KeyedLockManager:
    def __init__(self):
        # Maps key -> [asyncio.Lock, active_waiters_count]
        self._locks = {}

    @asynccontextmanager
    async def lock(self, key):
        # Synchronous: cannot be interrupted by another coroutine
        if key not in self._locks:
            self._locks[key] = [asyncio.Lock(), 0]
        
        entry = self._locks[key]
        entry[1] += 1  # Increment reference count
        lock = entry[0]

        try:
            async with lock:
                yield
        finally:
            # Synchronous cleanup
            entry[1] -= 1  # Decrement reference count
            if entry[1] == 0 and self._locks.get(key) is entry:
                del self._locks[key]

How It Works

  • Atomic Reference Increment: When a task requests a lock, it registers its interest immediately by bumping the counter. Even if it gets suspended while waiting to acquire the lock (async with lock:), the lock remains safely in the dictionary.
  • Safe Garbage Collection: In the finally block, the counter decrements. If no other coroutine is waiting or executing (counter reaches 0), the dictionary entry is safely removed.
  • No Stray References: The identity check (self._locks.get(key) is entry) guarantees that another task hasn't replaced the lock instance under edge cases.

Complete Working Example

Let's apply this lock manager to the original problem:

import asyncio
from contextlib import asynccontextmanager

class KeyedLockManager:
    def __init__(self):
        self._locks = {}

    @asynccontextmanager
    async def lock(self, key):
        if key not in self._locks:
            self._locks[key] = [asyncio.Lock(), 0]
        
        entry = self._locks[key]
        entry[1] += 1
        lock = entry[0]

        try:
            async with lock:
                yield
        finally:
            entry[1] -= 1
            if entry[1] == 0 and self._locks.get(key) is entry:
                del self._locks[key]

lock_manager = KeyedLockManager()

async def send_request(user_id):
    print(f"Starting request for {user_id}")
    await asyncio.sleep(0.5)
    print(f"Finished request for {user_id}")

async def process(user_id):
    async with lock_manager.lock(user_id):
        await send_request(user_id)

async def main():
    first = asyncio.create_task(process(42))
    await asyncio.sleep(0.01)
    second = asyncio.create_task(process(42))
    
    await asyncio.gather(first, second)
    print(f"Remaining locks in memory: {len(lock_manager._locks)}")

asyncio.run(main())

Output

Starting request for 42
Finished request for 42
Starting request for 42
Finished request for 42
Remaining locks in memory: 0

Summary

Never delete an asyncio.Lock from a shared dictionary while code inside the lock is awaiting execution. By using a lightweight reference counter, you get the best of both worlds: strict sequential processing for active keys and automatic memory cleanup when idle.