代码之家 › 专栏 › 技术社区 › Elliott Slaughter

Python多处理中的嵌套并行性

python-multiprocessing multiprocessing python

Elliott Slaughter · 技术社区 · 8 年前

我知道这听起来像是以前问过的问题,但是等等,我会解释为什么其他选项不起作用。

我当前正在使用 multiprocessing.Pool 在应用程序中实现并行性,并希望将其扩展为能够利用嵌套并行性。只是通过 Pool 对象作为的参数 apply_async 不起作用 as noted in other answers 因为 水塘

以下是我的要求:

我需要某种类型的池来限制并发执行任务的数量。例如。 multiprocess.Pool 用于此目的,除非它不能传递给其他进程。
我需要嵌套并行。在我的应用程序中,我需要执行I/O以确定嵌套的工作是什么,所以我绝对不想从单个线程执行此操作。我想这排除了 this question .
它需要在标准库中;我无法添加依赖项。这排除了 this answer .
我非常希望它能与Python 2和Python 3一起使用。然而,如果可以证明迁移到Python 3可以解决我的问题,我会考虑它。

我不需要专门使用多个进程,使用线程也可以,因为大部分工作都是I/O或等待子进程完成。

我试过使用 multiprocessing.dummy ,它是相同的接口,但在 threading . 然而,当我试图打电话时 get() 为了检索测试结果,我得到了以下错误,因此我认为这是错误的。

  File "/usr/lib/python2.7/multiprocessing/pool.py", line 567, in get
    raise self._value
ValueError: signal only works in main thread

我知道 concurrent.futures Python 3中的库,但这似乎有一些严重的限制。例如,本节中的第二个示例在我的案例中似乎是一个止步符:

https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor

我看不出,基本上任何直接编写的嵌套并行算法都无法避免这种情况。所以,即使我愿意使用Python 3,我也认为这不是一个好方法。

在没有编写自己的实现之前,我不知道标准库中有任何其他可用选项。

1 回复 | 直到 8 年前

Michal Charemza 8 年前

你似乎排除了这个可能性,但我怀疑 https://docs.python.org/3/library/concurrent.futures.html#threadpoolexecutor 或 https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor 如果您能够移动到Python 3,或者为Python 2添加依赖项,那么这将是可行的。

如果在处理每个文件之前不必触发每个文件的额外工作,那么可以使用一个协调线程来触发所有其他线程,这样就可以防止死锁,如下例所示。

from concurrent.futures import ThreadPoolExecutor
import time

pool = ThreadPoolExecutor(max_workers=3)

def find_work_inputs(dummy_file):
    print("{}: Finding work...".format(dummy_file))
    time.sleep(1)
    work = range(0, dummy_file)
    print("{}: Work is {}".format(dummy_file, work))
    return work

def do_work(dummy_file, work_input):
    print("{}: {}".format(dummy_file, work_input))
    print("{}: Doing work {}...".format(dummy_file, work_input))
    time.sleep(1)
    return work_input * work_input

dummy_files = [1,2,3,4,5]

futures = []
for dummy_file in dummy_files:
    work_inputs = pool.submit(find_work_inputs, dummy_file)
    for work_input in work_inputs.result():
        result = work_input
        futures.append((dummy_file, result, pool.submit(do_work, dummy_file, result)))

for dummy_file, work_input, future in futures:
    print("Result from file:{} input:{} is {}".format(dummy_file, work_input, future.result()))

或者,如果第一级上的每个线程都需要自己触发工作,那么额外的工作可能需要在另一个池中以防止死锁(取决于何时 result() 在每个未来调用)如下所示。

from concurrent.futures import ThreadPoolExecutor
import time

find_work_pool = ThreadPoolExecutor(max_workers=3)
do_work_pool = ThreadPoolExecutor(max_workers=3)

def find_work_inputs(dummy_file):
    print("{}: Finding work...".format(dummy_file))
    time.sleep(1)
    work = range(0, dummy_file)
    print("{}: Work is {}".format(dummy_file, work))

    futures = []
    for work_input in work:
        futures.append((dummy_file, work_input, do_work_pool.submit(do_work, dummy_file, work_input)))
    return futures

def do_work(dummy_file, work_input):
    print("{}: {}".format(dummy_file, work_input))
    print("{}: Doing work {}...".format(dummy_file, work_input))
    time.sleep(1)
    return work_input * work_input

dummy_files = [1,2,3,4,5]

futures = []
for dummy_file in dummy_files:
    futures.extend(find_work_pool.submit(find_work_inputs, dummy_file).result())

for dummy_file, work_input, future in futures:
    print("Result from file:{} input:{} is {}".format(dummy_file, work_input, future.result()))