Передать несколько параметров в concurrent.futures.Executor.map?

Question

concurrent.futures.Executor.map принимает переменное число итераций, из которых вызывается данная функция. Как мне это назвать, если у меня есть генератор, который производит кортежи, которые обычно распаковываются на месте?

Следующее не работает, потому что каждый из сгенерированных кортежей указан в качестве отдельного аргумента для 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. Использование аргумента по умолчанию для f в lambda делает f локальным внутри lambda и, таким образом, сокращает время поиска.

3. Если у вас есть фиксированное количество аргументов, и вам нужно много раз вызывать f:

   from collections import deque
   def itemtee(iterable, n=2):
       def gen(it = iter(iterable), items = deque(), next = next):
           popleft = items.popleft
           extend = items.extend
           while True:
               if not items:
                   extend(next(it))
               yield popleft()
       return [gen()] * n
   
   executor.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 принял два параметра, вы можете использовать лямбда-вызов, например:

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

Вы можете использовать каррирование для создания новой функции с помощью частичного метода в Python

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 4

Для ProcessPoolExecutor.map():

Аналогично map (func, * iterables), за исключением:

итерации собираются сразу, а не лениво;

Функция func выполняется асинхронно, и несколько вызовов функции могут выполняться одновременно.

Попробуйте запустить следующий фрагмент кода в 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