当一个 MPI 进程执行 MPI_Barrier() 时，其他进程挂起

Question

我有一个 MPI 程序，用于从包含文件名列表的文件中读取多个进程，并根据读取的文件名 - 它读取相应的文件并计算单词的频率。

如果其中一个进程完成此操作并返回 - 以阻止执行 MPI_Barrier()，则其他进程也会挂起。调试时可以看到readFile()函数没有被当前在process_files()中的进程进入，无法弄清楚为什么会发生这种情况。请在下面找到代码：

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

void process_files(char*, int* , int, hashtable_t* );

void initialize_word(char *c,int size)
{
    int i;
    for(i=0;i<size;i++)
        c[i]=0;

    return;
}



char* readFilesList(MPI_File fh, char* file,int rank, int nprocs, char* block, const int overlap, int* length)
{
    char *text;
    int blockstart,blockend;

    MPI_Offset size;
    MPI_Offset blocksize;
    MPI_Offset begin;
    MPI_Offset end;
    MPI_Status status;

    MPI_File_open(MPI_COMM_WORLD,file,MPI_MODE_RDONLY,MPI_INFO_NULL,&fh);
    MPI_File_get_size(fh,&size);

    /*Block size calculation*/
    blocksize = size/nprocs;
    begin = rank*blocksize;
    end = begin+blocksize-1;

    end+=overlap;

    if(rank==nprocs-1)
        end = size;

    blocksize = end-begin+1;

    text = (char*)malloc((blocksize+1)*sizeof(char));
    MPI_File_read_at_all(fh,begin,text,blocksize,MPI_CHAR, &status);
    text[blocksize+1]=0;

    blockstart = 0;
    blockend = blocksize;

    if(rank!=0)
    {
        while(text[blockstart]!='\n' && blockstart!=blockend) blockstart++;
        blockstart++;
    }

    if(rank!=nprocs-1)
    {

        blockend-=overlap;
        while(text[blockend]!='\n'&& blockend!=blocksize) blockend++;
    }



    blocksize = blockend-blockstart;

    block = (char*)malloc((blocksize+1)*sizeof(char));
    block = memcpy(block, text + blockstart, blocksize);
    block[blocksize]=0;
    *length = strlen(block);

    MPI_File_close(&fh);
    return block;
}

void calculate_term_frequencies(char* file, char* text, hashtable_t *hashtable,int rank)
{
    printf("Start File %s, rank %d \n\n ",file,rank);
    fflush(stdout);
    if(strlen(text)!=0||strlen(file)!=0)
    {

        int i,j;
        char w[100];
        i=0,j=0;
        while(text[i]!=0)
        {
            if((text[i]>=65&&text[i]<=90)||(text[i]>=97&&text[i]<=122))
            {
                w[j]=text[i];
                j++; i++;
            }

            else
            {

                w[j] = 0;
                if(j!=0)
                {
                    //ht_set( hashtable, strcat(strcat(w,"#"),file),1);
                }
                j=0;
                i++;
                initialize_word(w,100);
            }

        }
    }
    return;
}

void readFile(char* filename, hashtable_t *hashtable,int rank)
{
    MPI_Status stat;
    MPI_Offset size;
    MPI_File fx;
    char* textFromFile=0;

    printf("Start File %d, rank %d \n\n ",strlen(filename),rank);
    fflush(stdout);

    if(strlen(filename)!=0)
    {
        MPI_File_open(MPI_COMM_WORLD,filename,MPI_MODE_RDONLY,MPI_INFO_NULL,&fx);
        MPI_File_get_size(fx,&size);

        printf("Start File %s, rank %d \n\n ",filename,rank);
        fflush(stdout);

        textFromFile = (char*)malloc((size+1)*sizeof(char));
        MPI_File_read_at_all(fx,0,textFromFile,size,MPI_CHAR, &stat);
        textFromFile[size]=0;
        calculate_term_frequencies(filename, textFromFile, hashtable,rank);

        MPI_File_close(&fx);

    }

    printf("Done File %s, rank %d \n\n ",filename,rank);
    fflush(stdout);
    return;   
}

void process_files(char* block, int* length, int rank,hashtable_t *hashtable)
{

    char s[2];
    s[0] = '\n';
    s[1] = 0;

    char *file;
    if(*length!=0)
    {
        /* get the first file */
        file = strtok(block, s);

        /* walk through other tokens */
        while( file != NULL ) 
        {
            readFile(file,hashtable,rank);
            file = strtok(NULL, s);
        }
    }
    return;
}

void execute_process(MPI_File fh, char* file, int rank, int nprocs, char* block, const int overlap, int * length, hashtable_t *hashtable)
{

    block = readFilesList(fh,file,rank,nprocs,block,overlap,length);
    process_files(block,length,rank,hashtable);
}


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

    /*Initialization*/
    MPI_Init(&argc, &argv);
    MPI_File fh=0;
    int rank,nprocs,namelen;
    char *block=0;
    const int overlap = 70;
    char* file = "filepaths.txt";
    int *length = (int*)malloc(sizeof(int));

    hashtable_t *hashtable = ht_create( 65536 );

    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Comm_size(MPI_COMM_WORLD, &nprocs);

    char processor_name[MPI_MAX_PROCESSOR_NAME];
    MPI_Get_processor_name(processor_name, &namelen);
    printf("Rank %d is on processor %s\n",rank,processor_name);
    fflush(stdout);

    execute_process(fh,file,rank,nprocs,block,overlap,length,hashtable);

    printf("Rank %d returned after processing\n",rank);
    MPI_Barrier(MPI_COMM_WORLD);

    MPI_Finalize();
    return 0;

}

filepaths.txt 是一个包含普通文本文件的绝对文件名的文件：

例如：

/home/mpiuser/mpi/MPI_Codes/code/test1.txt
/home/mpiuser/mpi/MPI_Codes/code/test2.txt
/home/mpiuser/mpi/MPI_Codes/code/test3.txt

Answer 1

您的 readFilesList 函数非常令人困惑，我相信它不会做您想要做的事情，但也许我只是没有正确理解它。我相信它应该从每个进程的列表文件中收集一堆文件名。每个进程都有不同的集合。它不会那样做，但这不是问题，即使这会做你想做的事，后续的 MPI IO 也不会工作。

读取文件时，您使用 MPI_File_read_all 和 MPI_COMM_WORLD 作为通信器。这需要所有进程都参与读取该文件。现在，如果每个进程都读取不同的文件，这显然是行不通的。

因此，您的实现存在一些问题，虽然我无法真正解释您描述的行为，但我宁愿先开始尝试修复它们，然后再详细调试，可能会出现什么问题。

我的印象是，您希望有一个这样的算法：

• 读取文件名列表

• 将该文件列表平均分配给所有进程

• 让每个进程处理自己的文件集

• 对来自此处理的数据做一些事情

我建议用以下方法试试这个：

• 在单个进程上读取列表（无 MPI IO）

• 将文件列表分散到所有进程，这样所有进程的工作量都相同

• 让每个进程独立并以串行方式处理其文件列表（串行文件访问和处理）

• 根据需要使用 MPI 减少一些数据

我相信，这将是您场景中最好（最简单和最快）的策略。请注意，这里根本不涉及 MPI IO。我不认为在第一步中对文件列表进行一些复杂的分布式读取会在这里产生任何优势，并且在实际处理中它实际上是有害的。您的流程越独立，通常您的可扩展性就越好。