同一线程嵌套threading.Lock引发死锁，求原因排查与解决

我编写了一个Counters类，用于封装Counter对象组成的字典。以下是简化后的示例代码（包含调试打印）：

import threading
import time
import logging
import random

logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-9s) %(message)s',)
                    
class Counter(object):
    
    def __init__(self, start : int = 0):
        self.lock = threading.Lock()
        self.value = start
        
    def increment(self):
        logging.debug('Waiting inc - %s', threading.current_thread().name)
        self.lock.acquire()
        try:
            logging.debug('Acquired inc - %s', threading.current_thread().name)
            self.value = self.value + 1
        finally:
            logging.debug('Released inc - %s', threading.current_thread().name)
            self.lock.release()
            
    def lastValue(self) -> str:
        logging.debug('Waiting lastValue - %s', threading.current_thread().name)
        self.lock.acquire()
        try:
            # return the last seen time in mysql format:
            logging.debug('Acquired lastValue - %s', threading.current_thread().name)
            return f" value -> {self.value}"
        except Exception as e:
            logging.error(e)
        finally:
            logging.debug('Released lastValue - %s', threading.current_thread().name)
            self.lock.release()
            
    def getAsDict(self, with_log=False) -> dict:
        logging.debug('Waiting getAsDict - %s', threading.current_thread().name)
        self.lock.acquire()
        try:
            logging.debug('Acquired getAsDict - %s', threading.current_thread().name)
            return {
                "counted" : self.lastValue(),
            }
        except Exception as e:
            logging.error(e)
        finally:
            logging.debug('Released getAsDict - %s', threading.current_thread().name)
            self.lock.release()

class Counters:
    
    def __init__(self,):
        self.lock = threading.Lock()
        self.store = {}
        
    def add(self, name : str) -> None:
        # add a counter object to the list:
        logging.debug('Waiting add - %s', threading.current_thread().name)
        self.lock.acquire()
        try:
            logging.debug('Acquired add - %s', threading.current_thread().name)
            self.store[name] = Counter(0)
        finally:
            logging.debug('Released add - %s', threading.current_thread().name)
            self.lock.release()
                    
    def remove(self, name : str) -> bool:
        # remove a counter from the dictionary:
        with self.lock:
            if name in self.store:
                del self.store[name]
                return True
            return False
    
    def get(self, name) -> Counter or None:
        with self.lock:
            return self.store.get(name, None)
    
    def getAll(self) -> dict:
        logging.debug('Waiting getAll - %s', threading.current_thread().name)
        self.lock.acquire()
        try:
            logging.debug('Acquired getAll - %s', threading.current_thread().name)
            ret = {}
            for name, counter in self.store.items():
                print(counter.getAsDict())
                ret[name] = counter.getAsDict()
            return ret
        except Exception as e:
            print(e)
        finally:
            logging.debug('Released getAll - %s', threading.current_thread().name)
            self.lock.release()

调用getAll()方法时，程序会在lastValue()处陷入死锁，调用路径如下：

• getAll（获取Counters的锁）
• getAsDict（获取Counter的锁）
• lastValue（尝试获取Counter的锁）—— 死锁发生在此处

该问题在单线程运行时也可复现，测试代码如下：

def worker(c):
    for i in range(1):
        r = random.random()
        n = random.randint(1, 500)
        #random name:
        name = f"counter_{n}"
        logging.debug('Counter [%s] Sleeping %0.02f', name, r)
        time.sleep(r)
        c.add(name)
        c.get(name).increment()
    logging.debug('Done')
    result = c.getAll()
    logging.debug('Result: %r', result)

if __name__ == '__main__':
    counters = Counters()
    for i in range(1):
        t = threading.Thread(target=worker, args=(counters,))
        t.start()

    logging.debug('Waiting for worker threads')
    main_thread = threading.current_thread()
    for t in threading.enumerate():
        if t is not main_thread:
            t.join()

运行输出如下：

(MainThread) Waiting for worker threads
(Thread-7 (worker)) Counter [counter_129] Sleeping 0.55
(Thread-7 (worker)) Waiting add - Thread-7 (worker)
(Thread-7 (worker)) Acquired add - Thread-7 (worker)
(Thread-7 (worker)) Released add - Thread-7 (worker)
(Thread-7 (worker)) Waiting inc - Thread-7 (worker)
(Thread-7 (worker)) Acquired inc - Thread-7 (worker)
(Thread-7 (worker)) Released inc - Thread-7 (worker)
(Thread-7 (worker)) Done
(Thread-7 (worker)) Waiting getAll - Thread-7 (worker)
(Thread-7 (worker)) Acquired getAll - Thread-7 (worker)
(Thread-7 (worker)) Waiting getAsDict - Thread-7 (worker)
(Thread-7 (worker)) Acquired getAsDict - Thread-7 (worker)
(Thread-7 (worker)) Waiting lastValue - Thread-7 (worker) <-- DEADLOCK

请问我忽略了什么问题？

你混淆了Python中两种锁的核心特性：threading.Lock不支持可重入，只有threading.RLock（可重入锁）允许同一线程多次获取同一把锁。

死锁的根源非常明确：

1. 当前线程调用getAsDict时，已经获取了Counter的Lock
2. 接着getAsDict内部调用lastValue，又尝试获取同一把Lock
3. 由于Lock不可重入，线程会阻塞在这里等待自己释放锁，最终形成死锁

修复方案 #

方案1：替换为可重入锁 #

把Counter类中的self.lock = threading.Lock()替换为self.lock = threading.RLock()即可解决问题。RLock会跟踪锁的持有次数，同一线程每次acquire都会增加计数，release则减少计数，只有当计数归零时，锁才会被真正释放。

方案2：优化锁逻辑（更高效） #

既然getAsDict已经持有了锁，内部调用lastValue时不需要再次加锁。可以拆分出一个无锁的内部方法，避免重复的锁操作：

class Counter(object):
    def __init__(self, start : int = 0):
        self.lock = threading.Lock()
        self.value = start
        
    def increment(self):
        logging.debug('Waiting inc - %s', threading.current_thread().name)
        with self.lock:
            logging.debug('Acquired inc - %s', threading.current_thread().name)
            self.value = self.value + 1
        logging.debug('Released inc - %s', threading.current_thread().name)
            
    def _lastValue(self) -> str:
        # 内部无锁版本，仅在已持有锁时调用
        return f" value -> {self.value}"
    
    def lastValue(self) -> str:
        logging.debug('Waiting lastValue - %s', threading.current_thread().name)
        with self.lock:
            logging.debug('Acquired lastValue - %s', threading.current_thread().name)
            return self._lastValue()
        logging.debug('Released lastValue - %s', threading.current_thread().name)
            
    def getAsDict(self, with_log=False) -> dict:
        logging.debug('Waiting getAsDict - %s', threading.current_thread().name)
        with self.lock:
            logging.debug('Acquired getAsDict - %s', threading.current_thread().name)
            return {
                "counted" : self._lastValue(),
            }
        logging.debug('Released getAsDict - %s', threading.current_thread().name)

这种方式既避免了对可重入锁的依赖，又减少了不必要的锁操作，代码执行效率更高。

内容的提问来源于stack exchange，提问作者Shlomi Hassid