如何修复MPI程序中重复使用MPI_Request及未等待通信完成的警告

我用MPI实现了基于环形算法的协调器选举程序，在4*4共16个进程的晶片机矩阵中运行，所有进程间交互采用点对点MPI传输。程序能正常运行，但存在两个需要修复的警告：

• 未等待通信完成就为不同通信操作重复使用同一个MPI_Request；
• 未等待通信完成就调用MPI_Finalize退出。

我推测问题2可通过在MPI_Finalize前使用MPI_Waitall或MPI_Barrier替代MPI_Wait解决，但需要彻底修复这两个问题。

问题1修复：避免重复使用MPI_Request #

MPI_Request对象不能在未完成前被重复赋值，每次异步通信（MPI_Isend/MPI_Irecv）都需要独立的Request变量。

关键修改点： #

• 在send_to_next函数中，为MPI_Isend和MPI_Irecv分别创建独立的MPI_Request，用MPI_Testany检查任意请求的完成状态；
• 主函数中所有异步通信操作，每个操作分配单独的MPI_Request，避免覆盖未完成的请求。

问题2修复：确保所有通信完成后再退出 #

不能直接调用MPI_Finalize，必须等待所有异步通信完成。使用MPI_Waitall统一等待所有未完成的请求，相比MPI_Barrier更可靠（MPI_Barrier仅保证进程同步，不保证通信操作完成）。

修改后的完整代码 #

#include <mpi.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <stdbool.h>
#include <string.h>

const int CIRCLE_START_RANK = 5;

const int ELECTION_TAG = 1;
const int I_AM_HERE_TAG = 2;
const int COORDINATOR_TAG = 3;

void roll_call(int rank) {
    printf("%d ", rank);
    fflush(stdout);
}

int send_to_next(int rank, int size, int *array, int tag) {
    // 尝试发送消息并返回下一个活跃进程的rank
    MPI_Request send_req, recv_req;
    MPI_Status statuses[2];
    int answer;
    int successfully_sent = 0;
    int next = 1;
    while (!successfully_sent) {
        // 为发送和接收分别创建独立的Request
        MPI_Isend(array, size, MPI_INT, (rank + next) % size, tag, MPI_COMM_WORLD, &send_req);
        MPI_Irecv(&answer, 1, MPI_INT, (rank + next) % size, I_AM_HERE_TAG, MPI_COMM_WORLD, &recv_req);

        MPI_Request reqs[2] = {send_req, recv_req};
        int completed_idx;
        double start = MPI_Wtime();
        while (!successfully_sent) {
            // 检查任意一个请求是否完成
            int flag;
            MPI_Testany(2, reqs, &completed_idx, &flag, &statuses[completed_idx]);
            if (flag) {
                // 如果是接收完成，说明对方活跃
                if (completed_idx == 1) {
                    successfully_sent = 1;
                    // 等待发送请求完成
                    MPI_Wait(&send_req, MPI_STATUS_IGNORE);
                }
            }
            // 超时时间为1秒
            if (MPI_Wtime() - start >= 1) {
                printf("%2d: Did not receive confirmation from %2d.\n", rank, (rank + next) % size);
                fflush(stdout);
                // 取消未完成的请求并等待完成
                MPI_Cancel(&send_req);
                MPI_Cancel(&recv_req);
                MPI_Wait(&send_req, MPI_STATUS_IGNORE);
                MPI_Wait(&recv_req, MPI_STATUS_IGNORE);
                break;
            }
        }
        next++;
    }
    return (rank + next - 1) % size;
}

int calculate_coordinator(int *array, int size) {
    // 选取rank最高的进程作为协调器
    int i;
    for (i = size - 1; i >= 0; i--) {
        if (array[i]){
            return i;
        }
    }
    return 0;
}

int main(int argc, char* argv[]) {
    setvbuf(stdout, NULL, _IOLBF, BUFSIZ);
    setvbuf(stderr, NULL, _IOLBF, BUFSIZ);
    int size, rank;
    MPI_Init(&argc, &argv);
    MPI_Comm_size(MPI_COMM_WORLD, &size);
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    srand(time(NULL) * rank);

    bool am_i_alive = true;

    if (rank == CIRCLE_START_RANK) {
        printf("\n\nLive processes:\n");
        fflush(stdout);
    } else {
        am_i_alive = rand() > 0.42 * RAND_MAX;
    }

    MPI_Barrier(MPI_COMM_WORLD);

    if (am_i_alive) {
        roll_call(rank);
    }
    MPI_Barrier(MPI_COMM_WORLD);

    if (!am_i_alive) {
        MPI_Finalize();
        return 0;
    }

    int *array = (int *) calloc(size, sizeof(int));

    MPI_Request reqs[5]; // 存储所有异步请求
    int req_count = 0;
    MPI_Status status;
    int next;

    if (rank == CIRCLE_START_RANK) {
        printf("\nAlgorithm:\n");
        fflush(stdout);
        array[rank] = 1;
        next = send_to_next(rank, size, array, ELECTION_TAG);
    }

    // 选举流程:
    int answer = 1;
    int new_coordinator;

    MPI_Irecv(array, size, MPI_INT, MPI_ANY_SOURCE, ELECTION_TAG, MPI_COMM_WORLD, &reqs[req_count++]);
    MPI_Wait(&reqs[req_count-1], &status);
    printf("%2d: Got an array from %2d\n", rank, status.MPI_SOURCE);
    fflush(stdout);

    if (array[rank]) {
        // 值为1表示已参与过选举，需统计结果
        new_coordinator = calculate_coordinator(array, size);
        printf("%2d: New coordinator: %2d\n", rank, new_coordinator);
        fflush(stdout);
        // 发送I_AM_HERE
        MPI_Isend(&answer, 1, MPI_INT, status.MPI_SOURCE, I_AM_HERE_TAG, MPI_COMM_WORLD, &reqs[req_count++]);
        // 发送COORDINATOR
        MPI_Isend(&new_coordinator, 1, MPI_INT, next, COORDINATOR_TAG, MPI_COMM_WORLD, &reqs[req_count++]);
    } else {
        // 更新数组并发送给下一个进程
        array[rank] = 1;
        // 发送I_AM_HERE
        MPI_Isend(&answer, 1, MPI_INT, status.MPI_SOURCE, I_AM_HERE_TAG, MPI_COMM_WORLD, &reqs[req_count++]);
        next = send_to_next(rank, size, array, ELECTION_TAG);
    }

    // 第二轮：传递新协调器信息
    MPI_Irecv(&new_coordinator, 1, MPI_INT, MPI_ANY_SOURCE, COORDINATOR_TAG, MPI_COMM_WORLD, &reqs[req_count++]);
    MPI_Wait(&reqs[req_count-1], &status);
    if (rank != CIRCLE_START_RANK) {
        printf("%2d: New coordinator: %2d\n", rank, new_coordinator);
        fflush(stdout);
        // 转发协调器信息
        MPI_Isend(&new_coordinator, 1, MPI_INT, next, COORDINATOR_TAG, MPI_COMM_WORLD, &reqs[req_count++]);
    }
    
    // 等待所有未完成的异步请求完成
    if (req_count > 0) {
        MPI_Waitall(req_count, reqs, MPI_STATUSES_IGNORE);
    }

    free(array); // 释放内存
    MPI_Finalize();
    return 0;
}

编译运行命令 #

mpicc dz_1.c -o main
mpirun --oversubscribe -n 16 main

内容的提问来源于stack exchange，提问作者Маша