如何实现优雅断言，确保函数未被多线程同时调用？

Great question! Making sure a function isn't invoked concurrently by multiple threads is a common concurrency safety check, and there are several clean, maintainable ways to add assertions for this. Below are my go-to approaches, tailored for different scenarios:

1. Thread-Local Storage (TLS) for Per-Thread State Tracking #

This approach uses thread-local storage to track whether the current thread is already executing the function. It's lightweight since it avoids global synchronization overhead.

Example (Python): #

import threading

_thread_local = threading.local()

def my_function():
    # Check if we're already in this function on the current thread
    assert not hasattr(_thread_local, 'in_function'), \
        f"Recursive/concurrent call to {my_function.__name__} detected!"
    
    # Mark the thread as executing the function
    _thread_local.in_function = True
    try:
        # Your function logic here
        print("Executing my_function...")
    finally:
        # Clean up the thread-local state
        del _thread_local.in_function

Pros & Cons: #

• ✅ No global locks or atomic operations needed
• ✅ Works well for detecting recursive calls on the same thread
• ❌ Won't catch cross-thread concurrent calls (since TLS is per-thread)
• ❌ Requires cleanup in finally to avoid state leakage if exceptions are thrown

2. Atomic Flag for Global Concurrency Checks #

If you need to block cross-thread concurrent calls, an atomic boolean flag ensures thread-safe state checks and updates without heavy locks.

Example (C++): #

#include <atomic>
#include <cassert>

std::atomic<bool> is_executing{false};

void my_function() {
    // Attempt to set the flag to true atomically
    bool expected = false;
    assert(is_executing.compare_exchange_strong(expected, true) && 
           "Concurrent call to my_function detected!");
    
    try {
        // Function logic here
        // ...
    } finally {
        // Reset the flag when done
        is_executing = false;
    }
}

Pros & Cons: #

• ✅ Catches both cross-thread concurrent calls and same-thread recursion
• ✅ Lightweight atomic operations are faster than full locks for simple checks
• ❌ Must use finally to reset the flag—if an exception is thrown without cleanup, the function will be permanently blocked
• ❌ Not suitable for recursive calls unless you track a call count instead of a boolean

3. Lock-Based Assertions (For Already Thread-Safe Functions) #

If your function already uses a lock to enforce thread safety, you can extend that lock to add an assertion that checks if the current thread holds the lock (preventing accidental concurrent calls).

Example (Java): #

import java.util.concurrent.locks.ReentrantLock;

public class MyClass {
    private final ReentrantLock lock = new ReentrantLock();

    public void myFunction() {
        // Assert that the lock is NOT held by any thread (before acquiring)
        assert !lock.isLocked() : "Concurrent call to myFunction detected!";
        
        lock.lock();
        try {
            // Function logic here
        } finally {
            lock.unlock();
        }
    }
}

Pros & Cons: #

• ✅ Reuses existing thread-safety infrastructure—no extra state to manage
• ✅ Works seamlessly with functions that already use locks
• ❌ Only useful if you're already using a lock; adds no value for lock-free functions
• ❌ For recursive functions, use a ReentrantLock and adjust the assertion to check if the current thread holds the lock

4. Decorator/Annotation Wrappers (Clean Reusable Code) #

Wrap your assertion logic in a decorator (Python) or annotation (Java) to keep your function code clean and reuse the check across multiple functions.

Example (Python Decorator): #

import threading
from functools import wraps

def single_threaded_only(func):
    # Use a reentrant lock to allow recursive calls (if needed)
    lock = threading.RLock()
    
    @wraps(func)
    def wrapper(*args, **kwargs):
        # Assert that the lock isn't held by another thread
        assert not lock.locked() or lock._is_owned(), \
            f"Concurrent call to {func.__name__} detected!"
        
        with lock:
            return func(*args, **kwargs)
    return wrapper

@single_threaded_only
def my_function():
    # Function logic here
    print("Running my_function safely...")

@single_threaded_only
def another_protected_function():
    # Reuse the same assertion logic
    # ...

Pros & Cons: #

• ✅ Clean, declarative syntax—your function code stays focused on logic
• ✅ Easily reusable across multiple functions
• ✅ Can be configured to allow/disallow recursion via lock type
• ❌ Slight overhead from the wrapper, but negligible in debug mode

Key Notes: #

• Assertions are for debugging only: Remember that assertions are typically disabled in production builds (e.g., with NDEBUG in C++ or -O in Python). Never rely on them as the primary enforcement of thread safety—use proper locks/atomic operations for production.
• Handle recursion: If your function might call itself recursively, use a call count (in TLS) or a reentrant lock instead of a simple boolean flag.
• Async functions: For async/await code, adapt these approaches to use async-aware primitives (e.g., asyncio.Lock in Python instead of threading.Lock).

内容的提问来源于stack exchange，提问作者Brans Ds