在Python中反转ProcessPoolExecutor

Posted 2021-04-13

这与我正在努力解决的earlier problem有关。基本上我需要ProcessPoolExecutor的逆设计，其中我有许多查询过程和一个工人，它可以批量计算和发回结果。

使用一个共享队列可以轻松发送工作项，但我仍然没有一个很好的解决方案，可以将所有结果发送回正确的进程中的正确线程。

答案

我认为每个查询过程都有一个单独的multiprocessing.pipe是最有意义的。工作进程等待任何管道上的可用项目，并将其出列并处理它，跟踪它来自哪个管道。在返回数据时，它会将结果提供给正确的管道。

这是一个简单的例子：

#!/usr/bin/env python3

import multiprocessing as mp

def worker(pipes):
    quit = [False] * len(pipes)
    results = [''] * len(pipes)

    # Wait for all workers to send None before quitting
    while not all(quit):
        ready = mp.connection.wait(pipes)
        for pipe in ready:

            # Get index of query proc's pipe
            i = pipes.index(pipe)

            # Receive and "process"
            obj = pipe.recv()
            if obj is None:
                quit[i] = True
                continue
            result = str(obj)
            results[i] += result

            # Send back to query proc
            pipes[i].send(result)
    print(results)


def query(pipe):
    for i in 'do some work':
        pipe.send(i)
        assert pipe.recv() == i
    pipe.send(None) # Send sentinel

if __name__ == '__main__':
    nquery_procs = 8
    work_pipes, query_pipes = zip(*(mp.Pipe() for _ in range(nquery_procs)))

    query_procs = [mp.Process(target=query, args=(pipe,)) for pipe in query_pipes]
    for p in query_procs:
        p.start()
    worker(work_pipes)
    for p in query_procs:
        p.join()

或者，您可以为每个查询过程提供一个ID号（可能只是它的管道索引），并且任何请求必须是(id_num, data)的元组。这只是绕过每个循环上执行pipes.index(pipe)的工作进程，所以我不确定它会给你多少钱。