c++ - 跨 MPI 节点共享内存以防止不必要的复制

Question

我有一个算法，在每次迭代中，每个节点都必须计算数组的一部分，其中 x_ 的每个元素都依赖于 x 的所有元素。

x_[i] = some_func(x)//每个 x_[i] 依赖于整个 x

也就是说，每次迭代都取x并计算x_，这将是下一次迭代的新x。

MPI 的一种并行化方法是在节点之间拆分 x_，并在计算完 x_ 后调用 Allgather，因此每个处理器会将其 x_ 发送到所有其他处理器的 x 中的适当位置，然后重复。这是非常低效的，因为每次迭代都需要昂贵的 Allgather 调用，更不用说它需要与节点一样多的 x 拷贝。

我想到了一种不需要复制的替代方法。如果程序在一台机器上运行，共享 RAM，是否可以在节点之间共享 x_(不复制)？也就是说，在计算完 x_ 之后，每个处理器都会使其对其他节点可见，然后这些节点可以将其用作下一次迭代的 x 而无需制作多个拷贝。我可以设计算法，以便没有处理器同时访问相同的 x_，这就是为什么为每个节点制作一个私有(private)拷贝是多余的。

我想我想问的是:我可以通过将数组标记为节点间共享来共享 MPI 中的内存，而不是手动为每个节点制作拷贝吗？ (为简单起见，假设我在一个 CPU 上运行)

Answer 1

您可以使用 MPI-3 中的 MPI_Win_allocate_shared 在节点内共享内存。它提供了一种可移植的方式来使用 Sys5 和 POSIX 共享内存(以及任何类似的东西)。

MPI 函数

以下摘自MPI 3.1 standard .

分配共享内存

MPI_WIN_ALLOCATE_SHARED(size, disp_unit, info, comm, baseptr, win)
IN size size of local window in bytes (non-negative integer)
IN disp_unit local unit size for displacements, in bytes (positive integer)
IN info info argument (handle) IN comm intra-communicator (handle)
OUT baseptr address of local allocated window segment (choice)
OUT win window object returned by the call (handle)

int MPI_Win_allocate_shared(MPI_Aint size, int disp_unit, MPI_Info info, MPI_Comm comm, void *baseptr, MPI_Win *win)

(如果您需要 Fortran 声明，请单击链接)

您使用 MPI_Win_free 释放内存。分配和释放都是集体。这与 Sys5 或 POSIX 不同，但使用户界面更加简单。

查询节点分配情况

为了知道如何对另一个进程的内存执行加载-存储，您需要在本地地址空间中查询该内存的地址。在其他进程的地址空间中共享地址是不正确的(在某些情况下它可能会起作用，但不能假设它会起作用)。

MPI_WIN_SHARED_QUERY(win, rank, size, disp_unit, baseptr)
IN win shared memory window object (handle)
IN rank rank in the group of window win (non-negative integer) or MPI_PROC_NULL
OUT size size of the window segment (non-negative integer)
OUT disp_unit local unit size for displacements, in bytes (positive integer)
OUT baseptr address for load/store access to window segment (choice)

int MPI_Win_shared_query(MPI_Win win, int rank, MPI_Aint *size, int *disp_unit, void *baseptr)

(如果您需要 Fortran 声明，请单击上面的链接)

同步共享内存

MPI_WIN_SYNC(win)
IN win window object (handle)

int MPI_Win_sync(MPI_Win win)

此函数用作对与共享内存窗口关联的数据进行加载-存储访问的内存屏障。

您还可以使用 ISO 语言功能(即 C11 和 C++11 原子提供的功能)或编译器扩展(例如 GCC intrinsics，例如 __sync_synchronize)来获得一致的数据 View 。

同步

如果您已经了解进程间共享内存语义，那么 MPI-3 实现将很容易理解。如果没有，请记住您需要正确同步内存和控制流。前者有 MPI_Win_sync，而现有的 MPI 同步函数如 MPI_Barrier 和 MPI_Send+MPI_Recv 将适用于后者。或者您可以使用 MPI-3 原子学来构建计数器和锁。

示例程序

以下代码来自https://github.com/jeffhammond/HPCInfo/tree/master/mpi/rma/shared-memory-windows ，其中包含 MPI 论坛用于讨论这些功能的语义的共享内存使用示例程序。

该程序演示了通过共享内存实现的单向成对同步。如果您只想创建一个 WORM(一次写入，多次读取)slab，那应该简单得多。

#include <stdio.h>
#include <mpi.h>

/* This function synchronizes process rank i with process rank j
 * in such a way that this function returns on process rank j
 * only after it has been called on process rank i.
 *
 * No additional semantic guarantees are provided.
 *
 * The process ranks are with respect to the input communicator (comm). */

int p2p_xsync(int i, int j, MPI_Comm comm)
{
    /* Avoid deadlock. */
    if (i==j) {
        return MPI_SUCCESS;
    }

    int rank;
    MPI_Comm_rank(comm, &rank);

    int tag = 666; /* The number of the beast. */

    if (rank==i) {
        MPI_Send(NULL, 0, MPI_INT, j, tag, comm);
    } else if (rank==j) {
        MPI_Recv(NULL, 0, MPI_INT, i, tag, comm, MPI_STATUS_IGNORE);
    }

    return MPI_SUCCESS;
}

/* If val is the same at all MPI processes in comm,
 * this function returns 1, else 0. */

int coll_check_equal(int val, MPI_Comm comm)
{
    int minmax[2] = {-val,val};
    MPI_Allreduce(MPI_IN_PLACE, minmax, 2, MPI_INT, MPI_MAX, comm);
    return ((-minmax[0])==minmax[1] ? 1 : 0);
}

int main(int argc, char * argv[])
{
    MPI_Init(&argc, &argv);

    int rank, size;
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Comm_size(MPI_COMM_WORLD, &size);

    int *   shptr = NULL;
    MPI_Win shwin;
    MPI_Win_allocate_shared(rank==0 ? sizeof(int) : 0,sizeof(int),
                            MPI_INFO_NULL, MPI_COMM_WORLD,
                            &shptr, &shwin);

    /* l=local r=remote */
    MPI_Aint rsize = 0;
    int rdisp;
    int * rptr = NULL;
    int lint = -999;
    MPI_Win_shared_query(shwin, 0, &rsize, &rdisp, &rptr);
    if (rptr==NULL || rsize!=sizeof(int)) {
        printf("rptr=%p rsize=%zu \n", rptr, (size_t)rsize);
        MPI_Abort(MPI_COMM_WORLD, 1);
    }

    /*******************************************************/

    MPI_Win_lock_all(0 /* assertion */, shwin);

    if (rank==0) {
        *shptr = 42; /* Answer to the Ultimate Question of Life, The Universe, and Everything. */
        MPI_Win_sync(shwin);
    }
    for (int j=1; j<size; j++) {
        p2p_xsync(0, j, MPI_COMM_WORLD);
    }
    if (rank!=0) {
        MPI_Win_sync(shwin);
    }
    lint = *rptr;

    MPI_Win_unlock_all(shwin);

    /*******************************************************/

    if (1==coll_check_equal(lint,MPI_COMM_WORLD)) {
        if (rank==0) {
            printf("SUCCESS!\n");
        }
    } else {
        printf("rank %d: lint = %d \n", rank, lint);
    }

    MPI_Win_free(&shwin);

    MPI_Finalize();

    return 0;
}