简单 MPI 发送/接收程序中的行为不明确

Question

我的代码中出现错误已经有一段时间了，但还不知道如何解决它。

我想要实现的目标很简单：每个工作节点（即等级！=0的节点）在涉及一些计算的方形结构中获取一行（由一维数组表示）。计算完成后，该行将被发送回主服务器。

出于测试目的，不涉及任何计算。所发生的事情是：

• master 将行号发送给worker，worker 使用行号来计算相应的值
• worker 将带有结果值的数组发送回来

现在，我的问题是：

• 所有工作都按预期进行，直到数字达到一定大小一行中的元素数（大小 = 1006）和工作人员数量 > 1
• 如果一行中的元素超过 1006，工作线程将无法关闭，并且程序不会终止，
• 只有当我尝试将数组发送回主节点时才会发生这种情况。如果我只是发回一个 INT，那么一切都可以（请参阅 doMasterTasks() 和 doWorkerTasks() 中的注释行）。

根据最后一个要点，我假设必须存在一些竞争条件，该条件仅在数组为达到一定大小后发送回master。

您知道问题出在哪里吗？

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

运行可执行文件，如下所示： mpirun -np 3 simple; （例如 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(¤tRow, 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(¤tRow, 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(¤tRow, 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(¤tRow, 1, MPI_INT, status.MPI_SOURCE, TAG_ROW, MPI_COMM_WORLD);
        rowsLeftToSend--;
        currentRow++;
    }
}

现在可以了。

为什么对于较小的消息大小，代码不会死锁（请注意，这是死锁，而不是竞争条件；这是分布式计算中更常见的并行错误），这是大多数 MPI 实现如何工作的微妙细节。一般来说，MPI 实现只是将小消息“推”到管道中，无论接收者是否准备好接收它们，但较大的消息（因为它们在接收端占用更多存储资源）需要发送者和接收者之间进行一些握手。 （如果您想了解更多信息，请搜索 eager 与 rendezvous 协议）。

因此，对于小消息情况（在这种情况下少于 1006 个整数，并且 1 个整数也肯定有效），无论主节点是否正在接收它们，工作节点都会发送它们。如果主设备调用了 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 库共有的实现细节，它恰好适用于小消息大小。