发送/接收奇数大小的numpy数组

Question

我想从所有处理器中将numpy数组内容收集到一个。如果所有数组的大小相同，则可以使用。但是我没有看到为proc相关大小的数组执行相同任务的自然方式。请考虑下面的代码：

from mpi4py import MPI 
import numpy 

comm = MPI.COMM_WORLD 
rank = comm.rank 
size = comm.size 

if rank >= size/2: 
    nb_elts = 5 
else: 
    nb_elts = 2 

# create data 
lst = [] 
for i in xrange(nb_elts): 
    lst.append(rank*3+i) 
array_lst = numpy.array(lst, dtype=int) 

# communicate array 
result = [] 
if rank == 0: 
    result = array_lst 
    for p in xrange(1, size): 
     received = numpy.empty(nb_elts, dtype=numpy.int) 
     comm.Recv(received, p, tag=13) 
     result = numpy.concatenate([result, received]) 
else: 
    comm.Send(array_lst, 0, tag=13) 

我的问题是在“收到”的分配。我怎么知道什么是分配的大小？我必须首先发送/接收每个数组的大小吗？

基于下面的建议，我会

data_array = numpy.ones(rank + 3, dtype=int) 
data_array *= rank + 5 
print '[{}] data: {} ({})'.format(rank, data_array, type(data_array)) 

# make all processors aware of data array sizes 
all_sizes = {rank: data_array.size} 
gathered_all_sizes = comm_py.allgather(all_sizes) 
for d in gathered_all_sizes: 
    all_sizes.update(d) 

# prepare Gatherv as described by @francis 
nbsum = 0 
sendcounts = [] 
displacements = [] 
for p in xrange(size): 
    n = all_sizes[p] 
    displacements.append(nbsum) 
    sendcounts.append(n) 
    nbsum += n 

if rank==0: 
    result = numpy.empty(nbsum, dtype=numpy.int) 
else: 
    result = None 

comm_py.Gatherv(data_array,[result, tuple(sendcounts), tuple(displacements), MPI.INT64_T], root=0) 

print '[{}] gathered data: {}'.format(rank, result) 

Answer 1

走在你粘贴代码，二者Send()和Recv()发送nb_elts元素。问题是，nb_elts不是每个过程一样...因此，收到不匹配已发送元素的数量和计划项目的数量抱怨：

mpi4py.MPI.Exception：MPI_ERR_TRUNCATE ：消息被截断

为防止出现这种情况，根进程必须计算其他进程发送的项目数。因此，在循环for p in xrange(1, size)中，必须根据p而不是rank来计算nb_elts。

以下基于您的代码已得到更正。我想补充一点，执行这种采集操作的自然方式是使用Gatherv()。例如，参见http://materials.jeremybejarano.com/MPIwithPython/collectiveCom.html和。我添加了相应的示例代码。唯一难点是numpy.int是64位长。因此，Gatherv()使用MPI类型MPI_DOUBLE。

from mpi4py import MPI 
import numpy 

comm = MPI.COMM_WORLD 
rank = comm.rank 
size = comm.size 

if rank >= size/2: 
    nb_elts = 5 
else: 
    nb_elts = 2 

# create data 
lst = [] 
for i in xrange(nb_elts): 
    lst.append(rank*3+i) 
array_lst = numpy.array(lst, dtype=int) 

# communicate array 
result = [] 
if rank == 0: 
    result = array_lst 
    for p in xrange(1, size): 

     if p >= size/2: 
      nb_elts = 5 
     else: 
      nb_elts = 2 

     received = numpy.empty(nb_elts, dtype=numpy.int) 
     comm.Recv(received, p, tag=13) 
     result = numpy.concatenate([result, received]) 
else: 
    comm.Send(array_lst, 0, tag=13) 

if rank==0: 
    print "Send Recv, result= "+str(result) 

#How to use Gatherv: 
nbsum=0 
sendcounts=[] 
displacements=[] 

for p in xrange(0,size): 
    displacements.append(nbsum) 
    if p >= size/2: 
      nbsum+= 5 
      sendcounts.append(5) 
    else: 
      nbsum+= 2 
      sendcounts.append(2) 

if rank==0: 
    print "nbsum "+str(nbsum) 
    print "sendcounts "+str(tuple(sendcounts)) 
    print "displacements "+str(tuple(displacements)) 
print "rank "+str(rank)+" array_lst "+str(array_lst) 
print "numpy.int "+str(numpy.dtype(numpy.int))+" "+str(numpy.dtype(numpy.int).itemsize)+" "+str(numpy.dtype(numpy.int).name) 

if rank==0: 
    result2=numpy.empty(nbsum, dtype=numpy.int) 
else: 
    result2=None 

comm.Gatherv(array_lst,[result2,tuple(sendcounts),tuple(displacements),MPI.DOUBLE],root=0) 

if rank==0: 
    print "Gatherv, result2= "+str(result2)