Paralléliser une boucle for en Python

Existe-t-il des outils en Python qui ressemblent au parfor de Matlab? J'ai trouvé ce fil , mais il a quatre ans. Je pensais que quelqu'un ici pourrait avoir une expérience plus récente.

Voici un exemple du type de chose que je voudrais paralléliser:

X = np.random.normal(size=(10, 3))
F = np.zeros((10, ))
for i in range(10):
    F[i] = my_function(X[i,:])

où my_functionprend une ndarraytaille (1,3)et retourne un scalaire.

Au moins, j'aimerais utiliser plusieurs cœurs simultanément - comme parfor. En d'autres termes, supposons un système de mémoire partagée avec 8 à 16 cœurs.

Joblib fait ce que vous voulez. Le modèle d'utilisation de base est:

from joblib import Parallel, delayed

def myfun(arg):
     do_stuff
     return result

results = Parallel(n_jobs=-1, verbose=verbosity_level, backend="threading")(
             map(delayed(myfun), arg_instances))

où arg_instancesest la liste des valeurs pour lesquelles myfunest calculé en parallèle. La principale restriction est que cela myfundoit être une fonction de niveau supérieur. Le backendparamètre peut être "threading"ou "multiprocessing".

Vous pouvez transmettre des paramètres communs supplémentaires à la fonction parallélisée. Le corps de myfunpeut également faire référence à des variables globales initialisées, les valeurs qui seront disponibles pour les enfants.

Les args et les résultats peuvent être à peu près tout avec le backend de threading, mais les résultats doivent être sérialisables avec le backend à multi-traitements.

Dask offre également des fonctionnalités similaires. Cela peut être préférable si vous travaillez avec des données non essentielles ou si vous essayez de paralléliser des calculs plus complexes.

Ce que vous cherchez, c'est Numba , qui peut automatiquement paralléliser une boucle for. De leur documentation

from numba import jit, prange

@jit
def parallel_sum(A):
    sum = 0.0
    for i in prange(A.shape[0]):
        sum += A[i]

    return sum

Sans supposer quelque chose de spécial sur le my_functionchoix multiprocessing.Pool().map()est une bonne idée pour paralléliser de telles boucles simples. joblib, dask, Des mpicalculs ou numbacomme proposé dans d' autres réponses ne regarde pas apporter aucun avantage pour les cas d'utilisation et ajouter des dépendances inutiles (pour résumer leur surpuissance). L'utilisation de threads comme proposé dans une autre réponse n'est probablement pas une bonne solution, car vous devez être familier avec l'interaction GIL de votre code, sinon votre code devrait principalement effectuer des entrées / sorties.

Cela dit numbapourrait être une bonne idée pour accélérer le code python pur séquentiel, mais j'estime que cela sort du cadre de la question.

import multiprocessing
import numpy as np

if __name__ == "__main__":
   #the previous line is necessary under windows to not execute 
   # main module on each child under windows

   X = np.random.normal(size=(10, 3))
   F = np.zeros((10, ))

   pool = multiprocessing.Pool(processes=16)
   # if number of processes is not specified, it uses the number of core
   F[:] = pool.map(my_function, (X[i,:] for i in range(10)) )

Il y a cependant quelques réserves (qui ne devraient pas affecter la plupart des applications):

• sous Windows, il n’existe pas de prise en charge, donc un interprète avec le module principal est lancé au démarrage de chaque enfant; il peut donc y avoir une surcharge (et c’est la raison de la if __name__ == "__main__"
• Les arguments et les résultats de my_function sont décapés et non décodés. Il peut s'agir d'un temps système trop important. Voir la réponse à cette question pour la réduire . Https://stackoverflow.com/a/37072511/128629 . Cela rend également les objets non sélectionnables inutilisables
• my_functionne devrait pas dépendre d'états partagés comme la communication avec des variables globales car les états ne sont pas partagés entre processus. Les fonctions pures (fonctions au sens mathématique) sont des exemples de fonctions qui ne partagent pas d'états

Mon impression de Parfor est que MATLAB encapsule les détails de l'implémentation, de sorte qu'il pourrait utiliser à la fois le parallélisme de mémoire partagée (comme vous le souhaitez) et le parallélisme de mémoire distribuée (si vous utilisez un serveur d'informatique distribuée MATLAB ).

Si vous voulez un parallélisme de mémoire partagée et que vous exécutez une sorte de boucle de tâche parallèle, le paquet de bibliothèque standard multitraitement est probablement ce que vous voulez, peut-être avec un joli front-end, comme joblib , comme mentionné dans l'article de Doug. La bibliothèque standard ne va pas disparaître, elle est maintenue et présente donc un risque faible.

Il existe également d'autres options, telles que les fonctionnalités parallèles de Parallel Python et IPython . Un coup d’œil rapide à Parallel Python me fait penser que c’est plus proche de l’esprit de parfor, dans la mesure où la bibliothèque encapsule les détails du cas distribué, mais le coût de cette opération est que vous devez adopter leur écosystème. Le coût d'utilisation d'IPython est similaire. vous devez adopter la façon de faire IPython, qui peut ne pas valoir la peine.

Si vous vous souciez de la mémoire distribuée, je vous recommande mpi4py . Lisandro Dalcin fait un excellent travail et mpi4py est utilisé dans les wrappers PETSc Python. Je ne pense donc pas que cela disparaisse de si tôt. Comme le multitraitement, il s’agit d’une interface de niveau (inférieur) avec le parallélisme que parforfor, mais qui est susceptible de durer un certain temps.

Avant de rechercher un outil "boîte noire" pouvant être utilisé pour exécuter en parallèle des fonctions python "génériques", je suggérerais d'analyser comment il my_function()est possible de paralléliser à la main.

Premièrement, comparez le temps d’exécution de la surcharge de la boucle my_function(v)python for: [C] Les forboucles Python sont assez lentes, le temps d’attente my_function()pourrait donc être négligeable.

>>> timeit.timeit('pass', number=1000000)
0.01692986488342285
>>> timeit.timeit('for i in range(10): pass', number=1000000)
0.47521495819091797
>>> timeit.timeit('for i in xrange(10): pass', number=1000000)
0.42337894439697266

Deuxième vérification s’il existe une simple implémentation vectorielle my_function(v)ne nécessitant pas de boucles:F[:] = my_vector_function(X)

(Ces deux premiers points sont assez triviaux, pardonnez-moi si je les ai mentionnés ici pour des raisons de complétude.)

Le troisième point, et le plus important, du moins pour les implémentations CPython, consiste à vérifier si la my_functionplupart du temps est passé à l' intérieur ou à l' extérieur du verrou d'interpréteur global , ou GIL . Si vous passez du temps en dehors de GIL, vous devez utiliser le threadingmodule de bibliothèque standard . ( Voici un exemple). BTW, on pourrait penser à écrire my_function()comme une extension C juste pour libérer le GIL.

Enfin, si my_function()ne libère pas le GIL, on pourrait utiliser le multiprocessingmodule .

Références: Documents Python sur l'exécution simultanée et intro numpy / scipy sur le traitement parallèle .

Tu peux essayer Julia. C'est assez proche de Python et a beaucoup de constructions MATLAB. La traduction est ici:

F = @parallel (vcat) for i in 1:10
    my_function(randn(3))
end

Cela rend les nombres aléatoires en parallèle aussi, et concatène simplement les résultats à la fin de la réduction. Cela utilise le multitraitement (vous devez donc addprocs(N)ajouter des processus avant de l'utiliser et cela fonctionne également sur plusieurs nœuds d'un HPC, comme indiqué dans cet article de blog ).

Vous pouvez également utiliser à la pmapplace:

F = pmap((i)->my_function(randn(3)),1:10)

Si vous voulez un parallélisme entre les threads, vous pouvez utiliser Threads.@threads(assurez-vous toutefois de rendre l'algorithme thread-safe). Avant d'ouvrir Julia, définissez la variable d'environnement JULIA_NUM_THREADS, puis procédez comme suit:

Ftmp = [Float64[] for i in Threads.nthreads()]
Threads.@threads for i in 1:10
    push!(Ftmp[Threads.threadid()],my_function(randn(3)))
end
F = vcat(Ftmp...)

Ici, je crée un tableau séparé pour chaque thread, afin qu’ils ne se contredisent pas lors de l’ajout au tableau, puis concaténent les tableaux par la suite. Les threads étant assez nouveaux, nous utilisons maintenant directement les threads, mais je suis sûr que des réductions et des maps threadées seront ajoutées comme dans le cas du multitraitement.

Je recommande d'utiliser les fonctions parallèles et différées de la bibliothèque joblib, d'utiliser le module "tempfile" pour créer de la mémoire partagée temporaire pour de grands tableaux, les exemples et l'utilisation peuvent être trouvés ici https://pythonhosted.org/joblib/parallel.html