动态分配和销毁互斥锁​​？

Question

我有一个基于 Eventlet 构建的应用程序。

我正在尝试编写一个像样的装饰器来同步跨线程对某些方法的调用。

装饰器目前看起来像这样：

_semaphores_semaphore = semaphore.Semaphore()
_semaphores = {}

def synchronized(name):
    def wrap(f):
        def inner(*args, **kwargs):
            # Grab the lock protecting _semaphores.
            with _semaphores_semaphore:
                # If the named semaphore does not yet exist, create it.
                if name not in _semaphores:
                    _semaphores[name] = semaphore.Semaphore()
                sem = _semaphores[name]

            with sem:
                return f(*args, **kwargs)

这工作正常，对我来说看起来不错并且线程安全，尽管整个线程安全和锁定业务对我来说可能有点生疏。

问题在于，应用程序中其他地方对信号量的特定现有使用（我想使用此装饰器将其转换为该信号量）会动态创建这些信号量：基于用户输入，它必须创建一个文件。它在字典中检查是否已有该文件的信号量，如果没有，则创建一个信号量并锁定它。一旦完成并释放了锁，它就会检查它是否再次被锁定（同时被另一个进程锁定），如果没有，它就会删除信号量。该代码是在绿色线程的假设下编写的，并且在该上下文中是安全的，但如果我想将其转换为使用我的装饰器，这是我无法解决的。

如果我不关心清理可能永远不会再次使用的信号量（可能有数十万个），我很好。如果我确实想清理它们，我不知道该怎么做。

要删除信号量，显然我需要持有 _semaphores_semaphore，因为我正在操纵 _semaphores 字典，但我也必须对特定信号量做一些事情，但我能想到的一切似乎都很有趣： * 在“with sem:”块内，我可以从 _semaphores 获取 _semaphores_semaphore 和 sem。然而，其他线程可能会被阻塞等待（在“with sem:”），如果一个新线程出现想要接触相同的资源，它不会在 _semaphores 中找到相同的信号量，而是创建一个新的信号量 = >失败。 我可以通过检查 sem 的平衡来稍微改进这一点，看看是否有另一个线程已经在等待我释放它。如果有，请保留它，如果没有，请删除它。这样，最后一个等待对资源执行操作的线程将删除它。但是，如果线程刚刚离开“with _semaphores_semaphore：”块，但尚未到达“with sem：”，我会遇到与之前相同的问题=>失败。

感觉就像我错过了一些明显的东西，但我无法弄清楚它是什么。

Answer 1

我想你也许可以用读写锁来解决这个问题。 _semaphores 字典上的共享独占锁。
这是未经测试的代码，用于展示原理。 RWLock 实现可以在例如 http: //code.activestate.com/recipes/413393-multiple-reader-one-writer-mrow-resource-locking/

_semaphores_rwlock = RWLock()
_semaphores = {}

def synchronized(name):
    def wrap(f):
        def inner(*args, **kwargs):
            lock = _semaphores_rwlock.reader()
            # If the named semaphore does not yet exist, create it.
            if name not in _semaphores:
                lock = _semaphores_rwlock.writer()
                _semaphores[name] = semaphore.Semaphore()

            sem = _semaphores[name]

            with sem:
                retval = f(*args, **kwargs)
            lock.release()
            return retval

当你想清理时，你可以这样做：

wlock = _semaphores_rwlock.writer() #this might take a while; it waits for all readers to release
cleanup(_semaphores)
wlock.release()

Answer 2

mchro 的答案对我不起作用，因为每当一个线程需要创建新信号量时，它就会阻止单个信号量上的所有线程。

我想出的答案是使用 _semaphores 保留两个事务之间的占用者计数器（这两个事务都是在同一个互斥体后面完成的）：

A: get semaphore
A1: dangerzone
B: with sem: block etc
C: cleanup semaphore

问题是知道 A 之间有多少人 和 C。信号量的计数器不会告诉您这一点，因为有人可能在 A1 中。答案是在 _semaphores 中为每个信号量保留一个进入者计数器，在 A 处递增它，在 C 处递减它，如果是如果为 0，则您知道 A-C 中没有其他人具有相同的密钥，您可以安全地删除它。