Transmettre plusieurs paramètres à concurrent.futures.Executor.map?

Question

Le concurrent.futures.Executor.map prend un nombre variable d'itérables à partir desquels la fonction donnée est appelée. Comment dois-je l'appeler si j'ai un générateur qui produit des tuples qui sont normalement déballés sur place?

Le code suivant ne fonctionne pas car chaque tuple généré est donné comme un argument différent à map:

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

Sans le générateur, les arguments souhaités pour map pourraient ressembler à ceci:

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

Answer 1

Un argument qui se répète, un argument dans c

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

Besoin de déballer les éléments de c, et peut déballer c

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

Besoin de déballer les éléments de c, ne peut pas déballer c

1. Changer f pour prendre un seul argument et déballer l'argument dans la fonction.

2. Si chaque élément de c a un nombre variable de membres, ou si vous appelez f seulement quelques fois :

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

   Cela définit une nouvelle fonction qui déballe chaque élément de c et appelle f. L'utilisation d'un argument par défaut pour f dans le lambda rend f local à l'intérieur du lambda et réduit ainsi le temps de recherche.

3. Si vous avez un nombre fixe d'arguments, et que vous avez besoin d'appeler f de nombreuses fois :

   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))

Où n est le nombre d'arguments pour f. Ceci est adapté de itertools.tee.

Answer 2

Vous devez supprimer le * sur l'appel de la fonction map :

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

Cela appellera f, len(args) fois, où f doit accepter un paramètre.

Si vous voulez que f accepte deux paramètres, vous pouvez utiliser un appel lambda comme ceci :

args = ((a, b) for b in c)
for result in executor.map(lambda p: f(*p), args):   # (*p) fait la partie du déballage
    pass

Answer 3

Supposons que vous avez une fonction qui prend 3 arguments et que les 3 arguments sont dynamiques et changent à chaque appel. Par exemple:

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

Pour appeler cette fonction plusieurs fois en utilisant le threading, je créerais d'abord une liste de tuples où chaque tuple est une version de a, b, c:

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

Nous savons que la fonction map de concurrent.futures acceptera le premier argument comme la fonction cible et le deuxième argument comme la liste d'arguments pour chaque version de la fonction qui sera exécutée. Par conséquent, vous pourriez faire un appel comme ceci:

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

Mais cela vous donnera une erreur indiquant que la fonction attendait 3 arguments mais en a reçu seulement 1. Pour résoudre ce problème, nous créons une fonction d'aide:

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

Maintenant, nous pouvons appeler cette fonction en utilisant un exécuteur comme suit:

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

Cela devrait vous donner les résultats souhaités.

Answer 4

Vous pouvez utiliser le currying pour créer de nouvelles fonctions via la méthode partial en Python

from concurrent.futures import ThreadPoolExecutor
from functools import partial

def some_func(param1, param2):
    # some code

# le curry de some_func avec l'argument 'a' est répété
func = partial(some_func, a)
with ThreadPoolExecutor() as executor:
    executor.map(func, list_of_args):
    ...

Si vous devez passer plus d'un même paramètre, vous pouvez les passer à la méthode partial

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

Answer 5

Voici un extrait de code montrant comment envoyer plusieurs arguments à une fonction avec ThreadPoolExecutor:

import concurrent.futures

def hello(first_name: str, last_name: str) -> None:
    """Affiche un bonjour amical avec le prénom et le nom de famille"""
    print('Bonjour %s %s!' % (first_name, last_name))

def main() -> None:
    """Exemples montrant comment utiliser ThreadPoolExecutor et executor.map
    en envoyant plusieurs arguments à une fonction"""

    # Exemple 1: Envoyer plusieurs arguments en utilisant des tuples
    # Définir des tuples avec des arguments séquentiels à transmettre à hello()
    args_names = (
        ('Bruce', 'Wayne'),
        ('Clark', 'Kent'),
        ('Diana', 'Prince'),
        ('Barry', 'Allen'),
    )
    with concurrent.futures.ThreadPoolExecutor() as executor:
        # En utilisant lambda, décompresse le tuple (*f) en hello(*args)
        executor.map(lambda f: hello(*f), args_names)

    print()

    # Exemple 2: Envoyer plusieurs arguments en utilisant un dictionnaire avec des clés nommées
    # Définir des dictionnaires avec des arguments comme clés à transmettre à 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:
        # En utilisant lambda, décompresse le dictionnaire (**f) en hello(**kwargs)
        executor.map(lambda f: hello(**f), kwargs_names)

if __name__ == '__main__':
    main()