How could I decompose a text file in smaller chunks (using C++ on Unix)?

user_14534752 0
Viewed 179

I am working on a school assignment where I have to make a sequential code be concurrent. What the code does is called a MapReduce, more specifically it counts the number of times all words have appeared in all the input files.

  • Input: A directory that contains all text files to be processed.
  • Output: A directory that contains a text file with the result.

The implementation that we are given is made of 4 stages; Split Map Shuffle and Reduce. The first 3 have to happen concurrently between them and when all 3 of them are finished, Reduce takes place.

Concurrency, apart from happening between those 3 stages, has to also happen inside each one of the stages. In order to do that, we are told to divide the work between a certain number of threads (each stage has its threads).

From now on I am going to talk only about the Split and Map stage, which are the ones I am having trouble with right now.

The Split and Map stage will have one thread for each 'partition' (they will have the same amount of threads because each Split thread will have a Map thread associated), and there will be one 'partition' for each file in the input directory, except for big files, where I'll have to generate a 'partition' every 8 MB.

This exception is what I am trying to solve, let me explain one more thing and I'll get to it.

Since Split and Map have to have the same amount of threads, what I did is create a function that computes the number of partitions that will have to be created, in order to be able to tell Split and Map how many threads do they have to create when I start them.

What Split threads do is take the 'partition' they have been assigned (it will be an entire file in case its size is less than 8 MB) and send it to its associated Map thread line by line (by writing them in a thread-safe queue).

Okay so here's my problem, I thought to make the Map threads (who read from the mentioned queue and do their thing with each line, doesn't matter right now) read from the queue until they read an EOF, meaning that it's asociated Split thread has ended writing it's 'partition' to the queue, but that would only work with 'partitions' that represent an entire file.

So, my questions are:

  1. Which options do I have to decompose a file into chunks of 8 MB?
  2. How could I make a thread, who's been given a partition that represents the first chunk of many of a big file, know when to stop reading?
  3. Since there will be a fraction of time where Map threads will be trying to read from the queue but the Split threads won't have written anything yet, how could I make them 'wait' for something to be written in the queue?

This is the function that counts how many threads are going to be needed, I would like to make it generate an array of file descriptors containing one for each 'partition'

int MapReduce::getSplitNum(){
int split_num = 0;
char file_path[264];

DIR* dir = opendir(InputPath);
struct dirent* entity;
unsigned char isFile =0x8; // El valor que pren entity->d_type quan es tracta d'un fitxer

while ((entity = readdir(dir)) != NULL)
{
    // Evitem el directori que conté els fitxers (.) i el directori anterior (..).
    if( strcmp(entity->d_name, ".")!=0 && strcmp(entity->d_name, "..")!=0 && entity->d_type == isFile )
    {
        struct stat file_status;
        sprintf(file_path,"%s/%s",InputPath, entity->d_name);
        stat(file_path, &file_status);

        long file_size = file_status.st_size;

        // DEBUG: printf("Fitxer: %s\t Mida: %ld\n", entity->d_name, file_status.st_size);
        if (file_size < MAX_SPLIT_SIZE)
        {   
            split_num++;
        }
        else
        {
            long restant = file_size - MAX_SPLIT_SIZE;
            split_num = split_num + 2; // Sumem 2 perquè al ser un arxiu gran com a mínim usarà 2 splits, més els que vagi afegint el bucle while.

            while (restant > MAX_SPLIT_SIZE)
            {
                restant = restant - MAX_SPLIT_SIZE;
                split_num++;
            }
        }
    }
}
closedir(dir);
return split_num;

}

I guess there will be a ton of ways of decomposing a file, but I'd be interested in which would be the 'good practice' one.

Thanks!

thread-safety pthreads multithreading file c++
1 Answers

Maybe something like this? This function decomposes a file into kind of equal pieces, the last one being differen (the rest of the file). This could be modified to decompose files into chunks of specific size. I wrote this just now, it seems to work for me, but of course further testing is necessary. And of course I am not sure that this is the most optimal solution. This produces n files called 0, 1, ..., n.

#include <iostream>
#include <fstream>

void decompose_file (const std::string& filepath, const int number_of_files) {
    std::ifstream infile(filepath);

    infile.seekg(0, std::ios::end);
    size_t length = infile.tellg();
    infile.seekg(0, std::ios::beg);
    
    for (int i = 0; i < number_of_files; ++i) {
        char * buffer;
        size_t chunk_size = 0;
        if (i != number_of_files - 1) {
            chunk_size = length / number_of_files;
        }
        else {
            chunk_size = length - ((number_of_files - 1) * (length / number_of_files));
        }
        buffer = new char[chunk_size];
        
        infile.read (buffer, chunk_size);
        
        std::ofstream outfile (std::to_string(i));
        if (outfile.is_open()) {
            outfile.write(buffer, chunk_size);
            outfile.close();
        }
        delete[] buffer;
    }
    infile.close();
}

int main (int argc, char* argv[]) {
    decompose_file("my_file.txt", 4);
    return 0;
}
user_91566780
Related