简单的MPI发送/接收程序中的行为不清楚

Question

一段时间以来,我的代码中一直存在错误,无法弄清楚如何解决它.

我想要实现的是很容易的:每个工作节点(即具有等级!= 0的节点)在涉及一些计算的方形结构中获得一行(由1维arry表示).计算完成后,该行将被发送回主服务器.

出于测试目的,不涉及计算.所有发生的事情是:

• master向worker发送行号,worker使用行号来计算相应的值
• worker返回带有结果值的数组

现在,我的问题是:

• 对于一行中的元素数量(大小= 1006)和工人数量> 1,所有工作都按预期工作到一定大小
• 如果一行中的元素超过1006,则工作程序无法关闭,程序也不会终止
• 这只有在我尝试将数组发送回主服务器时才会发生.如果我只是发回一个INT,那么一切都OK(参见doMasterTasks()和doWorkerTasks()中注释掉的行)

基于最后一个要点,我假设必须存在一些竞争条件,这些竞争条件仅在要发送回主设备的阵列达到一定大小时才会出现.

你知道问题是什么吗？

使用以下代码编译以下代码:mpicc -O2 -std = c99 -o simple

像这样运行可执行文件:mpirun -np 3 simple <size>(例如1006或1007)

这是代码:

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

#define MASTER_RANK 0
#define TAG_RESULT 1
#define TAG_ROW 2
#define TAG_FINISHOFF 3

int mpi_call_result, my_rank, dimension, np;

// forward declarations
void doInitWork(int argc, char **argv);
void doMasterTasks(int argc, char **argv);
void doWorkerTasks(void);
void finalize();
void quit(const char *msg, int mpi_call_result);

void shutdownWorkers() {
    printf("All work has been done, shutting down clients now.\n");
    for (int i = 0; i < np; i++) {
        MPI_Send(0, 0, MPI_INT, i, TAG_FINISHOFF, MPI_COMM_WORLD);
    }
}

void doMasterTasks(int argc, char **argv) {
    printf("Starting to distribute work...\n");
    int size = dimension;
    int * dataBuffer = (int *) malloc(sizeof(int) * size);

    int currentRow = 0;
    int receivedRow = -1;
    int rowsLeft = dimension;
    MPI_Status status;

    for (int i = 1; i < np; i++) {
        MPI_Send(&currentRow, 1, MPI_INT, i, TAG_ROW, MPI_COMM_WORLD);
        rowsLeft--;
        currentRow++;

    }

    for (;;) {
//        MPI_Recv(dataBuffer, size, MPI_INT, MPI_ANY_SOURCE, TAG_RESULT, MPI_COMM_WORLD, &status);
        MPI_Recv(&receivedRow, 1, MPI_INT, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);

        if (rowsLeft == 0)
            break;

        if (currentRow > 1004)
            printf("Sending row %d to worker %d\n", currentRow, status.MPI_SOURCE);
        MPI_Send(&currentRow, 1, MPI_INT, status.MPI_SOURCE, TAG_ROW, MPI_COMM_WORLD);
        rowsLeft--;
        currentRow++;
    }
    shutdownWorkers();
    free(dataBuffer);
}

void doWorkerTasks() {
    printf("Worker %d started\n", my_rank);

    // send the processed row back as the first element in the colours array.
    int size = dimension;
    int * data = (int *) malloc(sizeof(int) * size);
    memset(data, 0, sizeof(size));

    int processingRow = -1;
    MPI_Status status;

    for (;;) {

        MPI_Recv(&processingRow, 1, MPI_INT, 0, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
        if (status.MPI_TAG == TAG_FINISHOFF) {
            printf("Finish-OFF tag received!\n");
            break;
        } else {
//            MPI_Send(data, size, MPI_INT, 0, TAG_RESULT, MPI_COMM_WORLD);
            MPI_Send(&processingRow, 1, MPI_INT, 0, TAG_RESULT, MPI_COMM_WORLD);
        }
    }

    printf("Slave %d finished work\n", my_rank);
    free(data);
}

int main(int argc, char **argv) {


    if (argc == 2) {
        sscanf(argv[1], "%d", &dimension);
    } else {
        dimension = 1000;
    }

    doInitWork(argc, argv);

    if (my_rank == MASTER_RANK) {
        doMasterTasks(argc, argv);
    } else {
        doWorkerTasks();
    }
    finalize();
}

void quit(const char *msg, int mpi_call_result) {
    printf("\n%s\n", msg);
    MPI_Abort(MPI_COMM_WORLD, mpi_call_result);
    exit(mpi_call_result);
}

void finalize() {
    mpi_call_result = MPI_Finalize();
    if (mpi_call_result != 0) {
        quit("Finalizing the MPI system failed, aborting now...", mpi_call_result);
    }
}

void doInitWork(int argc, char **argv) {
    mpi_call_result = MPI_Init(&argc, &argv);
    if (mpi_call_result != 0) {
        quit("Error while initializing the system. Aborting now...\n", mpi_call_result);
    }
    MPI_Comm_size(MPI_COMM_WORLD, &np);
    MPI_Comm_rank(MPI_COMM_WORLD, &my_rank);
}

任何帮助是极大的赞赏!

最好,克里斯

Answer 1

如果您查看doWorkerTasks,您会看到它们发送的数据消息与收到的数据完全相同; (并且他们又收到一个关闭它们).

但是你的主代码:

for (int i = 1; i < np; i++) {
    MPI_Send(&currentRow, 1, MPI_INT, i, TAG_ROW, MPI_COMM_WORLD);
    rowsLeft--;
    currentRow++;

}

for (;;) {
    MPI_Recv(dataBuffer, size, MPI_INT, MPI_ANY_SOURCE, TAG_RESULT, MPI_COMM_WORLD, &status);

    if (rowsLeft == 0)
        break;

    MPI_Send(&currentRow, 1, MPI_INT, status.MPI_SOURCE, TAG_ROW, MPI_COMM_WORLD);
    rowsLeft--;
    currentRow++;
}

发送比其收到的数据消息多np-2.特别是,它只保持接收数据,直到它不再发送,即使应该有np-2个未完成的数据消息.将代码更改为以下内容:

int rowsLeftToSend= dimension;
int rowsLeftToReceive = dimension;

for (int i = 1; i < np; i++) {
    MPI_Send(&currentRow, 1, MPI_INT, i, TAG_ROW, MPI_COMM_WORLD);
    rowsLeftToSend--;
    currentRow++;

}

while (rowsLeftToReceive > 0) {
    MPI_Recv(dataBuffer, size, MPI_INT, MPI_ANY_SOURCE, TAG_RESULT, MPI_COMM_WORLD, &status);
    rowsLeftToReceive--;

    if (rowsLeftToSend> 0) {
        if (currentRow > 1004)
            printf("Sending row %d to worker %d\n", currentRow, status.MPI_SOURCE);
        MPI_Send(&currentRow, 1, MPI_INT, status.MPI_SOURCE, TAG_ROW, MPI_COMM_WORLD);
        rowsLeftToSend--;
        currentRow++;
    }
}

现在有效.

为什么代码没有死锁(注意这是死锁,而不是竞争条件;这是分布式计算中更常见的并行错误),对于较小的消息大小,这是大多数MPI实现如何工作的细微细节.通常,MPI实现只是将小消息"推"到管道中,无论接收器是否为它们做好准备,但是更大的消息(因为它们在接收端占用更多的存储资源)需要在发送器和接收器之间进行一些握手.(如果您想了解更多信息,请搜索eager vs rendezvous协议).

因此,对于小消息情况(在这种情况下少于1006个int,并且1个int肯定也可以工作),无论主服务器是否正在接收它们,工作节点都会发送它们.如果主服务器已调用MPI_Recv(),则消息已存在,并且它将立即返回.但它没有,所以主人方面有未决的消息; 但没关系.主人发出了杀人信息,每个人都退出了.

但是对于较大的消息,剩余的send()必须让接收器参与清除,并且由于接收器从未这样做,所以其余的工作人员都会挂起.

请注意,即使对于没有死锁的小消息情况,代码也无法正常工作 - 缺少计算数据.

更新:你的问题有类似的问题shutdownWorkers:

void shutdownWorkers() {
    printf("All work has been done, shutting down clients now.\n");
    for (int i = 0; i < np; i++) {
        MPI_Send(0, 0, MPI_INT, i, TAG_FINISHOFF, MPI_COMM_WORLD);
    }
}

在这里,您将发送到所有进程,包括排名0,即发送的进程.原则上,MPI_Send应该是死锁,因为它是一个阻塞发送,并且没有匹配的接收已经发布.您可以在之前发布非阻塞接收以避免这种情况,但这是不必要的 - 排名0不需要让自己知道结束.所以只需将循环更改为

    for (int i = 1; i < np; i++)

tl; dr - 你的代码陷入僵局,因为主人没有从工人那里得到足够的信息; 它恰好适用于小型消息大小,因为大多数MPI库都有一个共同的实现细节.