python - 使用 Python 多处理进程/队列的奇怪错误

Question

很抱歉这个问题很长，但我找不到更好的方法来总结它。

我有一个程序，它使用 Pythonmultiprocessing并行运行一些计算。进程之间的通信是使用两个Queue对象 awork_queue和 a完成的result_queue。主进程填充work_queue用于计算的数据，然后启动几个子进程，这些子进程将使用该队列并将结果存储到result_queue.

一切似乎都很好，但是当我对样本数据量（即进入的数据量work_queue）和子进程的数量进行了一些尝试时，我开始遇到一个让我感到困惑的错误小时。

下面的代码说明了这个问题：

# -- queue_bug.py --

import sys
import time
import random
import datetime
import traceback
# Need this to catch the Queue.Empty exception
import Queue
import multiprocessing
from multiprocessing import Process
from multiprocessing import Queue as MultiprocessingQueue

# -------------------------------------------------------------------
# do_calculation
# -------------------------------------------------------------------
def do_calculation(p_name, work_queue, result_queue):
    def log(msg):
        print '%s [%s] %s' % (datetime.datetime.now(), p_name, msg)
    log('Starting up...')
    while True:
        # Get work from queue
        try:
            work = work_queue.get(timeout = 0.1)
            test_id   = work[0]
            test_data = work[1]
        except Queue.Empty:
            break
        # this is just a dummy loop
        for i in range(100):
            test_result = [x * random.random() for x in test_data]
        result_queue.put((test_id, test_data, test_result))
    log('Finished')

# -------------------------------------------------------------------
# main
# -------------------------------------------------------------------
def main():
    def log(msg):
        print '%s [main   ] %s' % (datetime.datetime.now(), msg)
    try:
        num_tests = int(sys.argv[1])
        num_procs = int(sys.argv[2])
    except Exception:
        print 'usage: <prog> number-of-tests number-of-subprocesses'
        sys.exit()

    log('Initializing queues...')
    work_queue   = MultiprocessingQueue()
    result_queue = MultiprocessingQueue()
    log('Creating subprocesses...')
    process_list = []
    for i in range(num_procs):
        p_name = 'PROC_%02d' % (i+1)
        log('    Initializing %s' % p_name)
        p = Process(
            target = do_calculation,
            args   = (p_name, work_queue, result_queue),
            name   = p_name)
        p.daemon = True
        process_list.append(p)
    log('Populating the work_queue...')
    for test_id in range(num_tests):
        work_queue.put((test_id, [test_id]*20))
    log('Work_queue size is %d' % work_queue.qsize())
    log('Starting the subprocesses...')
    for p in process_list:
        p.start()
    log('Waiting until the work_queue is empty...')
    while True:
        log('    Work_queue size is %d' % work_queue.qsize())
        if work_queue.qsize() > 0:
            time.sleep(0.5)
        else:
            break
    log('Waiting until the result_queue is completely filled...')
    while True:
        log('    Result_queue size is %d' % result_queue.qsize())
        if result_queue.qsize() < num_tests:
            time.sleep(0.5)
        else:
            break
    log('Getting results...')
    result_dict = {}
    while True:
        try:
            queue_data           = result_queue.get_nowait()
            test_id              = queue_data[0]
            test_data            = queue_data[1]
            test_result          = queue_data[2]
            result_dict[test_id] = test_result
        except Queue.Empty:
            log('    All results loaded from result_queue')
            break
    log('Storing test results in result_summary...')
    result_summary = []
    for test_id in range(num_tests):
        try:
            test_result = result_dict[test_id]
            result_summary.append((test_id, test_result))
        except KeyError:
            ex = traceback.format_exc()
            log('ERROR: Exception found: %s' % ex)
            sys.exit()
    log('Success.')
    return result_summary

if __name__ == '__main__':
    main()

现在，当我尝试运行它时：

尝试 1：10.000 次计算，10 个子流程 - OK

$ python queue_bug.py 10000 10
2012-12-04 19:24:25.430667 [main   ] Initializing queues...
2012-12-04 19:24:25.440521 [main   ] Creating subprocesses...
2012-12-04 19:24:25.440550 [main   ]     Initializing PROC_01
2012-12-04 19:24:25.440576 [main   ]     Initializing PROC_02
2012-12-04 19:24:25.440597 [main   ]     Initializing PROC_03
2012-12-04 19:24:25.440617 [main   ]     Initializing PROC_04
2012-12-04 19:24:25.440637 [main   ]     Initializing PROC_05
2012-12-04 19:24:25.440656 [main   ]     Initializing PROC_06
2012-12-04 19:24:25.440679 [main   ]     Initializing PROC_07
2012-12-04 19:24:25.440699 [main   ]     Initializing PROC_08
2012-12-04 19:24:25.440721 [main   ]     Initializing PROC_09
2012-12-04 19:24:25.440741 [main   ]     Initializing PROC_10
2012-12-04 19:24:25.440759 [main   ] Populating the work_queue...
2012-12-04 19:24:25.494263 [main   ] Work_queue size is 10000
2012-12-04 19:24:25.494301 [main   ] Starting the subprocesses...
2012-12-04 19:24:25.495515 [PROC_01] Starting up...
2012-12-04 19:24:25.495802 [PROC_02] Starting up...
2012-12-04 19:24:25.496212 [PROC_03] Starting up...
2012-12-04 19:24:25.496557 [PROC_04] Starting up...
2012-12-04 19:24:25.496896 [PROC_05] Starting up...
2012-12-04 19:24:25.497300 [PROC_06] Starting up...
2012-12-04 19:24:25.497705 [PROC_07] Starting up...
2012-12-04 19:24:25.498074 [PROC_08] Starting up...
2012-12-04 19:24:25.498258 [main   ] Waiting until the work_queue is empty...
2012-12-04 19:24:25.498349 [main   ]     Work_queue size is 9974
2012-12-04 19:24:25.498661 [PROC_09] Starting up...
2012-12-04 19:24:25.499765 [PROC_10] Starting up...
2012-12-04 19:24:25.998914 [main   ]     Work_queue size is 0
2012-12-04 19:24:25.998954 [main   ] Waiting until the result_queue is completely filled...
2012-12-04 19:24:25.998976 [main   ]     Result_queue size is 10000
2012-12-04 19:24:25.998993 [main   ] Getting results...
2012-12-04 19:24:26.029774 [PROC_06] Finished
2012-12-04 19:24:26.029798 [PROC_03] Finished
2012-12-04 19:24:26.029824 [PROC_08] Finished
2012-12-04 19:24:26.029853 [PROC_02] Finished
2012-12-04 19:24:26.029868 [PROC_01] Finished
2012-12-04 19:24:26.029898 [PROC_07] Finished
2012-12-04 19:24:26.029921 [PROC_09] Finished
2012-12-04 19:24:26.029942 [PROC_10] Finished
2012-12-04 19:24:26.031040 [PROC_04] Finished
2012-12-04 19:24:26.031057 [PROC_05] Finished
2012-12-04 19:24:26.087804 [main   ]     All results loaded from result_queue
2012-12-04 19:24:26.087844 [main   ] Storing test results in result_summary...
2012-12-04 19:24:26.092477 [main   ] Success.

尝试 2：70.000 次计算，10 个子进程 - 错误

$ python queue_bug.py 70000 10
2012-12-04 19:25:01.083092 [main   ] Initializing queues...
2012-12-04 19:25:01.093483 [main   ] Creating subprocesses...
2012-12-04 19:25:01.093520 [main   ]     Initializing PROC_01
2012-12-04 19:25:01.093548 [main   ]     Initializing PROC_02
2012-12-04 19:25:01.093570 [main   ]     Initializing PROC_03
2012-12-04 19:25:01.093591 [main   ]     Initializing PROC_04
2012-12-04 19:25:01.093612 [main   ]     Initializing PROC_05
2012-12-04 19:25:01.093632 [main   ]     Initializing PROC_06
2012-12-04 19:25:01.093656 [main   ]     Initializing PROC_07
2012-12-04 19:25:01.093676 [main   ]     Initializing PROC_08
2012-12-04 19:25:01.093699 [main   ]     Initializing PROC_09
2012-12-04 19:25:01.093720 [main   ]     Initializing PROC_10
2012-12-04 19:25:01.093738 [main   ] Populating the work_queue...
2012-12-04 19:25:01.395974 [main   ] Work_queue size is 70000
2012-12-04 19:25:01.396012 [main   ] Starting the subprocesses...
2012-12-04 19:25:01.397601 [PROC_01] Starting up...
2012-12-04 19:25:01.398183 [PROC_02] Starting up...
2012-12-04 19:25:01.398545 [PROC_03] Starting up...
2012-12-04 19:25:01.399021 [PROC_04] Starting up...
2012-12-04 19:25:01.399621 [PROC_05] Starting up...
2012-12-04 19:25:01.400137 [PROC_06] Starting up...
2012-12-04 19:25:01.400675 [PROC_07] Starting up...
2012-12-04 19:25:01.401200 [PROC_08] Starting up...
2012-12-04 19:25:01.401645 [main   ] Waiting until the work_queue is empty...
2012-12-04 19:25:01.401691 [PROC_09] Starting up...
2012-12-04 19:25:01.401738 [main   ]     Work_queue size is 69959
2012-12-04 19:25:01.402387 [PROC_10] Starting up...
2012-12-04 19:25:01.902063 [main   ]     Work_queue size is 58415
2012-12-04 19:25:02.402640 [main   ]     Work_queue size is 47302
2012-12-04 19:25:02.903067 [main   ]     Work_queue size is 36145
2012-12-04 19:25:03.403650 [main   ]     Work_queue size is 24992
2012-12-04 19:25:03.904065 [main   ]     Work_queue size is 13481
2012-12-04 19:25:04.404643 [main   ]     Work_queue size is 1951
2012-12-04 19:25:04.588562 [PROC_02] Finished
2012-12-04 19:25:04.588580 [PROC_06] Finished
2012-12-04 19:25:04.588611 [PROC_10] Finished
2012-12-04 19:25:04.588631 [PROC_03] Finished
2012-12-04 19:25:04.589705 [PROC_04] Finished
2012-12-04 19:25:04.589741 [PROC_09] Finished
2012-12-04 19:25:04.589764 [PROC_05] Finished
2012-12-04 19:25:04.589791 [PROC_08] Finished
2012-12-04 19:25:04.589814 [PROC_01] Finished
2012-12-04 19:25:04.589844 [PROC_07] Finished
2012-12-04 19:25:04.905065 [main   ]     Work_queue size is 0
2012-12-04 19:25:04.905098 [main   ] Waiting until the result_queue is completely filled...
2012-12-04 19:25:04.905121 [main   ]     Result_queue size is 70000
2012-12-04 19:25:04.905140 [main   ] Getting results...
2012-12-04 19:25:05.012083 [main   ]     All results loaded from result_queue
2012-12-04 19:25:05.012140 [main   ] Storing test results in result_summary...
2012-12-04 19:25:05.020498 [main   ] ERROR: Exception found: Traceback (most recent call last):
  File "queue_bug.py", line 95, in main
    test_result = result_dict[test_id]
KeyError: 10647

在第二次尝试时，我在尝试从result_dict. 这本字典充满了从中获取的数据result_queue，因此我怀疑它一定与它有关。

我还注意到，每次我使用失败的参数组合（例如 70000 / 10）运行它时，都会KeyError在不同的键上引发，这似乎表明存在一些并发/同步问题。

最后但并非最不重要的一点是，如果样本数据的大小或子进程的数量增加，重现它的可能性也会增加。

有任何想法吗？

score 3 · Accepted Answer

我在您的示例代码中添加了一些额外的调试，我想我找到了问题所在。除了现有的关于 using 的评论之外JoinableQueue，主要问题是最后当您处理结果时，您会执行以下操作：

try:
    queue_data = result_queue.get_nowait()
except Queue.Empty:
    break

但是Queue.Empty即使队列实际上不是空的，也可以引发，只是因为get_nowait()超时太快。相反，您应该尝试：

try:
    queue_data = result_queue.get_nowait()
except Queue.Empty:
    if result_queue.qsize() < 1:
        break

换句话说，在跳出循环之前确保队列实际上是空的。

score 3 · Accepted Answer

您的错误在从结果队列中获取数据的循环中

 while result_queue.qsize() > 0:
        try:
            queue_data           = result_queue.get()
            test_id              = queue_data[0]
            test_data            = queue_data[1]
            test_result          = queue_data[2]
            result_dict[test_id] = test_result
        except Queue.Empty:
            log('    All results loaded from result_queue')
            break

当仍然有结果时，你get_nowait的返回是空的。我怀疑在幕后有一些开销随着在它放弃之前没有完成的线程数量而增加。

python - 使用 Python 多处理进程/队列的奇怪错误

尝试 1：10.000 次计算，10 个子流程 - OK

尝试 2：70.000 次计算，10 个子进程 - 错误

2 回答 2

Related

Reference