MPI_Gather实现二维热板4 point stencil计算结果聚合问题咨询

问题描述 #

我正在学习用MPI计算二维热板的4-point stencil，现在遇到的问题是：计算结果写入输出文件时，文件里只有某一个任务（推测是最先完成的任务）的结果。我希望worker任务完成所有迭代后，把数据聚合回master节点，计划用MPI_Gather，但看教程还是没掌握正确用法。另外，我在worker进程的数据返回方式上有没有错误？相关代码如下：

#define BEGIN       1                  /* message tag */
#define LTAG        2                  /* message tag */
#define RTAG        3                  /* message tag */
#define NONE        0                  /* indicates no neighbor */
#define DONE        4                  /* message tag */
#define MASTER      0                  /* taskid of first process */
#include <iostream>
#include <math.h>
#include <fstream>
#include <string>
#include <cstring>
#include <iomanip>
#include <chrono>
#include <mpi.h>
#include <chrono>
using namespace std::chrono;
using namespace std;



void update(int start, int end, int ny, double* u1, double* u2)
{
    int ix, iy;
    for (ix = start; ix <= end; ix++)
    {
        for (iy = 1; iy <= ny - 2; iy++)
        {
            //calculate stencil
            *(u2 + ix * ny + iy) = .25 * (
                *(u1 + (ix + 1) * ny + iy) +
                *(u1 + (ix - 1) * ny + iy) +
                *(u1 + ix * ny + (iy + 1)) +
                *(u1 + ix * ny + (iy - 1)));
        }


    }

    //pass into buffer
    for (ix = start; ix <= end; ix++)
        for (iy = 1; iy <= ny - 2; iy++)
            *(u1 + ix * ny + iy) = *(u2 + ix * ny + iy);

}



bool isNumber(char number[])
{
    int i = 0;

    //checking for negative numbers
    if (number[0] == '-')
        return false;
    for (; number[i] != 0; i++)
    {
        //if (number[i] > '9' || number[i] < '0')
        if (!isdigit(number[i]))
            return false;
    }
    return true;
}


int main(int argc, char* argv[])
{
    //variable init
    int N = 0;
    int I = 0;
    int taskid,                     /* this task's unique id */
        numworkers,                 /* number of worker processes */
        numtasks,                   /* number of tasks */
        averow, rows, offset, extra,   /* for sending rows of data */
        dest, source,               /* to - from for message send-receive */
        left, right,        /* neighbor tasks */
        msgtype,                    /* for message types */
        rc, start, end,               /* misc */
        i, ix, iy, iz, it;              /* loop variables */
    MPI_Status status;

    //There should be five arguments, no more no less
    if (argc != 5)
    {
        std::cout << "Invalid parameters, please check your values." << std::endl;
        exit(1);
    }
    if (!isNumber(argv[2]) || !isNumber(argv[4]))
    {
        std::cout << "Invalid parameters, please check your values." << std::endl;
        exit(1);
    }

    //check the n flag in either valid position. 
    if (!strcmp(argv[1], "-n"))
    {
        N = atoi(argv[2]);
    }
    else if (!strcmp(argv[3], "-n"))
    {
        N = atoi(argv[4]);
    }
    else
    {
        std::cout << "Invalid flags, please check your values." << std::endl;
        exit(1);
    }

    //check the I flag in either valid position
    if (!strcmp(argv[1], "-I"))
    {
        I = atoi(argv[2]);
    }
    else if (!strcmp(argv[3], "-I"))
    {
        I = atoi(argv[4]);
    }
    else
    {
        std::cout << "Invalid flags, please check your values." << std::endl;
        exit(1);
    }

    if (N == 0 || I == 0)
    {
        std::cout << "Invalid parameters, please check your values." << std::endl;
        exit(1);
    }
    int n_x = N + 2; //Based on the piazza post, increase x and y dimension by two for the edges
    int n_y = N + 2;
    double** phi[2];
    double** resultArray;
    phi[0] = new double* [n_x];
    phi[1] = new double* [n_x];
    resultArray = new double* [n_x];
    for (int i = 0; i < n_x; i++)
    {
        phi[0][i] = new double[n_y];
        phi[1][i] = new double[n_y];
        resultArray[i] = new double[n_y];
    }
    int iterationMax = I;
    //Initialize the plate temperatures
    for (int j = 0; j < n_x; j++)
    {
        for (int i = 0; i < n_x; i++)
        {
            if ((i == 0) || (i == n_x - 1) || (j == 0) || (j == n_y - 1))
            {
                phi[0][i][j] = 20.0;
                phi[1][i][j] = 20.0;
            }
            else
            {
                phi[0][i][j] = 0.0;
                phi[1][i][j] = 0.0;
            }


        }
    }

    for (int j = 0; j < n_x; j++)
    {
        if (j > (int)(0.3 * n_x) && j < (int)(0.7 * n_x))
        {

            phi[0][0][j] = 100.0;
            phi[1][0][j] = 100.0;
        }
    }



    /* First, find out my taskid and how many tasks are running */
    MPI_Init(&argc, &argv);
    MPI_Comm_size(MPI_COMM_WORLD, &numtasks);
    MPI_Comm_rank(MPI_COMM_WORLD, &taskid);
    numworkers = numtasks - 1;
    if (taskid == 0) {
        auto start = high_resolution_clock::now();

        /************************* master code *******************************/
        /* Check if numworkers is within range - quit if not */

        //printf("Starting mpi_heat2D with %d worker tasks.
", numworkers);

        /* Initialize grid */
        //printf("Grid size: X= %d  Y= %d  Time steps= %d
", n_x, n_y, I);


        /* Distribute work to workers.  Must first figure out how many rows to */
        /* send and what to do with extra rows.  */
        averow = n_x / numworkers;
        extra = n_x % numworkers;
        offset = 0;
        for (i = 1; i <= numworkers; i++)
        {
            rows = (i <= extra) ? averow + 1 : averow;
            /* Tell each worker who its neighbors are, since they must exchange */
            /* data with each other. */
            if (i == 1)
                left = 0;
            else
                left = i - 1;
            if (i == numworkers)
                right = 0;
            else
                right = i + 1;
            /*  Now send startup information to each worker  */
            dest = i;
            MPI_Send(&offset, 1, MPI_INT, dest, 1, MPI_COMM_WORLD);
            MPI_Send(&rows, 1, MPI_INT, dest, 1, MPI_COMM_WORLD);
            MPI_Send(&left, 1, MPI_INT, dest, 1, MPI_COMM_WORLD);
            MPI_Send(&right, 1, MPI_INT, dest, 1, MPI_COMM_WORLD);
            MPI_Send(&phi[0][offset][0], rows * n_y, MPI_DOUBLE, dest, 1,
                MPI_COMM_WORLD);
            /* printf("Sent to task %d: rows= %d offset= %d ", dest, rows, offset);
             printf("left= %d right= %d
", left, right);*/
            offset = offset + rows;
        }
        /* Now wait for results from all worker tasks */
        for (i = 1; i <= numworkers; i++)
        {
            source = i;
            msgtype = DONE;
            MPI_Recv(&offset, 1, MPI_INT, source, msgtype, MPI_COMM_WORLD,
                &status);
            MPI_Recv(&rows, 1, MPI_INT, source, msgtype, MPI_COMM_WORLD, &status);
            MPI_Recv(&phi[0][offset][0], rows * n_y, MPI_DOUBLE, source,
                msgtype, MPI_COMM_WORLD, &status);
        }


        /* Write final output, call X graph and finalize MPI */

        auto stop = high_resolution_clock::now();
        auto duration = duration_cast<milliseconds>(stop - start);

        std::cout.precision(5);
        // output the time to the console
        std::cout << "Time taken to solve diffusion plate: " << duration.count() << "ms" << std::fixed << std::endl;

    }   /* End of master code */

    /************************* workers code **********************************/
    if (taskid != MASTER)
    {

        /* Receive my offset, rows, neighbors and grid partition from master */
        source = MASTER;
        msgtype = BEGIN;
        MPI_Recv(&offset, 1, MPI_INT, source, msgtype, MPI_COMM_WORLD, &status);
        MPI_Recv(&rows, 1, MPI_INT, source, msgtype, MPI_COMM_WORLD, &status);
        MPI_Recv(&left, 1, MPI_INT, source, msgtype, MPI_COMM_WORLD, &status);
        MPI_Recv(&right, 1, MPI_INT, source, msgtype, MPI_COMM_WORLD, &status);
        MPI_Recv(&phi[0][offset][0], rows * n_y, MPI_DOUBLE, source, msgtype,
            MPI_COMM_WORLD, &status);

        /* Determine border elements.  Need to consider first and last columns. */
        /* Obviously, row 0 can't exchange with row 0-1.  Likewise, the last */
        /* row can't exchange with last+1.  */
        start = offset;
        end = offset + rows - 1;
        if (offset == 0)
            start = 1;
        if ((offset + rows) == n_x)
            end--;
        // printf("task=%d  start=%d  end=%d
", taskid, start, end);

         /* Begin doing STEPS iterations.  Must communicate border rows with */
         /* neighbors.  If I have the first or last grid row, then I only need */
         /*  to  communicate with one neighbor  */
         //printf("Task %d received work. Beginning time steps...
", taskid);
        iz = 0;
        for (it = 1; it <= iterationMax; it++)
        {
            if (left != NONE)
            {
                MPI_Send(&phi[iz][offset][0], n_y, MPI_DOUBLE, left,
                    RTAG, MPI_COMM_WORLD);
                source = left;
                msgtype = LTAG;
                MPI_Recv(&phi[iz][offset - 1][0], n_y, MPI_DOUBLE, source,
                    msgtype, MPI_COMM_WORLD, &status);
            }
            if (right != NONE)
            {
                MPI_Send(&phi[iz][offset + rows - 1][0], n_y, MPI_DOUBLE, right,
                    LTAG, MPI_COMM_WORLD);
                source = right;
                msgtype = RTAG;
                MPI_Recv(&phi[iz][offset + rows][0], n_y, MPI_DOUBLE, source, msgtype,
                    MPI_COMM_WORLD, &status);
            }
            /* Now call update to update the value of grid points */
            update(start, end, n_y, &phi[iz][0][0], &phi[1-iz][0][0]);
            iz = 1 - iz;
        }
        /* Finally, send my portion of final results back to master */
        MPI_Send(&offset, 1, MPI_INT, MASTER, DONE, MPI_COMM_WORLD);
        MPI_Send(&rows, 1, MPI_INT, MASTER, DONE, MPI_COMM_WORLD);
        MPI_Send(&phi[iz][offset][0], rows * n_y, MPI_DOUBLE, MASTER, DONE,
            MPI_COMM_WORLD);
        //printf("Task %d finished work...
", taskid);
        MPI_Finalize();
    }

    //This is probably not correct
    /*double gsize;
    int root, myrank;
    double* mybuf;

    MPI_Gather(&mybuf, 1, MPI_DOUBLE, phi, 1, MPI_DOUBLE, 0, MPI_COMM_WORLD);*/

    std::ofstream outJob;
    outJob.open("finalTemperatures.csv");
    std::cout.precision(15);
    double* printAddr = &phi[0][0][0];
    for (int j = 0; j < n_y; j++) //file IO
    {
        for (int i = 0; i < n_x; i++)
        {
            outJob << phi[0][j][i] << "," << std::fixed;
        }
        outJob << std::endl;
    }
    outJob.close();


    return 0;
}

核心问题分析 #

1. 文件写入的错误根源 #

所有MPI进程都会执行最后的文件写入代码，多个进程同时打开并写入同一个文件finalTemperatures.csv，会导致文件内容被互相覆盖，最终只保留最后完成写入的进程数据。必须仅让master节点执行文件写入操作。

2. 当前worker数据返回方式的问题 #

你用MPI_Send/MPI_Recv手动发送结果的逻辑本身是正确的，但存在两个规范问题：

• worker进程发送完结果后直接调用MPI_Finalize()，但master进程未在接收完所有结果后调用该函数，违反MPI规范（所有进程都需调用MPI_Finalize()，且必须在所有通信完成后执行）。
• 你注释的MPI_Gather代码完全错误：MPI_Gather要求所有进程发送的数据长度一致，而你的worker进程处理的行数因extra分配存在差异，因此当前手动的点对点通信更适配你的场景，无需强行替换为MPI_Gather。

修复步骤 #

步骤1：将文件写入逻辑移至master进程内部 #

把文件写入代码块放到if (taskid == 0)分支中，确保master在接收完所有worker的结果后，再统一写入完整数据。

步骤2：统一MPI_Finalize()调用 #

• 删除worker进程中的MPI_Finalize()，在main函数末尾（所有逻辑执行完毕后）统一调用，确保所有进程都执行这一步。

步骤3：修正代码细节 #

• 迭代完成后发送的phi[iz]是正确的，无需修改；update函数中的数据复制逻辑可优化，但不影响功能。

修复后的关键代码片段 #

master进程补充文件写入逻辑：

if (taskid == 0) {
    // ... 原有接收worker结果的代码 ...

    // 写入文件
    std::ofstream outJob;
    outJob.open("finalTemperatures.csv");
    std::cout.precision(15);
    for (int j = 0; j < n_y; j++)
    {
        for (int i = 0; i < n_x; i++)
        {
            outJob << std::fixed << phi[0][j][i] << ",";
        }
        outJob << std::endl;
    }
    outJob.close();

    auto stop = high_resolution_clock::now();
    auto duration = duration_cast<milliseconds>(stop - start);

    std::cout.precision(5);
    std::cout << "Time taken to solve diffusion plate: " << duration.count() << "ms" << std::fixed << std::endl;
}

移除worker内的MPI_Finalize()，在main末尾统一调用：

// ... 原有所有代码 ...

// 统一执行MPI结束操作
MPI_Finalize();
return 0;
}

#