多进程运行比单个进程慢答案

【问题标题】：Multiprocessing Running Slower than a Single Process多进程运行比单个进程慢
【发布时间】：2019-01-26 01:46:55
【问题描述】：

我正在尝试使用多处理来跨多个进程运行许多模拟；但是，据我所知，我编写的代码仅使用了 1 个进程。

更新

感谢@PaulBecotte，我已经完成了所有流程（我认为）；但是，多处理似乎比其非多处理对应物的运行速度要慢得多。

例如，不包括函数和类声明/实现和导入，我有：

def monty_hall_sim(num_trial, player_type='AlwaysSwitchPlayer'):
    if player_type == 'NeverSwitchPlayer':
        player = NeverSwitchPlayer('Never Switch Player')
    else:
        player = AlwaysSwitchPlayer('Always Switch Player')

    return (MontyHallGame().play_game(player) for trial in xrange(num_trial))

def do_work(in_queue, out_queue):
    while True:
        try:
            f, args = in_queue.get()
            ret = f(*args)
            for result in ret:
                out_queue.put(result)
        except:
            break

def main():
    logging.getLogger().setLevel(logging.ERROR)

    always_switch_input_queue = multiprocessing.Queue()
    always_switch_output_queue = multiprocessing.Queue()

    total_sims = 20
    num_processes = 5
    process_sims = total_sims/num_processes

    with Timer(timer_name='Always Switch Timer'):
        for i in xrange(num_processes):
            always_switch_input_queue.put((monty_hall_sim, (process_sims, 'AlwaysSwitchPlayer')))

        procs = [multiprocessing.Process(target=do_work, args=(always_switch_input_queue, always_switch_output_queue)) for i in range(num_processes)]

        for proc in procs:
            proc.start()

        always_switch_res = []
        while len(always_switch_res) != total_sims:
            always_switch_res.append(always_switch_output_queue.get())

        always_switch_success = float(always_switch_res.count(True))/float(len(always_switch_res))

    print '\tLength of Always Switch Result List: {alw_sw_len}'.format(alw_sw_len=len(always_switch_res))
    print '\tThe success average of switching doors was: {alw_sw_prob}'.format(alw_sw_prob=always_switch_success)

产生：

    Time Elapsed: 1.32399988174 seconds
    Length: 20
    The success average: 0.6

但是，我试图将其用于total_sims = 10,000,000 而不是num_processes = 5，这样做所花费的时间比使用 1 个进程要长得多（1 个进程在大约 3 分钟内返回）。我与之比较的非多处理对应物是：

def main():
    logging.getLogger().setLevel(logging.ERROR)

    with Timer(timer_name='Always Switch Monty Hall Timer'):
        always_switch_res = [MontyHallGame().play_game(AlwaysSwitchPlayer('Monty Hall')) for x in xrange(10000000)]

        always_switch_success = float(always_switch_res.count(True))/float(len(always_switch_res))

    print '\n\tThe success average of not switching doors was: {not_switching}' \
          '\n\tThe success average of switching doors was: {switching}'.format(not_switching=never_switch_success,
                                                                               switching=always_switch_success)

【问题讨论】：

您的子进程 - 他们从输入队列消费并导出到输出队列？您只将 4 条（现在 5 条）消息放入队列，所以这不应该是您获得的输出数量吗？
嗨@PaulBecotte，Python 新手，所以我可能完全搞砸了上面的代码。我正在尝试做的是创建一个包含 5 个元组 (fcn_name, (num_of_trials_to_perform,)) 的输入队列，其中函数是 Monty Hall 模拟，要执行的试验次数是要在Monty Hall 函数用于子进程。因此，总共有 20 个模拟，我想要 5 个带有 (fcn_name, (4,)) 的元组，然后在子进程中使用这些元组中的每一个。如果这有意义？

标签： python python-2.7 queue multiprocessing python-multiprocessing

【解决方案1】：

您可以尝试在某些 if 语句下导入“process”

【讨论】：

在monty_hall_sim函数@26099 ?

【解决方案2】：

编辑-你改变了一些东西，让我试着解释一下。

您放入输入队列的每条消息都会调用 monty_hall_sim 函数并将 num_trial 消息发送到输出队列。

所以你最初的实现是正确的——得到 20 条输出消息，发送 5 条输入消息。

但是，你的函数有点错误。

for trial in xrange(num_trial):
    res = MontyHallGame().play_game(player)
    yield res

这将把函数变成一个生成器，它会在每次调用 next() 时提供一个新值——太棒了！问题来了

while True:
    try:
        f, args = in_queue.get(timeout=1)
        ret = f(*args)
        out_queue.put(ret.next())
    except:
        break

在这里，每次通过循环时，您都会创建一个带有 NEW 消息的 NEW 生成器。旧的被扔掉了。因此，在这里，每条输入消息只在队列中添加一条输出消息，然后再将其丢弃并获取另一条消息。正确的写法是-

while True:
    try:
        f, args = in_queue.get(timeout=1)
        ret = f(*args)
        for result in ret:
            out_queue.put(ret.next())
    except:
        break

这样做将继续从生成器产生输出消息，直到它完成（在这种情况下产生 4 条消息之后）

【讨论】：

谢谢，这让我得到了我一直试图获得@PaulBecotte 的输出！但是，我的代码仍然比使用 1 个进程实现 10,000,000 次总模拟的相同功能运行得慢，这很奇怪。
您确实添加了整整 1 秒的等待时间作为进程的终止条件。这似乎可能会主导这种复杂程序的处理速度。
我听从了你的建议，在两个 get() 调用中都取消了 1 秒的超时，我将进程的 procs 列表更改为生成器，这降低了所花费的时间，因此它只是使用 1 个进程 @PaulBecotte 将同一程序所花费的时间加倍。此外，添加更多进程不会改变所花费的时间。
您的设计有几个问题。对我来说最明显的是你在启动子进程之前填满了队列——你可能应该首先启动你的进程池。其次，您正在做的具体事情是为传递的每条消息做很少的工作——因此您承担了序列化消息并来回传递它的开销，只做一点点处理。也许每条消息都会导致数千次迭代？
我以为我当前的代码是让每个子进程调用do_work函数，它调用monty_hall_sim函数来创建数千个模拟并将它们存储在生成器中，然后存储项目输出中的生成器。我不了解该程序如何真正运作@PaulBecotte？

【解决方案3】：

通过将 monty_hall_sim 的返回更改为列表理解，让 do_work 将列表添加到输出队列，然后使用输出队列返回的列表扩展 main 的结果列表，我能够使我的代码运行得更快。让它在大约 13 秒内运行。

【讨论】：