使用 MPI_Send 和 MPI_Recv 从所有处理器发送到根。

Question

我无法从所有处理器收集一些数据到根，这是我想做的一个例子：

我在每个处理器中都有一对（实际上它们是边缘），理想情况下我想将它们发送到根，或者如果没有办法我可以发送它们相应的索引（一个数字而不是一对。

例如：

Processor 0: sends {(0,5), (1,6)} to root, or it sould send {5,17}
Processor 1: sends {(2,3)} to root, or it sould send {14}
Processor 2: sends {} to root, or it sould send {}
Processor 3: sends {(4,0)} to root, or it sould send {20}

我想知道存储对或数字并发送和接收它们的最佳方式是什么。理想情况下，我更喜欢将它们存储在二维向量中，因为从一开始我就不知道我需要多少 space，并在二维向量中再次接收它们。我知道这可能不可能或可能非常复杂。

这是我正在寻找但不知道如何在 MPI 中实现的程序的伪代码。

vector<vector<int > >allSelectedEdges;
vector<vector<int > >selectedEdgesLocal; 
int edgeCount=0;   
if(my_rank!=0){                          
        for(int i = 0; i < rows; ++i)
            for(int j = 0; j < nVertex; ++j)
                if (some conditions)
                {
                    vector<int> tempEdge;
                    tempEdge.push_back(displs[my_rank]+i);
                    tempEdge.push_back(j);
                    selectedEdgesLocal.push_back(tempEdge);
                    edgeCount++;
                }
        }
        "send selectedEdgesLocal to root"
}else
{
      "root recieve sselectedEdgesLocal and store in allSelectedEdges"
}

我也考虑过 MPI_Gatherv，但似乎没有用。灵感来自 here

vector<vector<int > >selectedEdgesLocal; 
int edgeCount=0;                            
for(int i = 0; i < rows; ++i)
    for(int j = 0; j < nVertex; ++j)
        if (some conditions)
        {
            vector<int> tempEdge;
            tempEdge.push_back(displs[my_rank]+i);
            tempEdge.push_back(j);
            selectedEdgesLocal.push_back(tempEdge);
            edgeCount++;
         }
int NumdgesToAdd;
MPI_Reduce(&edgeCount, &NumdgesToAdd, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);

vector<vector<int > > allSelectedEdges(NumdgesToAdd);

int rcounts[comm_size];
int rdisp[comm_size];
int sumE=0;
for(int i=0; i<comm_size; ++i) {
    rcounts[i] = edgeCount;
    rdisp[i]=sumE;
    sumE+=edgeCount;
}

MPI_Gatherv(&selectedEdgesLocal.front(), rcounts[my_rank], MPI_INT, &allSelectedEdges.front(), rcounts, rdisp, MPI_INT, 0, MPI_COMM_WORLD);

Answer 1

你应该为此使用 Gather。问题是每个进程都有不同数量的值要发送到根。因此，您可以确定要发送的最大值数，并让每个进程发送那么多值（例如，未使用的值为 NAN），或者执行 Gilles Gouaillardet 在评论中建议的操作，并使用两个步骤：

让每个进程计算它需要发送的值的数量。将这些计数收集到根作为 rcounts.
使用 Gather 收集值——既然根进程知道 rcounts，它可以轻松计算 rdisp 作为 rcounts 的累加和。

"use a fixed maximum number of values and fill unused slots with NAN"比较简单，如果数据总量较小，效果会很好。如果数据总量很大并且每个进程发送的值的数量差异很大，two-step 解决方案可能更有效。

Answer 2

我按如下方式更新我的代码，现在它可以正常工作了。

vector <int> selectedEdgesIndicesLocal;

    int edgeCount=0;
    for(int i = 0; i < rows; ++i)
        for(int j = 0; j < nVertex; ++j)
            if (some condistions)
            {
                int index=...;
                selectedEdgesIndicesLocal.push_back(index);
                edgeCount++;
            }

    int NumEdgesToAdd;
    MPI_Reduce(&edgeCount, &NumEdgesToAdd, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);

    int *edgeCountsInRoot;
    if (my_rank == 0)edgeCountsInRoot = (int *)malloc(comm_size * sizeof(int));
    MPI_Gather(&edgeCount, 1, MPI_INT, edgeCountsInRoot, 1, MPI_INT, 0,MPI_COMM_WORLD);

    int *allSelectedIndicesEdges;
    if (my_rank == 0)allSelectedIndicesEdges = (int *)malloc(NumEdgesToAdd * sizeof(int));

    int * edgeCounts, *edgeDisp;

    cout<<edgeCount<<endl;
    if (my_rank==0) {
        edgeCounts= (int *)malloc(comm_size * sizeof(int));
        edgeDisp= (int *)malloc(comm_size * sizeof(int));
        int edgeSum=0;
        for(int i=0; i<comm_size; ++i) {
            edgeCounts[i] = edgeCountsInRoot[i];
            edgeDisp[i]=edgeSum;
            edgeSum+=edgeCountsInRoot[i];
        }
    }
    MPI_Gatherv(&selectedEdgesIndicesLocal.front(), edgeCount, MPI_INT, &allSelectedIndicesEdges[0], edgeCounts,edgeDisp, MPI_INT, 0, MPI_COMM_WORLD);

使用 MPI_Send 和 MPI_Recv 从所有处理器发送到根。

Sending from all processor to root, using MPI_Send and MPI_Recv.

c++

distributed-computing

mpi

sendmessage

2d-vector