MPI 集合通信函數 MPI_Reduce()，MPI_Allreduce()，MPI_Bcast()，MPI_Scatter()，MPI_Gather()，MPI_Allgather()，MPI_Scan()，MPI_Reduce_Scatter()

? 八個常用的集合通信函數

? 規約函數 MPI_Reduce()，將通信子內各進程的同一個變量參與規約計算，并向指定的進程輸出計算結果

● 函數原型

MPI_METHOD MPI_Reduce(_In_range_(!= , recvbuf) _In_opt_ const void* sendbuf, // 指向輸入數據的指針_When_(root != MPI_PROC_NULL, _Out_opt_) void* recvbuf, // 指向輸出數據的指針，即計算結果存放的地方_In_range_(>= , 0) int count, // 數據尺寸，可以進行多個標量或多個向量的規約_In_ MPI_Datatype datatype, // 數據類型_In_ MPI_Op op, // 規約操作類型_mpi_coll_rank_(root) int root, // 目標進程號，存放計算結果的進程_In_ MPI_Comm comm // 通信子 );

● 使用范例

{int size, rank, data, dataCollect;MPI_Init(NULL, NULL);MPI_Comm_size(MPI_COMM_WORLD, &size);MPI_Comm_rank(MPI_COMM_WORLD, &rank);data = rank;// 參與計算的數據MPI_Reduce((void *)&data, (void *)&dataCollect, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);// 所有的進程都要調用，而不是只在目標進程中調用MPI_Finalize(); }

● 操作類型，定義于 mpi.h

#define MPI_OP_NULL ((MPI_Op)0x18000000)#define MPI_MAX ((MPI_Op)0x58000001) #define MPI_MIN ((MPI_Op)0x58000002) #define MPI_SUM ((MPI_Op)0x58000003) #define MPI_PROD ((MPI_Op)0x58000004) #define MPI_LAND ((MPI_Op)0x58000005)// 邏輯與 #define MPI_BAND ((MPI_Op)0x58000006)// 按位與 #define MPI_LOR ((MPI_Op)0x58000007) #define MPI_BOR ((MPI_Op)0x58000008) #define MPI_LXOR ((MPI_Op)0x58000009) #define MPI_BXOR ((MPI_Op)0x5800000a) #define MPI_MINLOC ((MPI_Op)0x5800000b)// 求最小值所在位置 #define MPI_MAXLOC ((MPI_Op)0x5800000c)// 求最大值所在位置 #define MPI_REPLACE ((MPI_Op)0x5800000d)

? 規約并廣播函數 MPI_Allreduce()，在計算規約的基礎上，將計算結果分發到每一個進程中，相比于 MPI_Reduce()，只是少了一個 root 參數。除了簡單的先規約再廣播的方法，書中介紹了蝶形結構全局求和的方法。

● 函數原型

_Pre_satisfies_(recvbuf != MPI_IN_PLACE) MPI_METHOD MPI_Allreduce(_In_range_(!= , recvbuf) _In_opt_ const void* sendbuf,_Out_opt_ void* recvbuf,_In_range_(>= , 0) int count,_In_ MPI_Datatype datatype,_In_ MPI_Op op,_In_ MPI_Comm comm );

● 使用范例

{int size, rank, data, dataCollect;MPI_Init(NULL, NULL);MPI_Comm_size(MPI_COMM_WORLD, &size);MPI_Comm_rank(MPI_COMM_WORLD, &rank);data = rank;MPI_Reduce((void *)&data, (void *)&dataCollect, 1, MPI_INT, MPI_SUM, MPI_COMM_WORLD);// 所有的進程都要調用MPI_Finalize(); }

? 廣播函數 MPI_Bcast()，將某個進程的某個變量的值廣播到該通信子中所有進程的同名變量中

● 函數原型

MPI_METHOD MPI_Bcast(_Pre_opt_valid_ void* buffer, // 指向輸入 / 輸出數據的指針_In_range_(>= , 0) int count, // 數據尺寸_In_ MPI_Datatype datatype, // 數據類型_mpi_coll_rank_(root) int root, // 廣播源進程號_In_ MPI_Comm comm // 通信子 );

● 使用范例

{int size, rank, data;MPI_Init(NULL, NULL);MPI_Comm_size(MPI_COMM_WORLD, &size);MPI_Comm_rank(MPI_COMM_WORLD, &rank);data = rank;MPI_Bcast((void *)&data, 1, MPI_INT, 0, MPI_COMM_WORLD);// 所有的進程都要調用，調用后所有 data 均被廣播源進程的值覆蓋MPI_Finalize(); }

? 散射函數 MPI_Scatter()，將向量數據分段發送到各進程中

● 函數原型和宏定義

_Pre_satisfies_(sendbuf != MPI_IN_PLACE) MPI_METHOD MPI_Scatter(_In_range_(!= , recvbuf) _In_opt_ const void* sendbuf, // 指向需要分發的數據的指針_In_range_(>= , 0) int sendcount, // 分發到每一個進程的數據量，注意不是分發的數據總量_In_ MPI_Datatype sendtype, // 分發數據類型_When_(root != MPI_PROC_NULL, _Out_opt_) void* recvbuf, // 指向接收的數據的指針_In_range_(>= , 0) int recvcount, // 接受數據量，不小于上面分發到每一個進程的數據量_In_ MPI_Datatype recvtype, // 接收數據類型_mpi_coll_rank_(root) int root, // 分發數據源進程號_In_ MPI_Comm comm // 通信子 );// 宏定義，mpi.h #define MPI_IN_PLACE ((void*)(MPI_Aint)-1 // MPI_Aint 為 __int64 類型，表示地址

? 聚集函數 MPI_Gather()，將各進程中的向量數據分段聚集到一個進程的大向量中

● 函數原型

_Pre_satisfies_(recvbuf != MPI_IN_PLACE) MPI_METHOD MPI_Gather(_In_opt_ _When_(sendtype == recvtype, _In_range_(!= , recvbuf)) const void* sendbuf,// 指向需要聚集的數據的指針_In_range_(>= , 0) int sendcount, // 每個進程中進行聚集的數據量，不是聚集的數據總量_In_ MPI_Datatype sendtype, // 發送數據類型_When_(root != MPI_PROC_NULL, _Out_opt_) void* recvbuf, // 指向接收數據的指針_In_range_(>= , 0) int recvcount, // 從每個進程接收的接收數據量，不是聚集的數據總量_In_ MPI_Datatype recvtype, // 接收數據類型_mpi_coll_rank_(root) int root, // 聚集數據匯進程號_In_ MPI_Comm comm // 通信子 );

● 函數 MPI_Scatter() 和 MPI_Gather() 的范例

{const int dataSize = 8 * 8;const int localSize = 8;int globalData[dataSize], localData[localSize], globalSum, i, comSize, comRank;MPI_Init(&argc, &argv);MPI_Comm_size(MPI_COMM_WORLD, &comSize);MPI_Comm_rank(MPI_COMM_WORLD, &comRank);if (comRank == 0) // 初始化for (i = 0; i < dataSize; globalData[i] = i, i++);for (i = 0; i < localSize; localData[i++] = 0);MPI_Scatter((void *)&globalData, localSize, MPI_INT, (void *)&localData, localSize, MPI_INT, 0, MPI_COMM_WORLD); // 分發數據for (i = 0; i < localSize; localData[i++]++);MPI_Barrier(MPI_COMM_WORLD); // 進程同步MPI_Gather((void *)&localData, localSize, MPI_INT, (void *)&globalData, localSize, MPI_INT, 0, MPI_COMM_WORLD); // 聚集數據for (i = globalSum = 0; i < dataSize; globalSum += globalData[i++]);if (comRank == 0)printf("\nSize = %d, Rank = %d, result = %d\n", comSize, comRank, globalSum);MPI_Finalize();return 0;// 輸出結果：Size = 8, Rank = 0, result = 2080，表示 0 + 1 + 2 + …… + 63 }

? 全局聚集函數 MPI_Allgather()，將各進程的向量數據聚集為一個大向量，并分發到每個進程中，相當于各進程同步該大向量的各部分分量。相比于 MPI_Gather()，只是少了一個 root 參數。

● 函數原型

_Pre_satisfies_(recvbuf != MPI_IN_PLACE) MPI_METHOD MPI_Allgather(_In_opt_ _When_(sendtype == recvtype, _In_range_(!= , recvbuf)) const void* sendbuf,_In_range_(>= , 0) int sendcount,_In_ MPI_Datatype sendtype,_Out_opt_ void* recvbuf,_In_range_(>= , 0) int recvcount,_In_ MPI_Datatype recvtype,_In_ MPI_Comm comm );

● 函數 MPI_Scatter() 和 MPI_Allgather() 的范例，相當于從上面的范例中修改了一部分

{const int dataSize = 8 * 8;const int localSize = 8;int globalData[dataSize], localData[localSize], globalSum, i, comSize, comRank;MPI_Init(&argc, &argv);MPI_Comm_size(MPI_COMM_WORLD, &comSize);MPI_Comm_rank(MPI_COMM_WORLD, &comRank);for (i = 0; i < dataSize; globalData[i] = i, i++);// 改動for (i = 0; i < localSize; localData[i++] = 0);MPI_Scatter((void *)&globalData, localSize, MPI_INT, (void *)&localData, localSize, MPI_INT, 0, MPI_COMM_WORLD); // 分發數據for (i = 0; i < localSize; localData[i++]++);MPI_Barrier(MPI_COMM_WORLD);MPI_Allgather((void *)&localData, localSize, MPI_INT, (void *)&globalData, localSize, MPI_INT, MPI_COMM_WORLD); // 聚集數據，改動for (i = globalSum = 0; i < dataSize; globalSum += globalData[i++]);printf("\nSize = %d, rank = %d, result = %d\n", comSize, comRank, globalSum);// 改動MPI_Finalize();return 0;// 輸出結果，八個進程亂序輸出 2080 }

? 前綴和函數 MPI_Scan()，將通信子內各進程的同一個變量參與前綴規約計算，并將得到的結果發送回每個進程，使用與函數 MPI_Reduce() 相同的操作類型

● 函數原型

_Pre_satisfies_(recvbuf != MPI_IN_PLACE) MPI_METHOD MPI_Scan(_In_opt_ _In_range_(!= , recvbuf) const void* sendbuf, // 指向參與規約數據的指針_Out_opt_ void* recvbuf, // 指向接收規約結果的指針_In_range_(>= , 0) int count, // 每個進程中參與規約的數據量_In_ MPI_Datatype datatype, // 數據類型_In_ MPI_Op op, // 規約操作類型_In_ MPI_Comm comm // 通信子 );

● 范例代碼

int main(int argc, char **argv) {const int nProcess = 8, localSize = 8, globalSize = localSize * nProcess;int globalData[globalSize], localData[localSize], sumData[localSize];int comRank, comSize, i;MPI_Init(&argc, &argv);MPI_Comm_rank(MPI_COMM_WORLD, &comRank);MPI_Comm_size(MPI_COMM_WORLD, &comSize);if (comRank == 0)for (i = 0; i < globalSize; globalData[i] = i, i++);MPI_Scatter(globalData, localSize, MPI_INT, localData, localSize, MPI_INT, 0, MPI_COMM_WORLD);for (i = 0; i < localSize; i++)printf("%2d, ", localData[i]);MPI_Scan(localData, sumData, localSize, MPI_INT, MPI_SUM, MPI_COMM_WORLD);for (i = 0; i < localSize; i++)printf("%2d, ", sumData[i]);MPI_Finalize();return 0; }

● 輸出結果，分別展示了 localSize 取 1 和 8 的結果，每個進程的輸出中，前一半（分別為 1 個和 8 個元素）為個進程的原始數據，后一半為進行完前綴求和后的結果。注意到 localSize 取 8 時，程序將各進程保存向量的每一個元素分別進行前綴和，但同一進程中各元素之間不相互影響。

D:\Code\MPI\MPIProjectTemp\x64\Debug>mpiexec -n 8 -l MPIProjectTemp.exe [1] 1, 1, [6] 6, 21, [5] 5, 15, [4] 4, 10, [3] 3, 6, [0] 0, 0, [2] 2, 3, [7] 7, 28, D:\Code\MPI\MPIProjectTemp\x64\Debug>mpiexec -n 8 -l MPIProjectTemp.exe [4] 32, 33, 34, 35, 36, 37, 38, 39, 80, 85, 90, 95, 100, 105, 110, 115, [1] 8, 9, 10, 11, 12, 13, 14, 15, 8, 10, 12, 14, 16, 18, 20, 22, [0] 0, 1, 2, 3, 4, 5, 6, 7, 0, 1, 2, 3, 4, 5, 6, 7, [7] 56, 57, 58, 59, 60, 61, 62, 63, 224, 232, 240, 248, 256, 264, 272, 280, [3] 24, 25, 26, 27, 28, 29, 30, 31, 48, 52, 56, 60, 64, 68, 72, 76, [6] 48, 49, 50, 51, 52, 53, 54, 55, 168, 175, 182, 189, 196, 203, 210, 217, [5] 40, 41, 42, 43, 44, 45, 46, 47, 120, 126, 132, 138, 144, 150, 156, 162, [2] 16, 17, 18, 19, 20, 21, 22, 23, 24, 27, 30, 33, 36, 39, 42, 45,

? 規約分發函數 MPI_Reduce_Scatter()，將數據進行規約計算，結果分段分發到各進程中

● 函數原型

_Pre_satisfies_(recvbuf != MPI_IN_PLACE) MPI_METHOD MPI_Reduce_scatter(_In_opt_ _In_range_(!= , recvbuf) const void* sendbuf, // 指向輸入數據的指針_Out_opt_ void* recvbuf, // 指向接收數據的指針_In_ const int recvcounts[], // 各進程接收規約結果的元素個數_In_ MPI_Datatype datatype, // 數據類型_In_ MPI_Op op, // 規約操作類型_In_ MPI_Comm comm // 通信子 );

● 使用范例

{const int nProcess = 8, localSize = 8, globalSize = nProcess * localSize, countValue = 1;int globalData[globalSize], localData[localSize], count[localSize], localSum[countValue], i, comSize, comRank;MPI_Init(&argc, &argv);MPI_Comm_size(MPI_COMM_WORLD, &comSize);MPI_Comm_rank(MPI_COMM_WORLD, &comRank);if (comRank == 0)for (i = 0; i < globalSize; globalData[i] = i, i++);MPI_Scatter(globalData, localSize, MPI_INT, localData, localSize, MPI_INT, 0, MPI_COMM_WORLD);for (i = 0; i < localSize; count[i++] = 1);for (i = 0; i < localSize; i++)printf("%3d, ", localData[i]);MPI_Reduce_scatter(localData, localSum, count, MPI_INT, MPI_SUM, MPI_COMM_WORLD); for (i = 0; i < countValue; i++)printf("%3d, ", localSum[i]);MPI_Finalize();return 0; }

● 輸出結果，這里取定 localSize 為 8，輸出結果的前 8 個元素為分發到各進程中參與規約計算的原始數據，后面元素為規約計算結果。程序將各進程保存向量的每一個元素分別進行前綴和，但同一進程中各元素之間不相互影響，通過修改 countValue（即參數 count 各元素的值），可以將規約計算的結果分發到各進程中

■ countValue == 1（count == { 1, 1, 1, 1, 1, 1, 1, 1 }）情況，每個進程分得一個結果（注意與上面的函數 MPI_Scan() 作對比）

■ countValue == 2（count == { 2, 2, 2, 2, 2, 2, 2, 2 }）情況，前 4 個進程每個進程分得 2 個結果，后 4 的進程訪問越界，得到無意義的值

■ count == { 2, 0, 2, 0, 2, 0, 2, 0 } 情況，偶數號進程每個進程分得 2 個結果，奇數號進程分得 0 個結果，表現為無意義的值

■ 思考，這列每個 localData 長度為 8，所以規約計算的結果為一個長度為 8 的向量，可以在不同進程中進行分發（注意數據尺寸大小 localSize 與運行程序的進程數 nProcess 沒有任何關系，只是在范例中恰好相等），而函數 MPI_Scan() 則相當于在此基礎上保留了所有中間結果（部分前綴結果），所以其輸出為一個長為 localSize，寬度為 nProcess 的矩陣，并且自動按照進程號均分。

D:\Code\MPI\MPIProjectTemp\x64\Debug>mpiexec -n 8 -l MPIProjectTemp.exe // countValue = 1 [6] 48, 49, 50, 51, 52, 53, 54, 55, 272, [0] 0, 1, 2, 3, 4, 5, 6, 7, 224, [2] 16, 17, 18, 19, 20, 21, 22, 23, 240, [4] 32, 33, 34, 35, 36, 37, 38, 39, 256, [3] 24, 25, 26, 27, 28, 29, 30, 31, 248, [1] 8, 9, 10, 11, 12, 13, 14, 15, 232, [5] 40, 41, 42, 43, 44, 45, 46, 47, 264, [7] 56, 57, 58, 59, 60, 61, 62, 63, 280,D:\Code\MPI\MPIProjectTemp\x64\Debug>mpiexec -n 8 -l MPIProjectTemp.exe // countValue = 2 [0] 0, 1, 2, 3, 4, 5, 6, 7, 224, 232, [6] 48, 49, 50, 51, 52, 53, 54, 55, 1717986912, 1717986912, [1] 8, 9, 10, 11, 12, 13, 14, 15, 240, 248, [3] 24, 25, 26, 27, 28, 29, 30, 31, 272, 280, [4] 32, 33, 34, 35, 36, 37, 38, 39, 1717986912, 1717986912, [5] 40, 41, 42, 43, 44, 45, 46, 47, 1717986912, 1717986912, [2] 16, 17, 18, 19, 20, 21, 22, 23, 256, 264, [7] 56, 57, 58, 59, 60, 61, 62, 63, 1717986912, 1717986912,D:\Code\MPI\MPIProjectTemp\x64\Debug>mpiexec -n 8 -l MPIProjectTemp.exe // countValue = 2，count[i] = (i + 1) % 2 * 2 [4] 32, 33, 34, 35, 36, 37, 38, 39, 256, 264, [2] 16, 17, 18, 19, 20, 21, 22, 23, 240, 248, [3] 24, 25, 26, 27, 28, 29, 30, 31, -858993460, -858993460, [1] 8, 9, 10, 11, 12, 13, 14, 15, -858993460, -858993460, [7] 56, 57, 58, 59, 60, 61, 62, 63, -858993460, -858993460, [0] 0, 1, 2, 3, 4, 5, 6, 7, 224, 232, [5] 40, 41, 42, 43, 44, 45, 46, 47, -858993460, -858993460, [6] 48, 49, 50, 51, 52, 53, 54, 55, 272, 280,

?

總結

以上是生活随笔為你收集整理的MPI 集合通信函数 MPI_Reduce()，MPI_Allreduce()，MPI_Bcast()，MPI_Scatter()，MPI_Gather()，MPI_Allgather()，MPI_S的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。