python - 从 multiprocessing.Queue 获得近后进先出行为的干净方法？ (甚至只是*不*接近先进先出)

Question

有谁知道从 multiprocessing.Queue 获得接近 LIFO 甚至不接近 FIFO(例如随机)行为的干净方法？

替代问题:有人可以指出管理 multiprocessing.Queue 背后的实际存储结构的线程的代码吗？ 这似乎是微不足道的提供大约后进先出的访问权限，但我在试图找到它的过程中迷失了方向。

注意事项:

1. 我相信multiprocessing.Queue does not guarantee order .美好的。但它是 near-FIFO，所以 near-LIFO 会很棒。
2. 我可以将所有当前项目从队列中取出并在处理它们之前颠倒顺序，但我更愿意尽可能避免困惑。

---

(编辑)澄清一下:我正在使用 multiprocessing 进行 CPU 限制模拟，因此不能使用 Queue 中的专用队列。由于我已经几天没有看到任何答案，所以我在上面添加了备选问题。

---

如果这是一个问题，下面是 multiprocessing.Queue 接近 FIFO 的轻微证据。它只是表明在一个简单的情况下(单线程)，它在我的系统上完全是 FIFO:

import multiprocessing as mp
import Queue

q = mp.Queue()

for i in xrange(1000):
    q.put(i)

deltas = []
while True:
    try:
        value1 = q.get(timeout=0.1)
        value2 = q.get(timeout=0.1)
        deltas.append(value2-value1)
    except Queue.Empty:
        break

#positive deltas would indicate the numbers are coming out in increasing order
min_delta, max_delta = min(deltas), max(deltas)
avg_delta = sum(deltas)/len(deltas)

print "min", min_delta
print "max", max_delta
print "avg", avg_delta

打印:最小值、最大值和平均值恰好为 1(完美的 FIFO)

Answer 1

我查看了我的 Python 安装(Python 2.7，但与我简要介绍的 Python 3.2 版本没有明显不同的 Lib/multiprocessing/queues.py 中的 Queue 类检查)。以下是我的理解:

队列对象维护着两组对象。一组是所有进程共享的多进程安全原语。其他的由每个进程单独创建和使用。

跨进程对象在__init__方法中设置:

1. 一个Pipe对象，其两端分别保存为self._reader和self._writer。
2. 一个 BoundedSemaphore 对象，它计算(并可选地限制)队列中的对象数量。
3. Lock 对象用于读取管道，在非 Windows 平台上另一个用于写入。 (我假设这是因为写入管道在 Windows 上本质上是多进程安全的。)

每个进程对象在 _after_fork 和 _start_thread 方法中设置:

1. collections.deque 对象，用于缓冲写入 Pipe。
2. 一个 threading.condition 对象，用于在缓冲区不为空时发出信号。
3. 一个执行实际写入的threading.Thread 对象。它是延迟创建的，因此在给定进程中至少请求对队列进行一次写入之前，它不会存在。
4. 各种 Finalize 对象，在进程结束时清除内容。

从队列中get 非常简单。您获取读锁，递减信号量，并从 Pipe 的读端获取一个对象。

put 更复杂。它使用多个线程。 put 的调用者获取条件的锁，然后将其对象添加到缓冲区并在解锁之前向条件发出信号。它还会增加信号量并启动编写器线程(如果它尚未运行)。

编写器线程在 _feed 方法中永远循环(直到被取消)。如果缓冲区为空，它会等待 notempty 条件。然后它从缓冲区中取出一个项目，获取写锁(如果它存在)并将该项目写入管道。

---

那么，鉴于所有这些，您能否修改它以获得后进先出队列？这似乎并不容易。管道本质上是 FIFO 对象，虽然 Queue 不能保证整体的 FIFO 行为(由于来自多个进程的写入的异步性质)它总是主要是 FIFO。

如果您只有一个消费者，您可以从队列中获取 对象并将它们添加到您自己的进程本地堆栈中。做一个多消费者堆栈会更难，尽管使用共享内存一个有界大小的堆栈不会太难。您需要一把锁、一对条件(用于在满状态和空状态时阻塞/发出信号)、一个共享整数值(用于保存值的数量)和一个适当类型的共享数组(用于值本身)。