Получение нескольких команд отправки с использованием mpi4py

Question

Как я могу изменить следующий код (адаптированный от http://materials.jeremybejarano.com/MPIwithPython/pointToPoint.html), чтобы каждый экземпляр comm.Send был получен root = 0 и напечатан выход. На данный момент принимается только первая команда отправки.

#passRandomDraw.py 
import numpy 
from mpi4py import MPI 
from mpi4py.MPI import ANY_SOURCE 
import numpy as np 

comm = MPI.COMM_WORLD 
rank = comm.Get_rank() 

if rank == 0: 
    randNum = numpy.zeros(1) 
    print "Process before receiving random numbers" 


else: 
    for i in range(0,np.random.randint(1,10),1): 
     randNum = numpy.zeros(1) 
     randNum = numpy.random.random_sample(1) 
     print "Process", rank, "iteration", i, "drew the number", randNum[0] 
     comm.Send(randNum, dest=0) 


if rank == 0: 
    comm.Recv(randNum, ANY_SOURCE) 
    print "Process", rank, "received the number", randNum[0] 

Answer 1

Если вы не знаете, сколько сообщений вы будете отправлять, то вы должны ввести сообщение, отмечающий конец сообщения. Вы можете использовать это в общих чертах, используя специальные теги. Для того, чтобы избежать предоставлений несовпадения буфера для сообщения завершения, вы можете использовать probe проверку, какие сообщения приходят в

tag_data = 42 
tag_end = 23 

if rank == 0: 
    randNum = numpy.zeros(1) 
    print "Process before receiving random numbers" 
else: 
    for i in range(0,np.random.randint(1,10),1): 
     randNum = numpy.zeros(1) 
     randNum = numpy.random.random_sample(1) 
     print "Process", rank, "iteration", i, "drew the number", randNum[0] 
     comm.Send(randNum, dest=0, tag=tag_data) 
    # send the termination message. Using the lower-case interface is simpler 
    comm.send(None, dest=0, tag=tag_end) 

if rank == 0: 
    # For debugging it might be better to use a list of still active procsses 
    remaining = comm.Get_size() - 1 
    while remaining > 0: 
     s = MPI.Status() 
     comm.Probe(status=s) 
     # make sure we post the right kind of message 
     if s.tag == tag_data: 
      comm.Recv(randNum, s.source, tag=tag_data) 
      print "Process ", s.source, " received the number", randNum[0] 
     elif s.tag == tag_end: 
      # don't need the result here 
      print "Process ", rank, " is done" 
      comm.recv(source=s.source, tag=tag_end) 
      remaining -= 1 

Есть много вариантов этого. Например, если вы знаете, что сообщение является последним сообщением, вы можете объединить сообщение о завершении.

Answer 2

Если каждый процесс знает количество отправляемых сообщений, следующие шаги могут быть разработаны, чтобы решить эту проблему:

1) Снизить количество сообщений для отправки в корневой процесс. Каждый процесс отправляет корню количество сообщений, которые он отправит позже. Эта операция называется редукцией и может быть выполнено с помощью функции comm.reduce(...)

2) Получить все сообщения о процессе 0.

Вот код, основанный на ваших, что должно сделать трюк. Это может быть пробежало mpirun -np 4 python main.py

#passRandomDraw.py 
import numpy 
from mpi4py import MPI 
from mpi4py.MPI import ANY_SOURCE 
import numpy as np 

comm = MPI.COMM_WORLD 
rank = comm.Get_rank() 
size = comm.Get_size() 

#just in case, if numpy.random is seed with 
np.random.seed(np.random.randint(np.iinfo(np.uint32).min,np.iinfo(np.uint32).max)+rank) 

if rank == 0: 
    randNum = numpy.zeros(1) 
    print "Process before receiving random numbers" 
    nb=np.empty((1,),dtype=int) 
    nb0=np.zeros((1,),dtype=int) 
    comm.Reduce([nb0, MPI.INT],[nb, MPI.INT],op=MPI.SUM, root=0) #sums the total number of random number from every process on rank 0, in nb. 
    #print "rank"+str(rank)+" nb "+str(nb) 
else: 
    nb=np.empty((1,),dtype=int) 
    nb[0]=np.random.randint(1,10) 
    #print "rank"+str(rank)+" nb "+str(nb) 
    comm.Reduce([nb, MPI.INT],None,op=MPI.SUM, root=0) 
    for i in range(0,nb[0],1): 
     randNum = numpy.zeros(1) 
     randNum = numpy.random.random_sample(1) 
     print "Process", rank, "iteration", i, "drew the number", randNum[0] 
     comm.Send(randNum, dest=0) 



if rank == 0: 
    for i in range(nb[0]): #receives nb message, each one with its int. 
     comm.Recv(randNum, ANY_SOURCE) 
     print "Process", rank, "received the number", randNum[0] 

Согласно documentation of numpy.random() генератору номера Вихря Мерсена псевдослучайному первоначально посеянный рядом, извлеченным из /dev/urandom (или аналога Windows), если таковые имеются, или семян от часов в противном случае. Следовательно, в последнем случае все процессы могут получать одно и то же семя и генерировать одни и те же случайные числа. Чтобы этого не произошло, я добавил следующую строку:

np.random.seed(np.random.randint(np.iinfo(np.uint32).min,np.iinfo(np.uint32).max)+rank)