任意休眠线程等待超时

Question

在我开始描述我的问题之前，值得一提的是我使用的是 Python 2.7。我没有检查，但这可能与 Python 3.x 无关。

在使用 Python 的 Queues 时，我发现了一些奇怪的东西。通常，当从队列中获取一个对象时，我允许长时间但有限的超时（例如几秒钟），以便在没有找到对象的情况下进行调试和错误报告，而这是预期的。我发现有时在将对象插入到以前为空的队列中的时间与同一队列的 get 方法返回该对象的时间之间存在奇怪的差距，即使该方法是在为该对象调用 put 之前调用。

稍微挖掘一下，我发现这个空隙是通过睡觉来填补的。在Queue 模块中，如果传递给get 方法的timeout 参数不是None，并且是正数，则使用non_empty Condition 的wait 方法调用正参数（这不是 100% 精确的；事实上，Queue 的“_qsize”方法，它返回底层 deque 的长度首先验证返回 0，但只要队列是首先是空的，接下来就是条件的等待）。

Conditions 的wait 方法在是否超时时会有所不同。如果它没有得到任何超时，它只是调用waiter.acquire。这是在C 中定义的，超出了我的理解，但它似乎可以正常工作。但是，如果给定超时 ，则会发生奇怪的睡眠序列，此时睡眠时间以任意大小（1 毫秒）开始，并且随着时间的推移变得更长。这是运行的确切代码：

# Balancing act:  We can't afford a pure busy loop, so we
# have to sleep; but if we sleep the whole timeout time,
# we'll be unresponsive.  The scheme here sleeps very
# little at first, longer as time goes on, but never longer
# than 20 times per second (or the timeout time remaining).
endtime = _time() + timeout
delay = 0.0005 # 500 us -> initial delay of 1 ms
while True:
    gotit = waiter.acquire(0)
    if gotit:
        break
    remaining = endtime - _time()
    if remaining <= 0:
        break
    delay = min(delay * 2, remaining, .05)
    _sleep(delay)

这显然是我发现新对象放入先前空队列的时间与已调用的 get 方法返回该对象的时间之间存在差距的原因。随着延迟时间呈指数增长，直到被 0.05 秒的巨大（从我的角度）大小阻塞，它会在我的应用程序的生命周期中产生令人惊讶且不需要的重要睡眠。

你能解释一下这样做的目的是什么吗？ Python 开发人员是否假设没有 Python 用户会关心这样的时间长度？是否有快速解决方法或适当的修复方法？你建议我重载线程模块吗？

Answer 1

我最近遇到了同样的问题，我还追踪到了 threading 模块中的这段代码。

很烂。

---

你能解释一下这样做的目的是什么吗？ Python 开发人员是否假设没有 Python 用户会关心这样的时间长度？

打败我...

---

你建议我重载线程模块吗？

要么重载线程模块，要么迁移到python3，这部分实现已经修复。

在我的情况下，迁移到 python3 将是一项巨大的努力，所以我选择了前者。我所做的是：

1. 我创建了一个带有pthread 接口的快速.so 文件（使用cython）。它包括调用相应pthread_mutex_* 函数的python 函数，以及针对libpthread 的链接。具体来说，与我们感兴趣的任务最相关的函数是pthread_mutex_timedlock。
2. 我创建了一个新的threading2 模块，（并用import threading2 替换了我的代码库中的所有import threading 行）。在threading2 中，我重新定义了来自threading 的所有相关类（Lock、Condition、Event），以及来自Queue 的那些我经常使用的类（Queue 和@987654339 @)。 Lock 类完全使用 pthread_mutex_* 函数重新实现，但其余的要容易得多——我只是将原来的子类化（例如 threading.Event），并覆盖 __init__ 以创建我的新 Lock 类型。其余的工作正常。

新的Lock 类型的实现与threading 中的原始实现非常相似，但我基于python3 的threading 模块中找到的代码（当然，这比上面提到的“平衡行为”块要简单得多）。这部分相当简单。

（顺便说一句，在我的案例中，我的大规模多线程进程的速度提高了 30%。甚至超出了我的预期。）

我希望这会有所帮助。

Answer 2

您可以使用 get_nowait 方法和 Exception Empty 来确保 Queue 没有做奇怪的事情。看看我在生产服务器中的这些行。 （当然修改为适合这个例子）。

from Queue import Queue, Empty

while receiver.isAlive:
    try:
        rec = Record(queue.get_nowait())
    except Empty:
        # Set someTime with the value you want
        someTime = 0.1
        sleep(someTime)
    else:
        doSomething(rec)

另外，请记住以下几点：

time.sleep() 函数使用底层操作系统的 sleep() 函数。最终，此功能存在局限性。例如，在标准 Windows 安装中，您可以休眠的最小时间间隔是 10 – 13 毫秒。 Linux 内核往往具有更高的滴答率，其间隔通常更接近 1 毫秒。