如何删除 openmpi 进程的内存限制？

Question

我正在运行一个具有 mpirun 和 2 个内核的进程，当我在两个进程之间混合值时，它会被杀死。两个进程都使用了大约 15% 的机器内存，即使混合时内存会增加，但仍然应该有足够的内存。所以我假设用于在进程之间传递消息的内存量是有限制的。 我如何知道这个限制是什么以及如何删除它？

当 mpirun 死机时我收到的错误消息是：

File "Comm.pyx", line 864, in mpi4py.MPI.Comm.bcast (src/mpi4py.MPI.c:67787)
File "pickled.pxi", line 564, in mpi4py.MPI.PyMPI_bcast (src/mpi4py.MPI.c:31462)
File "pickled.pxi", line 93, in mpi4py.MPI._p_Pickle.alloc (src/mpi4py.MPI.c:26327)
SystemError: Negative size passed to PyBytes_FromStringAndSize

这是导致错误的代码：

sum_updates_j_k = numpy.zeros((self.col.J_total, self.K), dtype=numpy.float64))        
comm.Reduce(self.updates_j_k, sum_updates_j_k, op=MPI.SUM) 
sum_updates_j_k = comm.bcast(sum_updates_j_k, root=0) 

代码通常可以工作，它只会遇到大量数据的问题，这使得我在进程之间交换的矩阵的大小增加了

Answer 1

罪魁祸首可能是在PyMPI_bcast()的代码中发现的以下几行：

cdef int count = 0
...
if dosend: smsg = pickle.dump(obj, &buf, &count)  # <----- (1)
with nogil: CHKERR( MPI_Bcast(&count, 1, MPI_INT, # <----- (2)
                              root, comm) )
cdef object rmsg = None
if dorecv and dosend: rmsg = smsg
elif dorecv: rmsg = pickle.alloc(&buf, count)
...

这里发生的情况是对象首先使用pickle.dump() 在(1) 序列化，然后在(2) 广播腌制流的长度。

这里有两个问题，它们都与int 用于长度这一事实有关。第一个问题是pickle.dump 内部的整数转换，另一个问题是MPI_INT 用于传输腌制流的长度。这将矩阵中的数据量限制为一定的大小 - 即导致腌制对象不大于 2 GiB（2³¹-1 字节）的大小。任何更大的对象都会导致整数溢出，从而导致 count 中出现负值。

这显然不是 MPI 问题，而是mpi4py 中的错误（或功能？）。

Answer 2

我最近在使用 mpi4py 时遇到了同样的问题。正如 Hristo Iliev 在他的回答中指出的那样，这是一个泡菜问题。

这可以通过使用大写方法comm.Reduce()、comm.Bcast()等which do not resort to pickle来避免，而不是像comm.reduce()这样的小写方法。作为奖励，大写方法也应该更快一些。

实际上，您已经在使用 comm.Reduce()，所以我希望切换到 comm.Bcast() 应该可以解决您的问题 - 它对我有用。

注意：大写方法的语法略有不同，但this tutorial 可以帮助您入门。

例如，而不是：

sum_updates_j_k = comm.bcast(sum_updates_j_k, root=0) 

你会使用：

comm.Bcast(sum_updates_j_k, root=0) 

Answer 3

对于这种情况，有一个可以发送numpy数组部分的函数很有用，例如：

from mpi4py import MPI
import math, numpy
comm = MPI.COMM_WORLD
rank = comm.Get_rank()
def bcast_array_obj(obj = None, dtype = numpy.float64, root = 0):
    """Function for broadcasting of a numpy array object"""
    reporter = 0 if root > 0 else 1
    if rank == root:
        for exp in range(11):
            parts = pow(2, exp)
            err = False
            part_len = math.ceil(len(obj) / parts)
            for part in range(parts):
                part_begin = part * part_len
                part_end = min((part + 1) * part_len, len(obj))
                try:
                    comm.bcast(obj[part_begin: part_end], root = root)
                except:
                    err = True
                err *= comm.recv(source = reporter, tag = 2)
                if err:
                    break
            if err:
                continue
            comm.bcast(None, root = root)
            print('The array was successfully sent in {} part{}'.\
                  format(parts, 's' if parts > 1 else ''))
            return
        sys.stderr.write('Failed to send the array even in 1024 parts')
        sys.stderr.flush()
    else:
        obj = numpy.zeros(0, dtype = dtype)
        while True:
            err = False
            try:
                part_obj = comm.bcast(root = root)
            except:
                err = True
                obj = numpy.zeros(0, dtype = dtype)
            if rank == reporter:
                comm.send(err, dest = root, tag = 2)
            if err:
                continue
            if type(part_obj) != type(None):
                frags = len(obj)
                obj.resize(frags + len(part_obj))
                obj[frags: ] = part_obj
            else:
                break
        return obj

此函数自动确定分解输入数组的最佳零件数。

例如，

if rank != 0:
    z = bcast_array_obj(root = 0)
else:
    z = numpy.zeros(1000000000, dtype = numpy.float64)
    bcast_array_obj(z, root = 0)

输出

The array was successfully sent in 4 parts

Answer 4

显然这是 MPI 本身的问题，而不是 MPI4py 中的问题。保存正在传输的数据大小的实际变量是一个带符号的 32 位整数，对于大约 2GB 的数据，它将溢出为负值。

Maximum amount of data that can be sent using MPI::Send

以前也曾作为 MPI4py 的问题提出过 here。