数组结构的MPI-3共享内存

MPI-3 Shared Memory for Array Struct

本文关键字:共享 内存 MPI-3 结构 数组      更新时间:2023-10-16

我有一个简单的c++结构体,基本上包装了一个标准的C数组:

struct MyArray {
    T* data;
    int length;
    // ...
}

,其中T是与floatdouble类似的数字类型。length是数组中元素的个数。通常我的数组非常大(数万到数千万个元素)。

我有一个MPI程序,我想通过MPI 3共享内存公开MyArray的两个实例,比如a_olda_new,作为共享内存对象。上下文是每个MPI排名从a_old读取。然后,每个MPI排名写入a_new的某些索引(每个排名只写入自己的索引集-没有重叠)。最后,a_old = a_new必须在所有等级上设置。a_olda_new大小相同。现在我通过同步(Isend/Irecv)每个等级的更新值与其他等级,使我的代码工作。然而,由于数据访问模式的原因,我没有理由需要引起消息传递的开销,而是可以使用一个共享内存对象,并在a_old = a_new之前放置一个屏障。我认为这会给我更好的表现(虽然请纠正我,如果我错了)。

我有困难找到完整的代码示例做共享内存与MPI 3。大多数站点只提供参考文档或不完整的片段。有人可以走我通过一个简单的和完整的代码的例子,做的事情我试图实现(更新和同步一个数字数组通过MPI共享内存)?我理解了创建共享内存通信器和窗口、设置栅栏等主要概念,但看看一个将所有这些结合在一起的例子真的会帮助我理解。

另外,我应该提到我将只在一个节点上运行我的代码,所以我不需要担心跨节点需要多个共享内存对象的副本;我只需要为运行MPI进程的单个节点复制一份数据。尽管如此,在这种情况下,像OpenMP这样的其他解决方案对我来说是不可行的,因为我有大量的MPI代码,不能为了一个或两个我想分享的数组而重写所有内容。

在MPI-3中使用共享内存相对简单。

首先,使用MPI_Win_allocate_shared分配共享内存窗口:
MPI_Win win;
MPI_Aint size;
void *baseptr;
if (rank == 0)
{
   size = 2 * ARRAY_LEN * sizeof(T);
   MPI_Win_allocate_shared(size, sizeof(T), MPI_INFO_NULL,
                           MPI_COMM_WORLD, &baseptr, &win);
}
else
{
   int disp_unit;
   MPI_Win_allocate_shared(0, sizeof(T), MPI_INFO_NULL,
                           MPI_COMM_WORLD, &baseptr, &win);
   MPI_Win_shared_query(win, 0, &size, &disp_unit, &baseptr);
}
a_old.data = baseptr;
a_old.length = ARRAY_LEN;
a_new.data = a_old.data + ARRAY_LEN;
a_new.length = ARRAY_LEN;

在这里,只有rank 0分配内存。哪个进程分配它并不重要,因为它是共享的。甚至可以让每个进程分配一部分内存,但由于默认情况下分配是连续的,因此这两种方法是等效的。然后,所有其他进程使用MPI_Win_shared_query来查找共享内存块开始的虚拟地址空间中的位置。该地址可能会因秩而异,因此不应该传递绝对指针。

您现在可以简单地从a_new.data分别加载和存储到a_old.data。由于在您的示例中,排序处理的是不相交的内存位置集,因此实际上不需要锁定窗口。使用窗口锁来实现a_old的保护初始化或其他需要同步的操作。你可能还需要显式地告诉编译器不要重新排序代码,并发出一个内存围栏,以便在调用MPI_Barrier()之前完成所有未完成的加载/存储操作。

a_old = a_new代码建议将一个数组复制到另一个数组。相反,您可以简单地交换数据指针,最终交换大小字段。由于只有数组的数据在共享内存块中,交换指针是一个本地操作,即不需要同步。假设两个数组的长度相等:

T *temp;
temp = a_old.data;
a_old.data = a_new.data;
a_new.data = temp;

您仍然需要一个屏障来确保所有其他进程在继续之前已经完成处理。

在最后,简单地释放窗口:

MPI_Win_free(&win);

一个完整的例子(C语言)如下:

#include <stdio.h>
#include <mpi.h>
#define ARRAY_LEN 1000
int main (void)
{
   MPI_Init(NULL, NULL);
   int rank, nproc;
   MPI_Comm_rank(MPI_COMM_WORLD, &rank);
   MPI_Comm_size(MPI_COMM_WORLD, &nproc);
   MPI_Win win;
   MPI_Aint size;
   void *baseptr;
   if (rank == 0)
   {
      size = ARRAY_LEN * sizeof(float);
      MPI_Win_allocate_shared(size, sizeof(int), MPI_INFO_NULL,
                              MPI_COMM_WORLD, &baseptr, &win);
   }
   else
   {
      int disp_unit;
      MPI_Win_allocate_shared(0, sizeof(int), MPI_INFO_NULL,
                              MPI_COMM_WORLD, &baseptr, &win);
      MPI_Win_shared_query(win, 0, &size, &disp_unit, &baseptr);
   }
   printf("Rank %d, baseptr = %pn", rank, baseptr);
   int *arr = baseptr;
   for (int i = rank; i < ARRAY_LEN; i += nproc)
     arr[i] = rank;
   MPI_Barrier(MPI_COMM_WORLD);
   if (rank == 0)
   {
      for (int i = 0; i < 10; i++)
         printf("%4d", arr[i]);
      printf("n");
   }
   MPI_Win_free(&win);
   MPI_Finalize();
   return 0;
}

免责声明:对此持保留态度。我对MPI的RMA的了解还比较薄弱。

这是一个提供描述的代码。在注释中,我对代码做了一些描述。通常它呈现一个动态的RMA窗口,内存必须被分配到该窗口。

MPI_Win_lock_all(0, win)描述来自Open MPI Documentation:

启动对win中所有进程的RMA访问纪元,锁类型为MPI_LOCK_SHARED。在epoch期间,调用进程可以使用RMA操作访问win中所有进程的窗口内存。

在我使用MPI_INFO_NULL的地方,您可以使用MPI_Info对象向MPI提供额外的信息,但这取决于您的内存访问模式。

#include <stdio.h>
#include <stdlib.h>
#include <mpi.h>
typedef struct MyArray {
    double* data;
    int length;
}MyArray;
#define ARRAY_SIZE 10
int main(int argc, char *argv[]) {
    int rank, worldSize, i;
    MPI_Win win;
    MPI_Aint disp;
    MPI_Aint *allProcessDisp;
    MPI_Request *requestArray;
    MyArray myArray;
    MPI_Init(&argc, &argv);
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Comm_size(MPI_COMM_WORLD, &worldSize);
    MPI_Win_create_dynamic(MPI_INFO_NULL, MPI_COMM_WORLD, &win);
    allProcessDisp = malloc(sizeof(MPI_Aint) * worldSize);
    requestArray = malloc(sizeof(MPI_Request) * worldSize);
    for (i = 0; i < worldSize; i++) 
        requestArray[i] = MPI_REQUEST_NULL;
    myArray.data = malloc(sizeof(double) * ARRAY_SIZE);
    myArray.length = ARRAY_SIZE;
    //Allocating memory for each process share window space 
    MPI_Alloc_mem(sizeof(double) * ARRAY_SIZE, MPI_INFO_NULL, &myArray.data);
    for (i = 0; i < ARRAY_SIZE; i++)
        myArray.data[i] = rank;
    //attach the allocating memory to each process share window space 
    MPI_Win_attach(win, myArray.data, sizeof(double) * ARRAY_SIZE);
    MPI_Get_address(myArray.data, &disp);
    if (rank == 0) {
        allProcessDisp[0] = disp;
        //Collect all displacements
        for (i = 1; i < worldSize; i++) {
            MPI_Irecv(&allProcessDisp[i], 1, MPI_AINT, i, 0, MPI_COMM_WORLD, &requestArray[i]);
        }
        MPI_Waitall(worldSize, requestArray, MPI_STATUS_IGNORE);
        MPI_Bcast(allProcessDisp, worldSize, MPI_AINT, 0, MPI_COMM_WORLD);
    }
    else {
        //send displacement 
        MPI_Send(&disp, 1, MPI_AINT, 0, 0, MPI_COMM_WORLD);
        MPI_Bcast(allProcessDisp, worldSize, MPI_AINT, 0, MPI_COMM_WORLD);
    }
    // here you can do RMA operations 
    // Each time you need an RMA operation you start with 
    double otherRankData = -1.0;
    int otherRank = 1;
    if (rank == 0) {
        MPI_Win_lock_all(0, win);
        MPI_Get(&otherRankData, 1, MPI_DOUBLE, otherRank, allProcessDisp[otherRank], 1, MPI_DOUBLE, win);
        // and end with 
        MPI_Win_unlock_all(win);
        printf("Rank 0 : Got %.2f from %dn", otherRankData, otherRank);
    }
    if (rank == 1) {
        MPI_Win_lock_all(0, win);
        MPI_Put(myArray.data, ARRAY_SIZE, MPI_DOUBLE, 0, allProcessDisp[0], ARRAY_SIZE, MPI_DOUBLE, win);
        // and end with 
        MPI_Win_unlock_all(win);
    }
    printf("Rank %d: ", rank);
    for (i = 0; i < ARRAY_SIZE; i++)
        printf("%.2f ", myArray.data[i]);
    printf("n");
    //set rank 0 array
    if (rank == 0) {
        for (i = 0; i < ARRAY_SIZE; i++)
            myArray.data[i] = -1.0;
        printf("Rank %d: ", rank);
        for (i = 0; i < ARRAY_SIZE; i++)
            printf("%.2f ", myArray.data[i]);
        printf("n");
    }
    free(allProcessDisp);
    free(requestArray);
    free(myArray.data);
    MPI_Win_detach(win, myArray.data);
    MPI_Win_free(&win);
    MPI_Finalize();
    return 0;
}