在 python 中实现“竞争”进程

Question

我正在尝试实现一个函数，它接受 2 个函数作为参数，同时运行这两个函数，返回首先返回的函数的值，并在它完成执行之前杀死较慢的函数。 我的问题是，当我尝试清空用于收集返回值的 Queue 对象时，我被卡住了。 是否有更“正确”的方法来处理这种情况甚至现有模块？如果没有，谁能解释我做错了什么？ 这是我的代码（上面函数的实现是'run_both()'）：

import multiprocessing as mp
from time import sleep


Q = mp.Queue()

def dump_queue(queue):
    result = []
    for i in iter(queue.get, 'STOP'):
        result.append(i)
    return result

def rabbit(x):
    sleep(10)
    Q.put(x)

def turtle(x):
    sleep(30)
    Q.put(x)

def run_both(a,b):
    a.start()
    b.start()
    while a.is_alive() and b.is_alive():
            sleep(1)
    if a.is_alive():
            a.terminate()
    else:
            b.terminate()
    a.join()
    b.join()
    return dump_queue(Q)


p1 = mp.Process(target=rabbit, args=(1,))
p1 = mp.Process(target=turtle, args=(2,))
run_both(p1, p2)

Answer 1

这是一个使用 multiprocessing 调用 2 个或更多函数并返回最快结果的示例。但是，有一些重要的事情需要注意。

1. 在 IDLE 中运行 multiprocessing 代码有时会导致问题。此示例有效，但我在尝试解决此问题时确实遇到了该问题。
2. 多处理代码应该从if __name__ == '__main__' 子句内部开始，否则如果主模块被另一个进程重新导入，它将再次运行。阅读多处理文档页面了解更多信息。
3. 结果队列直接传递给使用它的每个进程。当您通过引用模块中的全局名称来使用队列时，代码在 Windows 上会失败，因为每个进程都使用队列的新实例。在此处阅读更多信息Multiprocessing Queue.get() hangs

我还在这里添加了一点功能，可以知道实际使用了哪个进程的结果。

import multiprocessing as mp
import time
import random

def task(value):
    # our dummy task is to sleep for a random amount of time and
    # return the given arg value
    time.sleep(random.random())
    return value

def process(q, idx, fn, args):
    # simply call function fn with args, and push its result in the queue with its index
    q.put([fn(*args), idx])

def fastest(calls):
    queue = mp.Queue()
    # we must pass the queue directly to each process that may use it
    # or else on Windows, each process will have its own copy of the queue
    # making it useless

    procs = []
    # create a 'mp.Process' that calls our 'process' for each call and start it
    for idx, call in enumerate(calls):
        fn = call[0]
        args = call[1:]
        p = mp.Process(target=process, args=(queue, idx, fn, args))
        procs.append(p)
        p.start()
    # wait for the queue to have something
    result, idx = queue.get()
    for proc in procs: # kill all processes that may still be running
        proc.terminate()
        # proc may be using queue, so queue may be corrupted.
        # https://docs.python.org/3.8/library/multiprocessing.html?highlight=queue#multiprocessing.Process.terminate
        # we no longer need queue though so this is fine
    return result, idx

if __name__ == '__main__':

    from datetime import datetime

    start = datetime.now()
    print(start)
    # to be compatible with 'fastest', each call is a list with the first
    # element being callable, followed by args to be passed
    calls = [
        [task, 1],
        [task, 'hello'],
        [task, [1,2,3]]
        ]
    val, idx = fastest(calls)
    end = datetime.now()
    print(end)
    print('elapsed time:', end-start)
    print('returned value:', val)
    print('from call at index', idx)

示例输出：

2019-12-21 04:01:09.525575
2019-12-21 04:01:10.171891
elapsed time: 0:00:00.646316
returned value: hello
from call at index 1

Answer 2

除了倒数第二行的错字应该是：

p2 = mp.Process(target=turtle, args=(2,))       # not p1

使程序运行的最简单的更改是添加：

Q.put('STOP')

到turtle() 和rabbit() 的末尾。

---

你也不需要一直循环观察进程是否还活着，根据定义，如果你只是读取消息队列并收到STOP，其中一个已经完成，所以你可以将run_both()替换为：

def run_both(a,b):
    a.start()
    b.start()
    result =  dump_queue(Q)
    a.terminate()
    b.terminate()
    return result

---

您可能还需要考虑如果两个进程同时将一些消息放入队列中会发生什么。他们可能会混淆。也许考虑使用 2 个队列，或者将所有结果合并到一条消息中，而不是将来自 queue.get() 的多个值附加在一起