将多个参数传递给concurrent.futures.Executor.map？

Question

concurrent.futures.Executor.map 接受可变数量的可迭代对象，从中调用给定的函数。 如果我有一个生成器，可以生成通常在适当位置解包的元组，我应该如何调用它？

以下内容不起作用，因为每个生成的元组都作为映射的不同参数给出

args = ((a, b) for (a, b) in c)
for result in executor.map(f, *args):
    pass

：生成器，映射所需的参数可能如下所示：

executor.map(
    f,
    (i[0] for i in args),
    (i[1] for i in args),
    ...,
    (i[N] for i in args),
)

Answer 1

一个重复的参数，c 中的一个参数

from itertools import repeat
for result in executor.map(f, repeat(a), c):
    pass

需要解包 c 的项，并且可以解包 c

from itertools import izip
for result in executor.map(f, *izip(*c)):
    pass

需要解包 的项c，无法解包 c

1. 更改 f 以采用单个参数并解包函数中的参数。

2. 如果 c 中的每个项目都有可变数量的成员，或者您仅调用 f 几次：

   executor.map(lambda args, f=f: f(*args), c)
   

   它定义了一个新函数，用于从 c 中解压每个项目并调用 f。在 lambda 中使用 f 的默认参数使 f 在 lambda 内部成为本地变量，从而减少查找时间。

3. 如果您有固定数量的参数，并且需要多次调用 f：

   从集合导入双端队列
   def itemtee（可迭代，n=2）：
       def gen(it = iter(iterable), items = deque(), next = next):
           popleft = 项目.popleft
           扩展 = items.extend
           而真实：
               如果不是物品：
                   扩展（下一个（它））
               产生 popleft()
       返回 [gen()] * n
   
   执行器.map(f, *itemtee(c, n))
   

其中 n 是参数的数量f 的参数。这是改编自 itertools.tee。

Answer 2

您需要删除 map 调用上的 *：

args = ((a, b) for b in c)
for result in executor.map(f, args):
    pass

这将调用 f、len(args) 次，其中 f 应该接受一个参数。

如果您希望 f 接受两个参数，您可以使用 lambda 调用，例如：

args = ((a, b) for b in c)
for result in executor.map(lambda p: f(*p), args):   # (*p) does the unpacking part
    pass

Answer 3

因此，假设您有一个接受 3 个参数 的函数，并且所有 3 个参数都是动态，并且随着每次调用而不断变化。例如：

def multiply(a,b,c):
    print(a * b * c)

要使用线程多次调用此函数，我首先创建一个元组列表，其中每个元组都是 a、b、c 的一个版本：

arguments = [(1,2,3), (4,5,6), (7,8,9), ....]

我们知道 concurrent.futures< /code> 的 map 函数将接受第一个参数作为目标函数，第二个参数作为函数每个版本的参数列表将被执行。因此，您可能会进行如下调用：

for _ in executor.map(multiply, arguments) # Error

但这会给您带来错误，即该函数需要 3 个参数，但只得到 1。为了解决这个问题，我们创建一个辅助函数：

def helper(numbers):
    multiply(numbers[0], numbers[1], numbers[2])

现在，我们可以使用执行器调用这个函数，如下所示：

with ThreadPoolExecutor() as executor:
     for _ in executor.map(helper, arguments):
         pass

这应该会给你想要的结果。

Answer 4

您可以通过 Python 中的 partial 方法使用柯里化创建新函数

from concurrent.futures import ThreadPoolExecutor
from functools import partial


def some_func(param1, param2):
    # some code

# currying some_func with 'a' argument is repeated
func = partial(some_func, a)
with ThreadPoolExecutor() as executor:
    executor.map(func, list_of_args):
    ...

如果您需要传递多个相同的参数，可以将它们传递到 部分方法

func = partial(some_func, a, b, c)

Answer 5

下面的代码片段展示了如何使用 ThreadPoolExecutor 向函数发送多个参数：

import concurrent.futures


def hello(first_name: str, last_name: str) -> None:
    """Prints a friendly hello with first name and last name"""
    print('Hello %s %s!' % (first_name, last_name))


def main() -> None:
    """Examples showing how to use ThreadPoolExecutor and executer.map
    sending multiple arguments to a function"""

    # Example 1: Sending multiple arguments using tuples
    # Define tuples with sequential arguments to be passed to hello()
    args_names = (
        ('Bruce', 'Wayne'),
        ('Clark', 'Kent'),
        ('Diana', 'Prince'),
        ('Barry', 'Allen'),
    )
    with concurrent.futures.ThreadPoolExecutor() as executor:
        # Using lambda, unpacks the tuple (*f) into hello(*args)
        executor.map(lambda f: hello(*f), args_names)

    print()

    # Example 2: Sending multiple arguments using dict with named keys
    # Define dicts with arguments as key names to be passed to hello()
    kwargs_names = (
        {'first_name': 'Bruce', 'last_name': 'Wayne'},
        {'first_name': 'Clark', 'last_name': 'Kent'},
        {'first_name': 'Diana', 'last_name': 'Prince'},
        {'first_name': 'Barry', 'last_name': 'Allen'},
    )
    with concurrent.futures.ThreadPoolExecutor() as executor:
        # Using lambda, unpacks the dict (**f) into hello(**kwargs)
        executor.map(lambda f: hello(**f), kwargs_names)


if __name__ == '__main__':
    main()

Answer 6

对于 ProcessPoolExecutor.map()< /代码>：

与map(func, *iterables)类似，除了：

可迭代对象是立即收集的，而不是延迟收集的；

func 是异步执行的，并且可能会多次调用 func
同时。

因此，ProcessPoolExecutor.map()的用法与Python内置map()的用法相同。这是文档：

返回一个迭代器，该迭代器将函数应用于可迭代的每个项目，
产生结果。 如果传递额外的可迭代参数，
函数必须接受那么多参数并应用于项目
来自所有并行的可迭代。

结论：将几个参数传递给 map()。

尝试在 python 3 下运行以下代码片段，你就会很清楚了：

from concurrent.futures import ProcessPoolExecutor

def f(a, b):
    print(a+b)

with ProcessPoolExecutor() as pool:
    pool.map(f, (0, 1, 2, 3, 4, 5, 6, 7, 8, 9), (0, 1, 2))

# 0, 2, 4

array = [(i, i) for i in range(3)]
with ProcessPoolExecutor() as pool:
    pool.map(f, *zip(*array))

# 0, 2, 4

Answer 7

我在这里看到了很多答案，但没有一个像使用 lambda 表达式那样直接：

foo(x,y):
吗

想要使用相同的值（即 xVal 和 yVal）调用上述方法 10 次 ？
以 concurrent.futures.ThreadPoolExecutor() 作为执行器：

for _ in executor.map( lambda _: foo(xVal, yVal), range(0, 10)):
    pass

Answer 8

假设您在下面显示的数据框中有这样的数据，并且您希望将第一两列传递给一个函数，该函数将读取图像并预测特征，然后计算差异并返回差异值。

注意：您可以根据您的要求有任何场景，并且可以分别定义函数。

下面的代码片段将这两列作为参数并传递给线程池机制（还显示进度条）

''' function that will give the difference of two numpy feature matrix'''
def getDifference(image_1_loc, image_2_loc, esp=1e-7):
       arr1 = ''' read 1st image and extract feature '''
       arr2 = ''' read 2nd image and extract feature '''
       diff = arr1.ravel() - arr2.ravel() + esp    
       return diff

'''Using ThreadPoolExecutor from concurrent.futures with multiple argument'''

with ThreadPoolExecutor() as executor:
        result = np.array(
                         list(tqdm(
                                   executor.map(lambda x : function(*x), [(i,j) for i,j in df[['image_1','image_2']].values]),
                               total=len(df)
                                  ) 
                             )
                          )

Answer 9

下面是我一直使用的一个简单实用程序。

########### Start of Utility Code ###########

import os
import sys
import traceback

from concurrent import futures
from functools import partial


def catch(fn):
    def wrap(*args, **kwargs):
        result = None
        try:
            result = fn(*args, **kwargs)
        except Exception as err:
            type_, value_, traceback_ = sys.exc_info()
            return None, (
                args,
                "".join(traceback.format_exception(type_, value_, traceback_)),
            )
        else:
            return result, (args, None)

    return wrap


def top_level_wrap(fn, arg_tuple):
    args, kwargs = arg_tuple
    return fn(*args, *kwargs)


def create_processes(fn, values, handle_error, handle_success):
    cores = os.cpu_count()
    max_workers = 2 * cores + 1

    to_exec = partial(top_level_wrap, fn)

    with futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
        for result, error in executor.map(to_exec, values):
            args, tb = error
            if tb is not None:
                handle_error(args, tb)
            else:
                handle_success(result)


########### End of Utility Code ###########

示例用法 -

######### Start of example usage ###########

import time


@catch
def fail_when_5(val):
    time.sleep(val)
    if val == 5:
        raise Exception("Error - val was 5")
    else:
        return f"No error val is {val}"


def handle_error(args, tb):
    print("args is", args)
    print("TB is", tb)


def top_level(val, val_2, test=None, test2="ok"):
    print(val_2, test, test2)
    return fail_when_5(val)

handle_success = print


if __name__ == "__main__":
    # SHAPE -> ( (args, kwargs), (args, kwargs), ... )
    values = tuple(
        ((x, x + 1), {"test": f"t_{x+2}", "test2": f"t_{x+3}"}) for x in range(10)
    )
    create_processes(top_level, values, handle_error, handle_success)

######### End of example usage ###########