MPI_Gather使用异常：非根进程数据无法写入根进程latter_clusters

核心问题与修改点 #

• 所有进程必须调用MPI_Gather：原代码中根进程（rank=0）困在自身do-while循环里，从未执行MPI_Gather，导致数据无法被接收。必须让根进程也参与集体通信调用。
• job_size未正确初始化与广播：初始job_size为0，且根进程未根据进程数计算每个进程的任务量并广播，导致Scatter/Gather无数据传输。
• do-while循环逻辑错误：原循环条件while(job_done)（job_done初始为0）会直接跳过循环，应改为while(!job_done)，同时需添加循环终止条件（比如迭代次数或收敛判断）。
• points变量未初始化：根进程需确保points正确分配内存并读取数据，非根进程无需保留points变量。
• 内存分配时机错误：buffreceive应在job_size确定后再分配内存，避免用0大小分配无效内存。

修正后的完整代码 #

#include <stdio.h>
#include <stdlib.h>
#include <math.h>
#include "mpi.h"

typedef struct {
    double x;
    double y;
} Point;

void initialize(Point* centroids, int num_clusters) {
    // 示例初始化，可根据实际需求修改
    centroids[0].x = 0.0; centroids[0].y = 0.0;
    centroids[1].x = 1.0; centroids[1].y = 1.0;
}

int main(int argc, char *argv[]) {
    MPI_Init(&argc, &argv);
    int rank, size;

    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Comm_size(MPI_COMM_WORLD, &size);
    MPI_Datatype POINT;
    MPI_Type_contiguous(2, MPI_DOUBLE, &POINT);
    MPI_Type_commit(&POINT);

    int num_clusters = 2;
    int num_points = 9;
    int job_size = 0;
    int* latter_clusters = NULL;
    int* worker_clusters = NULL;
    int job_done = 0;
    int iter_count = 0;
    const int max_iter = 10; // 设置最大迭代次数作为终止条件

    Point* centroids = calloc(sizeof(Point), num_clusters);
    Point* points = NULL;
    Point* buffreceive = NULL;

    if (rank == 0) {
        // 模拟读取文件初始化points，实际代码替换为真实文件读取逻辑
        points = malloc(sizeof(Point) * num_points);
        for (int i = 0; i < num_points; i++) {
            points[i].x = (double)i / 3.0;
            points[i].y = (double)i / 3.0;
        }

        latter_clusters = malloc(sizeof(int) * num_points);
        // 去掉重复的centroids分配，避免覆盖之前的calloc
        initialize(centroids, num_clusters);

        // 计算每个进程的任务量，处理num_points不能被size整除的情况
        job_size = num_points / size;
        int remainder = num_points % size;
        if (remainder != 0) {
            job_size++;
        }
    }

    // 广播job_size给所有进程
    MPI_Bcast(&job_size, 1, MPI_INT, 0, MPI_COMM_WORLD);
    MPI_Bcast(&num_clusters, 1, MPI_INT, 0, MPI_COMM_WORLD);
    MPI_Bcast(centroids, num_clusters, POINT, 0, MPI_COMM_WORLD);

    // 在job_size确定后分配buffreceive内存
    buffreceive = malloc(sizeof(Point) * job_size);
    // 使用MPI_Scatterv处理任务量不均的情况（如果num_points不能被size整除）
    int* sendcounts = NULL;
    int* displs = NULL;
    if (rank == 0) {
        sendcounts = malloc(sizeof(int) * size);
        displs = malloc(sizeof(int) * size);
        int offset = 0;
        for (int i = 0; i < size; i++) {
            sendcounts[i] = (i < remainder) ? job_size : job_size - 1;
            displs[i] = offset;
            offset += sendcounts[i];
        }
    }
    MPI_Scatterv(points, sendcounts, displs, POINT, buffreceive, job_size, POINT, 0, MPI_COMM_WORLD);

    // 所有进程进入迭代循环，包括根进程
    do {
        if (rank != 0) {
            // 非根进程计算所属点的聚类
            worker_clusters = malloc(sizeof(int) * job_size);
            printf("Calculation of new clusters [%d]\n", rank);
            for (int i = 0; i < job_size; i++) {
                int cluster = 0;
                double min_distance = pow((buffreceive[i].x - centroids[0].x) * 100, 2)
                                    + pow((buffreceive[i].y - centroids[0].y) * 100, 2);

                for (int j = 1; j < num_clusters; j++) {
                    double distance = pow((buffreceive[i].x - centroids[j].x) * 100, 2)
                                    + pow((buffreceive[i].y - centroids[j].y) * 100, 2);
                    if (distance <= min_distance) {
                        cluster = j;
                        min_distance = distance;
                    }
                }
                worker_clusters[i] = cluster;
            }
        }

        // 所有进程调用MPI_Gatherv处理任务量不均的情况
        MPI_Gatherv(worker_clusters, job_size, MPI_INT, latter_clusters, sendcounts, displs, MPI_INT, 0, MPI_COMM_WORLD);

        // 根进程处理收集到的数据，比如更新质心，判断是否收敛
        if (rank == 0) {
            printf("Iteration %d: gathered cluster assignments\n", iter_count);
            for (int i = 0; i < num_points; i++) {
                printf("slave_clusters[%d] = %d\n", i, latter_clusters[i]);
            }

            // 这里添加质心更新和收敛判断逻辑，示例中用迭代次数终止
            iter_count++;
            if (iter_count >= max_iter) {
                job_done = 1;
            }
        }

        // 广播终止信号给所有进程
        MPI_Bcast(&job_done, 1, MPI_INT, 0, MPI_COMM_WORLD);
        // 如果需要更新质心，这里还要广播新的centroids

        // 释放非根进程的worker_clusters内存，避免内存泄漏
        if (rank != 0) {
            free(worker_clusters);
            worker_clusters = NULL;
        }

    } while (!job_done);

    // 释放所有内存
    if (rank == 0) {
        free(points);
        free(latter_clusters);
        free(sendcounts);
        free(displs);
    }
    free(centroids);
    free(buffreceive);

    MPI_Type_free(&POINT);
    MPI_Finalize();
    return 0;
}

额外说明 #

• 改用MPI_Scatterv和MPI_Gatherv是为了处理num_points不能被进程数整除的情况，确保每个进程分配合理的任务量。
• 添加了最大迭代次数作为循环终止条件，实际应用中可以替换为聚类收敛的判断逻辑（比如质心变化小于阈值）。
• 修复了内存泄漏问题，确保所有分配的内存都被释放。

内容的提问来源于stack exchange，提问作者Roberto