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

Question

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

我有一个使用 Python 的 multiprocessing 的程序并行运行一些计算。进程之间的通信是使用两个 Queue 完成的对象，work_queue 和 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 读取数据时，我遇到了 KeyError。这个字典中填满了从 result_queue 中获取的数据，因此我怀疑它一定与它有关。

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

最后但同样重要的是，如果样本数据的大小或子流程的数量增加，重现它的可能性也会增加。

有什么想法吗？

Answer 1

我在您的示例代码中添加了一些额外的调试，我想我找到了问题所在。除了关于使用 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

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