main.cpp

#include <bits/stdc++.h>
#include <pthread.h>
#include <ctype.h>
#include <string.h>

using namespace std;

// Structure to hold shared reducer data
struct ReducerData
{
    pthread_mutex_t lock; // Mutex for synchronized access
    vector<map<string, set<int>>> lists; // Vector for reducer lists
};

// Structure to hold thread information
struct threadData
{
    int threadID; // Thread ID
    vector<pair<string, int>> *file_data; // Vector with file names and IDs
    int nr_mappers; // Number of mapper threads
    int nr_reducers; // Number of reducer threads
    pthread_mutex_t *mutex; // Mutex for accessing shared file data
    pthread_barrier_t *barrier; // Barrier for synchronizing mappers and reducers
    ReducerData *reducerData; // Pointer to shared reducer data (files and mutex)
};

// Assign files to each reducer
vector<int> reducerFileIndex(int nrReducers)
{
    vector<int> indexes(26, 0);
    int range = 26 / nrReducers;

    for (int i = 0; i < 26; ++i)
    {
        indexes[i] = min(i / range, nrReducers - 1);
    }

    return indexes;
}

// Add the word in the list of the responsible reducer
void addWord(const string &word, int fileID, vector<map<string, set<int>>> &reducerWords, int numReducers)
{
    static vector<int> indexes = reducerFileIndex(numReducers);
    if (word.empty() || !isalpha(word[0]))
        return;

    char firstLetter = tolower(word[0]);
    int reducerIndex = indexes[firstLetter - 'a'];

    reducerWords[reducerIndex][word].insert(fileID);
}

// Compare function for sorting words
bool compare(const pair<string, set<int>> &wordA, const pair<string, set<int>> &wordB)
{
    if (wordA.second.size() != wordB.second.size())
    {
        return wordA.second.size() > wordB.second.size();
    }

    return wordA.first < wordB.first;
}

// Assign a file to a mapper thread
bool assignFileToMapper(threadData *t, pair<string, int> &file_data)
{
    pthread_mutex_lock(t->mutex);
    if (!t->file_data->empty())
    {
        auto it = prev(t->file_data->end());
        file_data = *it;
        t->file_data->erase(it);
        pthread_mutex_unlock(t->mutex);
        return true;
    }
    pthread_mutex_unlock(t->mutex);
    return false;
}

// Process files and distribute words to the responsible reducers
vector<map<string, set<int>>> processFilesForReducers(threadData *t)
{
    vector<map<string, set<int>>> reducerWords(t->nr_reducers);
    pair<string, int> file_data;

    while (assignFileToMapper(t, file_data))
    { // Process files assigned to the mapper
        ifstream file(file_data.first);
        string line;

        while (getline(file, line))
        {
            istringstream stream(line);
            string word;

            while (stream >> word)
            { // Extract words from each line
                word.erase(remove_if(word.begin(), word.end(), [](char c)
                                     { return !isalpha(c); }),
                           word.end()); // Remove non-alphabetic characters
                transform(word.begin(), word.end(), word.begin(), ::tolower); // Convert to lowercase

                if (!word.empty())
                {
                    addWord(word, file_data.second, reducerWords, t->nr_reducers);
                }
            }
        }
        file.close();
    }

    return reducerWords;
}

// Mapper function to process files and update the shared reducer data
void mapper(threadData *t)
{
    vector<map<string, set<int>>> reducerWords = processFilesForReducers(t);

    // Add reducerWords to shared data
    pthread_mutex_lock(&t->reducerData->lock);

    for (int i = 0; i < t->nr_reducers; ++i)
    {
        auto &localMap = reducerWords[i];
        auto &sharedMap = t->reducerData->lists[i];

        for (const auto &entry : localMap)
        {
            const string &word = entry.first;
            const set<int> &fileIds = entry.second;
            sharedMap[word].insert(fileIds.begin(), fileIds.end());
        }
    }

    pthread_mutex_unlock(&t->reducerData->lock);
}

// Calculate the letter range handled by a reducer
pair<char, char> calculateLetterRange(int reducer_id, int nr_reducers, int nr_mappers)
{
    int range = 26 / nr_reducers;
    int range_rest = 26 % nr_reducers;

    char startLetter = 'a' + (reducer_id - nr_mappers) * range;
    char endLetter = 'a' + (reducer_id - nr_mappers + 1) * range - 1;

    if (reducer_id == nr_mappers + nr_reducers - 1)
    {
        endLetter += range_rest;
    }

    return {startLetter, endLetter};
}

// Group words by their first letter
map<char, vector<pair<string, set<int>>>> groupWords(
    const map<string, set<int>> &ReducerData,
    char startLetter,
    char endLetter)
{
    map<char, vector<pair<string, set<int>>>> groupedWords;

    for (const auto &entry : ReducerData)
    {
        char firstChar = entry.first[0];
        if (firstChar >= startLetter && firstChar <= endLetter)
        {
            groupedWords[firstChar].push_back({entry.first, entry.second});
        }
    }

    return groupedWords;
}

// Write grouped words to output files
void writeGroupedWords(
    const map<char, vector<pair<string, set<int>>>> &groupedWords,
    function<bool(const pair<string, set<int>> &, const pair<string, set<int>> &)> compare)
{
    // Iterate over each group of words
    for (const auto &entry : groupedWords)
    {
        char letter = entry.first;
        string filename = string(1, letter) + ".txt";
        ofstream outputFile(filename); // Open the output file for writing

        auto string = entry.second; // The list of word and file ID pairs for the letter group
        stable_sort(string.begin(), string.end(), compare); // Sort the words using compare

        // Write each word and its associated file IDs to the output file
        for (const auto &wordEntry : string)
        {
            ostringstream lineStream; // Stream to construct the output line
            lineStream << wordEntry.first << ":[";

            // Write the file IDs associated with the word
            for (auto it = wordEntry.second.begin(); it != wordEntry.second.end(); ++it)
            {
                if (it != wordEntry.second.begin())
                {
                    lineStream << " ";
                }
                lineStream << *it;
            }

            lineStream << "]\n";
            outputFile << lineStream.str(); // Write the constructed line to the file
        }

        outputFile.close();
    }
}

// Reducer function to group words and write to files
void reducer(threadData *t)
{
    auto [startLetter, endLetter] = calculateLetterRange(t->threadID, t->nr_reducers, t->nr_mappers);

    const auto &ReducerData = t->reducerData->lists[t->threadID - t->nr_mappers];
    auto groupedWords = groupWords(ReducerData, startLetter, endLetter);

    writeGroupedWords(groupedWords, compare);
}

// Thread function
void *func(void *arg)
{
    threadData *t = (threadData *)arg;

    if (t->threadID < t->nr_mappers)
        mapper(t);

    pthread_barrier_wait(t->barrier);

    if (t->threadID >= t->nr_mappers)
        reducer(t);

    return NULL;
}

// Initialize thread data
void initializeThreadStruct(
    threadData &thread_data,
    vector<pair<string, int>> *file_data,
    pthread_mutex_t *mutex,
    pthread_barrier_t *barrier,
    ReducerData *reducerData,
    int threadID,
    int nr_mappers,
    int nr_reducers)
{
    thread_data.file_data = file_data;
    thread_data.mutex = mutex;
    thread_data.barrier = barrier;
    thread_data.reducerData = reducerData;
    thread_data.threadID = threadID;
    thread_data.nr_mappers = nr_mappers;
    thread_data.nr_reducers = nr_reducers;
}

int main(int argc, char *argv[])
{
    // Input data
    int nr_mappers = stoi(argv[1]);
    int nr_reducers = stoi(argv[2]);
    ifstream files(argv[3]);

    vector<pair<string, int>> file_data;
    vector<threadData> thread_data(nr_mappers + nr_reducers);
    vector<pthread_t> threads(nr_mappers + nr_reducers);

    // Save the files in file_data
    if (files)
    {
        int file_count;
        files >> file_count;
        files.ignore();

        for (int i = 0; i < file_count; ++i)
        {
            string fileName;
            getline(files, fileName);
            file_data.emplace_back(fileName, i + 1);
        }
    }
    
    // Initialize threads
    ReducerData reducerData;
    reducerData.lists.resize(nr_reducers);
    pthread_mutex_init(&reducerData.lock, NULL);

    pthread_barrier_t barrier;
    pthread_barrier_init(&barrier, NULL, nr_mappers + nr_reducers);

    pthread_mutex_t mutex;
    pthread_mutex_init(&mutex, NULL);

    for (int i = 0; i < nr_mappers + nr_reducers; ++i)
    {
        initializeThreadStruct(
            thread_data[i],
            &file_data,
            &mutex,
            &barrier,
            &reducerData,
            i,
            nr_mappers,
            nr_reducers);

        pthread_create(&threads[i], NULL, func, (void *)&thread_data[i]);
    }

    for (int i = 0; i < nr_mappers + nr_reducers; i++)
    {
        pthread_join(threads[i], NULL);
    }

    pthread_barrier_destroy(&barrier);
    pthread_mutex_destroy(&mutex);
    pthread_mutex_destroy(&reducerData.lock);

    return 0;
}