Разбор большого файла с MPI в C++

Question

У меня есть программа на C++, в которой я хочу разобрать огромный файл, ища какое-то регулярное выражение, которое я реализовал. Программа работала нормально, когда выполнялась последовательно, но затем я хотел запустить ее, используя MPI.

Я начал адаптацию к MPI дифференцирования мастера (тот, который координирует выполнение) из рабочих (те, которые анализируют файл параллельно) в главной функции . Что-то вроде этого:

MPI::Init(argc, argv); 
... 

if(rank == 0) { 
    ... 

    // Master sends initial and ending byte to every worker 
    for(int i = 1; i < total_workers; i++) { 
     array[0] = (i-1) * first_worker_file_part; 
     array[1] = i * first_worker_file_part; 
     MPI::COMM_WORLD.Send(array, 2, MPI::INT, i, 1); 
    } 
} 

if(rank != 0) 
    readDocument(); 

... 

MPI::Finalize(); 

мастер направит каждому работника массив с 2 позиции, которая содержит байт, где он будет начать чтение файла в позиции и байт, где он необходимо прекратить чтение в позиции .

readDocument() функция выглядит следующим образом теперь (не разбор, просто каждый работник читает его часть файла):

void readDocument() 
{ 
    array = new int[2]; 
    MPI::COMM_WORLD.Recv(array, 10, MPI::INT, 0, 1, status); 
    int read_length = array[1] - array[0]; 
    char* buffer = new char [read_length]; 

    if (infile) 
    { 
     infile.seekg(array[0]); // Start reading in supposed byte 
     infile.read(buffer, read_length); 
    } 
} 

Я пробовал различные примеры, от написания файл выводит показание для запуска его с различным количеством процессов. Случается, что когда я запускаю программу с 20 процессами вместо 10, например, она длится в два раза больше времени для чтения файла. Я ожидал, что это будет почти в половину времени, и я не могу понять, почему это происходит.

Кроме того, в другом вопросе, я хочу, чтобы сделать мастер ожидания для всех рабочих завершить их выполнение, а затем распечатать последний раз. Есть ли способ «блокировать» его, пока рабочие обрабатывают? Как cond_wait в C pthreads?

Answer 1

По моему опыту люди, работающие над компьютерными системами с параллельными файловыми системами, как правило, знают об этих параллельных файловых системах, поэтому ваш вопрос помечает вас, как изначально не работающего над такой системой.

Без специальной поддержки аппаратной поддержки чтение из одного файла сводится к тому, что система позиционирует одну головку чтения и считывает последовательность байтов с диска на память. Эта ситуация существенно не изменена сложными реалиями многих современных файловых систем, таких как RAID, которые могут фактически хранить файл на нескольких дисках. Когда несколько процессов запрашивают операционную систему для доступа к файлам одновременно, o/s отправляет доступ к диску по определенному понятию, возможно, справедливости, так что ни один процесс не умирает. В худшем случае o/s тратит столько времени на переключение доступа к диску от процесса к процессу, что скорость чтения значительно снижается. Наиболее эффективным с точки зрения пропускной способности является то, что один процесс может читать целый файл за один раз, в то время как другие процессы выполняют другие функции.

В этой ситуации несколько процессов, связанных с ограниченными ресурсами ввода-вывода диска, применяются независимо от того, являются ли эти процессы частью параллельной, MPI (или аналогичной) программы или полностью отдельных программ, работающих одновременно.

Воздействие - это то, что вы наблюдаете - вместо 10 процессов, каждый из которых ожидает получения своей 1/10-й доли файла, у вас есть 20 процессов, каждый из которых ждет их 1/20-й доли.О, вы плачете, но каждый процесс читает только половину данных, поэтому вся банда должна занять столько же времени, чтобы получить файл. Нет, я отвечаю, вы забыли добавить время, которое требуется для позиционирования o/s, и переместить головки чтения/записи между доступом. Время чтения включает в себя латентность (сколько времени требуется, чтобы чтение начиналось после запроса) и пропускной способности (насколько быстро система ввода-вывода может передавать байты туда и обратно).

Это должно быть легко придумать некоторые разумные оценки задержки и пропускной способности, которая объясняет два раза дольше чтении 20 процессов как на 10.

Как вы можете решить эту проблему? Вы не можете, не без параллельной файловой системы. Но вы можете обнаружить, что мастер-процесс читает весь файл, а затем выставляет его быстрее, чем ваш текущий подход. Вы не можете, вы можете просто найти, что текущий подход является самым быстрым для всего вашего расчета. Если время чтения составляет, скажем, 10% от общего времени вычисления, вы можете решить, что это разумные накладные расходы.

Answer 2

Чтобы добавить к правильному ответу High Performance Mark, можно использовать MPI-IO для чтения файла, предоставляя (в данном случае) намеки на то, что процедуры ввода-вывода не считываются с каждого процессора; но этот же код с измененным (или пустым) MPI_Info должен также иметь возможность использовать параллельную файловую систему, если вы перейдете к кластеру с одним. Для наиболее распространенной реализации MPI-IO, Romio, руководство, описывающее, какие подсказки доступны, - here; в частности, мы используем

MPI_Info_set(info, "cb_config_list","*:1"); 

, чтобы установить количество считывающих устройств на один узел. Код ниже позволит вам попробовать прочитать файл с помощью MPI-IO или POSIX (например, искать).

#include <iostream> 
#include <fstream> 
#include <mpi.h> 

void partitionFile(const int filesize, const int rank, const int size, 
        const int overlap, int *start, int *end) { 
    int localsize = filesize/size; 
    *start = rank * localsize; 
    *end = *start + localsize-1; 

    if (rank != 0)  *start -= overlap; 
    if (rank != size-1) *end += overlap; 
} 

void readdataMPI(MPI_File *in, const int rank, const int size, const int overlap, 
       char **data, int *ndata) { 
    MPI_Offset filesize; 
    int start; 
    int end; 

    // figure out who reads what 

    MPI_File_get_size(*in, &filesize); 
    partitionFile((int)filesize, rank, size, overlap, &start, &end); 

    *ndata = end - start + 1; 

    // allocate memory 
    *data = new char[*ndata + 1]; 

    // everyone reads in their part 
    MPI_File_read_at_all(*in, (MPI_Offset)start, *data, 
         (MPI_Offset)(*ndata), MPI_CHAR, MPI_STATUS_IGNORE); 
    (*data)[*ndata] = '\0'; 
} 

void readdataSeek(std::ifstream &infile, int array[2], char *buffer) 
{ 
    int read_length = array[1] - array[0]; 
    if (infile) 
    { 
     infile.seekg(array[0]); // Start reading in supposed byte 
     infile.read(buffer, read_length); 
    } 
} 

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

    MPI_File in; 
    int rank, size; 
    int ierr; 

    MPI_Init(&argc, &argv); 
    MPI_Comm_rank(MPI_COMM_WORLD, &rank); 
    MPI_Comm_size(MPI_COMM_WORLD, &size); 

    if (argc != 3) { 
     if (rank == 0) 
      std::cerr << "Usage: " << argv[0] << " infilename [MPI|POSIX]" << std::endl; 
     MPI_Finalize(); 
     return -1; 
    } 

    std::string optionMPI("MPI"); 

    if (!optionMPI.compare(argv[2])) { 
     MPI_Info info; 
     MPI_Info_create(&info); 
     MPI_Info_set(info, "cb_config_list","*:1"); // ROMIO: one reader per node 
                // Eventually, should be able to use io_nodes_list or similar 

     ierr = MPI_File_open(MPI_COMM_WORLD, argv[1], MPI_MODE_RDONLY, info, &in); 
     if (ierr) { 
      if (rank == 0) 
       std::cerr << "Usage: " << argv[0] << " Couldn't open file " << argv[1] << std::endl; 
      MPI_Finalize(); 
      return -1; 
     } 

     const int overlap=1; 
     char *data; 
     int ndata; 
     readdataMPI(&in, rank, size, overlap, &data, &ndata); 

     std::cout << "MPI: Rank " << rank << " has " << ndata << " characters." << std::endl; 

     delete [] data; 

     MPI_File_close(&in); 
     MPI_Info_free(&info); 
    } else { 
     int fsize; 
     if (rank == 0) { 
      std::ifstream file(argv[1], std::ios::ate); 
      fsize=file.tellg(); 
      file.close(); 
     } 
     MPI_Bcast(&fsize, 1, MPI_INT, 0, MPI_COMM_WORLD); 

     int start, end; 
     partitionFile(fsize, rank, size, 1, &start, &end); 

     int array[2] = {start, end}; 
     char *buffer = new char[end-start+2]; 

     std::ifstream infile; 
     infile.open(argv[1], std::ios::in); 
     readdataSeek(infile, array, buffer); 
     buffer[end-start+1] = '\0'; 
     std::cout << "Seeking: Rank " << rank << " has " << end-start+1 << " characters." << std::endl; 

     infile.close() ; 
     delete [] buffer; 
    } 

    MPI_Finalize(); 
    return 0; 
} 

На моем рабочем столе, я не получаю большую часть разницы в производительности, даже oversubscribing ядер (например, используя много испрашивает):

$ time mpirun -np 20 ./read-chunks moby-dick.txt POSIX 
Seeking: Rank 0 has 62864 characters. 
[...] 
Seeking: Rank 8 has 62865 characters. 

real 0m1.250s 
user 0m0.290s 
sys 0m0.190s 

$ time mpirun -np 20 ./read-chunks moby-dick.txt MPI 
MPI: Rank 1 has 62865 characters. 
[...] 
MPI: Rank 4 has 62865 characters. 

real 0m1.272s 
user 0m0.337s 
sys 0m0.265s