如何用Python编写Hybrid Mutex（混合互斥锁）？

Great question! Hybrid mutexes are perfect for balancing low latency (via spinning for short waits) and efficient CPU usage (via blocking when waits get long)—exactly what you're looking for. Since Python's standard threading.Lock only blocks by default, we can wrap it with a spin-first layer to create our hybrid implementation, while keeping things Pythonic (think context managers, standard library integration, etc.).

Core Idea #

The logic is straightforward:

1. When trying to acquire the lock, first spin (repeatedly check if the lock is available without blocking) for a short, predefined timeout.
2. If the lock becomes available during the spin phase, grab it immediately.
3. If the spin timeout expires, fall back to the standard blocking acquire to avoid wasting CPU cycles.

Pythonic Implementation #

Here's a clean, thread-safe implementation using Python's threading module and precise timing:

import threading
import time

class HybridMutex:
    def __init__(self, spin_timeout: float = 0.001):
        """
        Initialize a hybrid spin mutex.
        
        Args:
            spin_timeout: Time to spin (in seconds) before falling back to blocking.
                          Defaults to 1ms, adjust based on your workload.
        """
        self._lock = threading.Lock()
        self.spin_timeout = spin_timeout

    def acquire(self, blocking: bool = True, timeout: float | None = None) -> bool:
        """
        Acquire the lock, with optional spin-then-block behavior.
        
        Args:
            blocking: If False, return immediately without spinning or blocking.
            timeout: Total maximum time (in seconds) to wait for the lock (spin + block).
        
        Returns:
            True if the lock was acquired, False otherwise.
        """
        # Handle non-blocking case directly
        if not blocking:
            return self._lock.acquire(blocking=False)

        # Spin phase: Try non-blocking acquire until timeout or success
        start_time = time.perf_counter()
        while time.perf_counter() - start_time < self.spin_timeout:
            if self._lock.acquire(blocking=False):
                return True
            # Yield GIL to let other threads run (critical for Python's GIL model)
            threading.yield()

        # Spin timed out: Fall back to blocking acquire, adjusting for elapsed time
        elapsed = time.perf_counter() - start_time
        if timeout is not None:
            timeout -= elapsed
            if timeout <= 0:
                return False
        
        return self._lock.acquire(blocking=True, timeout=timeout)

    def release(self) -> None:
        """Release the lock (forwarded to the underlying lock)."""
        self._lock.release()

    # Support context manager syntax (Pythonic way to use locks)
    def __enter__(self) -> 'HybridMutex':
        self.acquire()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        self.release()

Key Details & Notes #

Let's break down the important parts:

• GIL Consideration: Unlike C, Python has a Global Interpreter Lock (GIL) that limits concurrent execution of Python bytecode. The threading.yield() call in the spin loop is critical—it releases the GIL, giving the thread holding the lock a chance to run and release it. Without this, spinning would just hog the GIL and prevent progress.
• Timeout Handling: The acquire method respects both the spin timeout and the user-provided total timeout. We adjust the remaining timeout after spinning to ensure we don't exceed the user's requested total wait time.
• Context Manager Support: Implementing __enter__ and __exit__ lets you use the mutex with with statements, which is the cleanest way to handle locks in Python (automatically releases the lock even if an exception occurs).
• Adjustable Spin Time: The spin_timeout parameter should be tuned to your workload. For short, frequent critical sections, use a smaller value (e.g., 0.5ms). For workloads where locks might be held longer, increase it slightly—but don't go too high, or you'll waste CPU on unnecessary spinning.

Usage Example #

Here's how to use the HybridMutex in a typical multi-threaded scenario:

def critical_task(mutex: HybridMutex, thread_id: int) -> None:
    with mutex:
        print(f"Thread {thread_id} acquired the lock")
        # Simulate a short critical section (adjust sleep time to test spin vs block)
        time.sleep(0.001)
        print(f"Thread {thread_id} released the lock")

# Initialize mutex with 2ms spin timeout
mutex = HybridMutex(spin_timeout=0.002)

# Create and start 5 threads
threads = []
for i in range(5):
    t = threading.Thread(target=critical_task, args=(mutex, i))
    threads.append(t)
    t.start()

# Wait for all threads to finish
for t in threads:
    t.join()

Bonus: Reentrant Variant #

If you need a reentrant hybrid mutex (where the same thread can acquire the lock multiple times), simply replace threading.Lock with threading.RLock in the __init__ method:

self._lock = threading.RLock()

内容的提问来源于stack exchange，提问作者Arohi Gupta