奇数大小的 numpy 数组 send/receive

Question

我想将 numpy 数组内容从所有处理器收集到一个。如果所有数组的大小都相同，它就可以工作。但是，我没有看到对 proc-dependent 大小的数组执行相同任务的自然方法。请考虑以下代码：

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)

我的问题出在 "received" 分配上。我怎么知道要分配的大小是多少？我是否必须首先 send/receive 每个数组大小？

根据下面的建议，我会选择

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: message truncated

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

以下基于您的代码已更正。我要补充一点 执行此收集操作的自然方法是使用 Gatherv(). See http://materials.jeremybejarano.com/MPIwithPython/collectiveCom.html and the documentation of mpi4py 为例。我添加了相应的示例代码。唯一棘手的一点是 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)

奇数大小的 numpy 数组 send/receive

Odd-size numpy arrays send/receive

numpy

mpi4py