使用MPI和C++從不同節點收集數據

Question

我正在處理一個包含多個從節點和一個主節點的項目。在某些時候，我需要從不同的從節點收集數據（主節點也可以作爲從節點）來掌握節點。數據可以是任何類型，但假設它是unsigned int。這是數據的外觀上從節點：

NODE0：| chunk01 | chunk02 | chunk03 | chunk04 | ....

節點1：| chunk11 | chunk12 | chunk13 | chunk14 | ....

...

節點N：| chunkn1 | chunkn2 | chunkn3 | chunkn4 | ....

數據應全部聚集到NODE0，看起來像這樣：

NODE0：| chunk01 |塊11 | chunk21 | .... | chunkn1 | chunk02 | chunk12 | ... | chunkn2 | ... | chunknm |

這意味着我們從每個節點串連在一起，從每個節點的第一塊在一起，那麼第二塊...

我不知道如何使用MPI_Gatherv來實現這一點，因爲每個chunkij有不同的大小，也每個節點只知道自己的塊大小和開始索引，而不知道其他節點的信息。

我對MPI並不是很熟悉，所以我想知道有沒有API可以將不同大小的數據從各個節點收集到一個節點？

Answer 1

下面是一個可以編輯的例子。這幾乎肯定不是解決問題的最佳方式 - 我需要更多的代碼細節來評論這個問題。我沒有檢查它是否編譯，但如果你修復任何錯別字，我很樂意嘗試修復任何未解決的錯誤。

我也不知道效率對你有多重要 - 這種操作每秒要做幾百次還是每天做一次？如果是後者，那麼這個代碼可能沒問題。我也假設C/C++。

// Populate this on each node from MPI_Comm_rank. 
int myRank; 
// Populate this on each node from MPI_Comm_size. 
int P; 
// Num chunks per core. 
const int M = 4; 

// I'm assuming 0 is the master. 
int masterNodeRank = 0; 

// Populate this. 
// It only needs to have meaningful data on the master node. 
//If master node doesn't have the data, fill with MPI_GATHER. 
int* sizeOfEachChunkOnEachRank[M]; 
// Populate this. 
//It needs to exist on every 'slave' node. 
int sizeOfMyChunks[M]; 

// Assuming you already have this array 
// it should be the contiguous store of each core's data. 
unsigned* myData; 
// This is what we'll gather all the data into on master node only. 
unsigned* gatheredData = new unsigned[totalDataSize]; 
// This array will keep all of the displacements from each sending node. 
int* displacements = new int[P]; 

// This keeps track of how many unsigneds we've received so far. 
int totalCountSoFar = 0; 

// We'll work through all the first chunks on each node at once, then all 
// the second chunks, etc. 
for(int localChunkNum = 0; localChunkNum < M; ++localChunkNum) 
{ 
    // On the receiving node we need to calculate all the displacements 
    // for the received data to go into the array 
    if (myRank == masterNodeRank) 
    { 
    displacements[0] = 0; 

    for(int otherCore = 1; otherCore < P; ++otherCore) 
    { 
     displacements[otherCore] = displacements[otherCore-1] + sizeOfEachChunkOnEachRank[localChunkNum][otherCore-1]; 
    } 
    } 

    // On all cores, we'll need to calculate how far into our local array 
    // to start the sending from.  
    int myFirstIndex = 0; 

    for(int previousChunk=0; previousChunk < localChunkNum; previousChunk++) 
    { 
    myFirstIndex += sizeOfMyChunks[previousChunk]; 
    } 

    // Do the variable gather 
    MPI_Gatherv(&myData[myFirstIndex], // Start address to send from 
       sizeOfMyChunks[localChunkNum], // Number to send 
       MPI_UNSIGNED, // Type to send 
       &gatheredData[totalCountSoFar], // Start address to receive into 
       sizeOfEachChunkOnEachRank[localChunkNum], // Number expected from each core 
       displacements, // Displacements to receive into from each core 
       MPI_UNSIGNED, // Type to receive 
       masterNodeRank, // Receiving core rank 
       MPI_COMM_WORLD); // MPI communicator. 

    // If this is the receiving rank, update the count we've received so far 
    // so that we don't overwrite data the next time we do the gather. 
    // Note that the total received is the displacement to the receive from the 
    // last core + the total received from that core. 
    if(myRank == masterNodeRank) 
    { 
    totalCountSoFar += displacements[P-1] + sizeOfEachChunkOnEachRank[localChunkNum][P-1]; 
    } 
} 

delete[] displacements; 
delete[] gatheredData;