并行执行功能列表

Question

因此，使用多处理模块，很容易与这样的不同参数并行运行一个函数：

from multiprocessing import Pool
def f(x):
    return x**2

p = Pool(2)
print(p.map(f, [1, 2]))

，但我有兴趣在同一参数上执行函数列表。假设我有以下两个功能：

def f(x):
    return x**2

def g(x):
    return x**3 + 2

如何并行执行相同参数(例如x = 1(？

Answer 1

您可以使用Pool.apply_async()。您以(函数，gright_tuple(的形式捆绑任务，然后将每个任务馈送到 apply_async()。

from multiprocessing import Pool
from itertools import repeat

def f(x):
    for _ in range(int(50e6)): # dummy computation
        pass
    return x ** 2

def g(x):
    for _ in range(int(50e6)): # dummy computation
        pass
    return x ** 3

def parallelize(n_workers, functions, arguments):
    # if you need this multiple times, instantiate the pool outside and
    # pass it in as dependency to spare recreation all over again
    with Pool(n_workers) as pool:
        tasks = zip(functions, repeat(arguments))
        futures = [pool.apply_async(*t) for t in tasks]
        results = [fut.get() for fut in futures]
    return results

if __name__ == '__main__':
    N_WORKERS = 2
    functions = f, g
    results = parallelize(N_WORKERS, functions, arguments=(10,))
    print(results)

示例输出：

[100, 1000]
Process finished with exit code 0

Answer 2

您可以返回元组。使用轻量级模块：Joblib可以很容易地以非常紧凑的方式进行操作。我推荐琼布利布，因为它是轻量级

from joblib import Parallel, delayed
import multiprocessing
import timeit

# Implementation 1
def f(x):    
    return x**2, x**3 + 2

#Implementation 2 for a more sophisticated second or more functions
def g(x):
    return  x**3 + 2

def f(x):    
    return x**2, g(x)

if __name__ == "__main__":
  inputs = [i for i in range(32)]
  num_cores = multiprocessing.cpu_count()
  t1 = timeit.Timer()
  result = Parallel(n_jobs=num_cores)(delayed(f)(i) for i in inputs)
  print(t1.timeit(1)) 

与您在问题中已经有过的多处理。

from multiprocessing import Pool, cpu_count
import timeit

def g(x):
    return x**3 + 2

def f(x):    
    return x**2, g(x)
if __name__ == "__main__":
  inputs = [i for i in range(32)]
  num_cores = cpu_count()
  p = Pool(num_cores)
  t1 = timeit.Timer()
  result = p.map(f, inputs)
  print(t1.timeit(1))
  print(result)    

示例输出：

print(result)
[(0, 2), (1, 3), (4, 10), (9, 29), (16, 66), (25, 127), (36, 218), (49, 345), 
(64, 514), (81, 731), (100, 1002), (121, 1333), (144, 1730), (169, 2199), 
(196, 2746), (225, 3377), (256, 4098), (289, 4915), (324, 5834), (361, 6861), 
(400, 8002), (441, 9263), (484, 10650), (529, 12169), (576, 13826), (625, 
15627), (676, 17578), (729, 19685), (784, 21954), (841, 24391), (900, 27002), 
(961, 29793)]
print(t1.timeit(1))
    5.000001692678779e-07         #(with 16 cpus and 64 Gb RAM) 

for：inputs = range(2000(，花了时间：1.100000190490391E-06