adapter multiprocessing Pool à mpi4py

Question

J'utilise multi-traitement Pool pour exécuter une simulation parallélisée en Python et cela fonctionne bien dans un ordinateur avec plusieurs cœurs. Maintenant, je veux exécuter le programme sur un cluster en utilisant plusieurs nœuds. Je suppose que le multitraitement ne peut pas s'appliquer sur la mémoire distribuée. Mais mpi4py semble une bonne option. Alors, quelle est l'équivalence de mpi4py plus simple à ces codes:

from multiprocessing import Pool 

pool = Pool(processes=16) 

pool.map(functionName,parameters_list) 

Answer 1

Il y a une classe MPIPool mis en œuvre here.

Pour un exemple de comment je l'utilise, consultez ce gist sur GitHub.

Answer 2

J'utilise le code suivant pour être équivalent à multiprocessing.Pool. Il n'a pas encore été testé intensivement, mais il semble très bien fonctionner:

from functools import partial 
function = partial(...) # Store all fixed parameters this way if needed 

if use_MPI: 
    arguments = range(num_runs) 
    run_data = None 

    # mpi4py 
    comm = MPI.COMM_SELF.Spawn(sys.executable, args=['MPI_slave.py'], maxprocs=num_runs) # Init 
    comm.bcast(function, root=MPI.ROOT)  # Equal for all processes 
    comm.scatter(arguments, root=MPI.ROOT) # Different for each process 
    comm.Barrier()       # Wait for everything to finish... 
    run_data = comm.gather(run_data, root=MPI.ROOT) # And gather everything up 
else:   
    # multiprocessing 
    p = Pool(multiprocessing.cpu_count()) 
    run_data = p.map(function, range(num_runs)) 

Il utilise ensuite un fichier séparé « MPI_slave.py »:

from mpi4py import MPI 
# import the function you actually pass to this file here!!! 
comm = MPI.COMM_SELF.Get_parent() 
size = comm.Get_size() 
rank = comm.Get_rank() 

def runSlaveRun(): 
    function = None 
    options = None 
    # print("Process {}/{} reporting for duty!".format(rank, size)) 

    function = comm.bcast(function, root=0) 
    arguments = comm.scatter(options, root=0) 
    results = function(arguments) 
    comm.Barrier() 
    comm.gather(results, root=0) 
    comm.Disconnect() 

if __name__ == '__main__': 
    runSlaveRun() 

Answer 3

Il y a un vieux paquet de mine qui est construit sur mpi4py qui permet une carte parallèle fonctionnelle pour les travaux MPI. Il n'est pas construit pour la vitesse - il a été construit pour activer une carte parallèle MPI de l'interpréteur sur un cluster de calcul (c'est-à-dire sans avoir besoin de lancer depuis la ligne de commande mpiexec). Pour l'essentiel:

>>> from pyina.launchers import MpiPool, MpiScatter 
>>> pool = MpiPool() 
>>> jobs = MpiScatter() 
>>> def squared(x): 
... return x**2 
... 
>>> pool.map(squared, range(4)) 
[0, 1, 4, 9] 
>>> jobs.map(sqaured, range(4)) 
[0, 1, 4, 9] 

Exhibant la stratégie « pool des travailleurs » et la stratégie « dispersion-réunion » de la distribution des emplois aux travailleurs. Bien sûr, je ne l'utiliserais pas pour un si petit travail comme squared parce que le temps de génération du monde MPI est vraiment très élevé (beaucoup plus élevé que la configuration d'un multiprocessingPool). Cependant, si vous avez un gros travail à exécuter, comme vous le feriez normalement sur un cluster en utilisant MPI, puis pyina peut être un gros avantage pour vous.

Cependant, le vrai avantage d'utiliser pyina est qu'il peut non seulement générer des tâches avec MPI, mais il peut générer des tâches dans un planificateur. pyina comprend et résume la syntaxe de lancement pour plusieurs planificateurs.

Un appel typique à une carte pyina en utilisant un programmateur va comme ceci:

>>> # instantiate and configure a scheduler 
>>> from pyina.schedulers import Torque 
>>> config = {'nodes'='32:ppn=4', 'queue':'dedicated', 'timelimit':'11:59'} 
>>> torque = Torque(**config) 
>>> 
>>> # instantiate and configure a worker pool 
>>> from pyina.launchers import Mpi 
>>> pool = Mpi(scheduler=torque) 
>>> 
>>> # do a blocking map on the chosen function 
>>> pool.map(pow, [1,2,3,4], [5,6,7,8]) 
[1, 64, 2187, 65536] 

Plusieurs configurations communes sont disponibles des cartes pré-configurées. Ce qui suit est identique à l'exemple ci-dessus:

>>> # instantiate and configure a pre-configured worker pool 
>>> from pyina.launchers import TorqueMpiPool 
>>> config = {'nodes'='32:ppn=4', 'queue':'dedicated', 'timelimit':'11:59'} 
>>> pool = TorqueMpiPool(**config) 
>>> 
>>> # do a blocking map on the chosen function 
>>> pool.map(pow, [1,2,3,4], [5,6,7,8]) 
[1, 64, 2187, 65536] 

pyina a besoin d'être retapé, en ce qu'il est encore python2.7 et qu'il n'a pas eu une sortie dans plusieurs années ... mais il a été tenu à jour (sur GitHub) et est capable de "faire le travail" pour moi en cours d'exécution sur les clusters informatiques à grande échelle au cours des 10 dernières années - en particulier lorsqu'il est couplé avec pathos (qui fournit ssh tunneling et une interface unifiée pour les cartes multiprocessing et ParallelPython). pyina n'utilise pas encore la mémoire partagée, mais fait assez bien l'informatique parallèle embarrassante fonctionnelle. Les interactions avec le planificateur sont plutôt bonnes en général, mais peuvent être un peu brouillon pour plusieurs cas de défaillance - et les cartes non bloquantes nécessitent beaucoup de travail.Cela étant dit, il fournit une interface assez utile pour exécuter des tâches parallèles embarrassantes sur un cluster avec MPI.

Get pyina (et pathos) ici: https://github.com/uqfoundation