Parsing large file with MPI in C++

Question

I have a C++ program in which I want to parse a huge file, looking for some regex that I've implemented. The program was working ok when executed sequentially but then I wanted to run it using MPI.

I started the adaptation to MPI by differentiating the master (the one who coordinates the execution) from the workers (the ones that parse the file in parallel) in the main function. Something like this:

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();

The master will send to every worker an array with 2 position that contains the byte where it will start the reading of the file in position 0 and the byte where it needs to stop reading in position 1.

The readDocument() function looks like this by now (not parsing, just each worker reading his part of the file):

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);
    }
}

I've tried different examples, from writing to a file the output of the reading to running it with different number of processes. What happens is that when I run the program with 20 processes instead of 10, for example, it lasts twice the time to read the file. I expected it to be nearly half the time and I can't figure why this is happening.

Also, in a different matter, I want to make the master wait for all the workers to complete their execution and then print the final time. Is there any way to "block" him while the workers are processing? Like a cond_wait in C pthreads?

Answer 1

To add to High Performance Mark's correct answer, one can use MPI-IO to do the file reading, providing (in this case) hints to the IO routines not to read from every processor; but this same code with a modified (or empty) MPI_Info should be able to take advantage of a parallel file system as well should you move to a cluster that has one. For the most common implementation of MPI-IO, Romio, the manual describing what hints are available is here; in particular, we're using

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

to set the number of readers to be one per node. The code below will let you try reading the file using MPI-IO or POSIX (eg, seek).

#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;
}

On my desktop, I don't get much of a performance difference, even oversubscribing the cores (eg, using lots of seeks):

$ 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

Answer 2

In my experience people working on computer systems with parallel file systems tend to know about those parallel file systems so your question marks you out, initially, as someone not working on such a system.

Without specific hardware support reading from a single file boils down to the system positioning a single read head and reading a sequence of bytes from the disk to memory. This situation is not materially altered by the complex realities of many modern file systems, such as RAID, which may in fact store a file across multiple disks. When multiple processes ask the operating system for access to files at the same time the o/s parcels out disk access according to some notion, possibly of fairness, so that no process gets starved. At worst the o/s spends so much time switching disk access from process to process that the rate of reading drops significantly. The most efficient, in terms of throughput, approach is for a single process to read an entire file in one go while other processes do other things.

This situation, multiple processes contending for scarce disk i/o resources, applies whether or not those processes are part of a parallel, MPI (or similar) program or entirely separate programs running concurrently.

The impact is what you observe -- instead of 10 processes each waiting to get their own 1/10th share of the file you have 20 processes each waiting for their 1/20th share. Oh, you cry, but each process is only reading half as much data so the whole gang should take the same amount of time to get the file. No, I respond, you've forgotten to add the time it takes the o/s to position and reposition the read/write heads between accesses. Read time comprises latency (how long does it take reading to start once the request has been made) and throughput (how fast can the i/o system pass the bytes to and fro).

It should be easy to come up with some reasonable estimates of latency and bandwidth that explains the twice as long reading by 20 processes as by 10.

How can you solve this ? You can't, not without a parallel file system. But you might find that having the master process read the whole file and then parcel it out to be faster than your current approach. You might not, you might just find that the current approach is the fastest for your whole computation. If read time is, say, 10% of total computation time you might decide it's a reasonable overhead to live with.