python - 一旦没有线程请求它，垃圾收集锁

Question

我有一个函数，绝不能同时从两个线程调用相同的值。为了强制执行此操作，我有一个 defaultdict 为给定的 key 生成新的 threading.Lock。因此，我的代码看起来类似于:

from collections import defaultdict
import threading

lock_dict = defaultdict(threading.Lock)
def f(x):
    with lock_dict[x]:
        print "Locked for value x"

问题是我无法弄清楚如何安全地删除 defaultdict 不再需要的锁。如果不这样做，我的程序就会出现内存泄漏，当使用许多不同的 x 值调用 f 时，这种情况会变得很明显。

我不能简单地在 f 的末尾 del lock_dict[x]，因为在另一个线程正在等待锁的情况下，第二个线程将锁定一个不再关联的锁lock_dict[x]，因此两个线程可能最终同时调用具有相同 x 值的 f。

Answer 1

我会使用不同的方法:

fcond = threading.Condition()
fargs = set()

def f(x):
    with fcond:
        while x in fargs:
            fcond.wait()
        fargs.add(x)  # this thread has exclusive rights to use `x`

    # do useful stuff with x
    # any other thread trying to call f(x) will
    # block in the .wait above()

    with fcond:
        fargs.remove(x)      # we're done with x
        fcond.notify_all()   # let blocked threads (if any) proceed

条件有一个学习曲线，但一旦爬上它，它们就会更容易编写正确的线程安全、无竞争代码。

原始代码的线程安全

@JimMischel 在评论中询问原始人对 defaultdict 的使用是否受制于种族。好问题!

答案是 - 唉 - “你必须盯着你的特定 Python 实现”。

假设 CPython 实现:如果 任何 defaultdict 调用的代码提供默认调用 Python 代码，或释放 GIL(全局解释器锁)的 C 代码，然后 2 个(或更多)线程可以“同时”调用 withlock_dict[x] 并使用字典中尚未包含的相同 x，并且:

1. 线程 1 发现 x 不在字典中，获得一个锁，然后丢失它的时间片(在字典中设置 x 之前)。
2. 线程 2 发现 x 不在字典中，也获得了锁。
3. 其中一个线程的锁在字典中结束，但两个线程都执行 f(x)。

查看3.4.0a4+(当前开发负责人)的源码，defaultdict和threading.Lock都是C代码实现的，没有释放GIL .我不记得早期版本是否在不同时期在 Python 中实现了 defaultdict 或 threading.Lock 的全部或部分。

我建议的替代代码充满了用 Python 实现的东西(所有 threading.Condition 方法)，但在设计上是无竞争的——即使你使用的是带有集合的旧版本 Python也是用 Python 实现的(该集合只有在条件变量锁的保护下才能访问)。

每个参数一个锁

没有条件，这似乎要困难得多。在最初的方法中，我相信您需要对想要使用 x 的线程进行计数，并且您需要一个锁来保护这些计数并保护字典。我为此想出的最好的代码是如此冗长以至于将它放在上下文管理器中似乎是最明智的。要使用，请为每个需要它的函数创建一个参数锁:

farglocker = ArgLocker() # for function `f()`

然后 f() 的主体可以简单地编码:

def f(x):
    with farglocker(x):
        # only one thread at a time can run with argument `x`

当然，条件方法也可以包装在上下文管理器中。这是代码:

import threading

class ArgLocker:
    def __init__(self):
        self.xs = dict() # maps x to (lock, count) pair
        self.lock = threading.Lock()

    def __call__(self, x):
        return AllMine(self.xs, self.lock, x)

class AllMine:
    def __init__(self, xs, lock, x):
        self.xs = xs
        self.lock = lock
        self.x = x

    def __enter__(self):
        x = self.x
        with self.lock:
            xlock = self.xs.get(x)
            if xlock is None:
                xlock = threading.Lock()
                xlock.acquire()
                count = 0
            else:
                xlock, count = xlock
            self.xs[x] = xlock, count + 1

        if count: # x was already known - wait for it
            xlock.acquire()
        assert xlock.locked

    def __exit__(self, *args):
        x = self.x
        with self.lock:
            xlock, count = self.xs[x]
            assert xlock.locked
            assert count > 0
            count -= 1
            if count:
                self.xs[x] = xlock, count
            else:
                del self.xs[x]
            xlock.release()

那么哪种方式更好呢？使用条件 ;-) 这种方式“几乎显然是正确的”，但是每个参数锁定 (LPA) 方法有点让人头疼。 LPA 方法确实有一个优点，当一个线程用x 完成时，唯一 允许继续进行的线程是那些想要使用相同x 的线程>;使用条件，.notify_all() 唤醒阻塞等待any 参数的所有线程。但是除非在尝试使用相同参数的线程之间存在非常激烈的争用，否则这并不重要:使用条件，唤醒的线程不会等待 x 保持清醒足够长的时间看到 x in fargs 为真，然后立即再次阻塞 (.wait())。