tema1.cpp

#include <iostream>
#include <fstream>
#include <vector>
#include <string>
#include <pthread.h>
#include <math.h>
#include <sstream>
#include <unordered_map>
#include <bits/stdc++.h>
using namespace std;


typedef struct data {
    unordered_map<string,vector<int>> *allData;

    vector<vector<pair<string,vector<int>>>> *wordsInfo;
    vector<pair<string,int>> *fileNames;

    pthread_mutex_t *sharedFileVector;
    pthread_mutex_t *sharedDataMap;

    pthread_barrier_t *waitForMap;
    pthread_barrier_t *waitMappersFinish;

    string currentFileName;

    int currentFileID;
    int threadID;
    int nrThreads;
} DATA;

typedef struct dataRed {
    vector<vector<pair<string,vector<int>>>> *wordsInfo;
    vector <ofstream> *files;
    vector<int> *indices;

    pthread_barrier_t *waitMappersFinish;
    pthread_barrier_t *waitForReducers;
    pthread_mutex_t *pickNextFile;
    int threadID;
    int nrThreads;
    unsigned start;
    unsigned ending;
} REDUCERS;

int compare2(pair<string, vector<int>> a, pair<string, vector<int>> b) {
    // comparam valorile perechiilor, mai intai dupa numarul de fisiere
    // in care cuvintele apar
    if (a.second.size() > b.second.size()) {
        return 1;
    } else if (a.second.size() == b.second.size()) {
        // apoi, daca apar in la fel de multe fisiere
        // comparam cuvintele
        if(a.first > b.first) {
            return 0;
        } else {
            return 1;
        }
    }
    return 0;
}

int min(int x,int y) {
    if( x > y) {
        return y;
    } else {
        return x;
    }
}


void *reducerFunction(void *argc) {
    REDUCERS *dt = (REDUCERS *)argc;
    // asteptam ca mappers sa termine de citit
    // fisierele
    pthread_barrier_wait(dt->waitMappersFinish);

    int reading = 1;
    while (reading) {
        int poz = -1;
        // alegem cate o litere din vector, pe care urmeaza sa o sortam
        // si sa o scriem in fisierul corespunzator
        pthread_mutex_lock(dt->pickNextFile);
        if ((*dt->indices).size() != 0) {
            poz = (*dt->indices)[0];
            // stergem litera pe care am ales-o
            (*dt->indices).erase((*dt->indices).begin());
        } else {
            // daca nu mai sunt litere, atunci thread-ul si a terminat
            // executia
            reading  = 0;
        }
        pthread_mutex_unlock(dt->pickNextFile);
        if ((*dt->wordsInfo)[poz].size() == 0) {
                // daca nu exista cuvinte care sa inceapa cu litera pe care am ales-o
                // cautam alta
                continue;
        }
        if (reading == 0) {
            // daca thread-ul nu a putut alege o litera, fiindca au fost deja toata luate
            // ne oprim
            break;
        }
        if ((*dt->wordsInfo)[poz].size() != 1) {
            // sortam datele cu un comparator custom
            sort((*dt->wordsInfo)[poz].begin(), (*dt->wordsInfo)[poz].end(),compare2);
        }
        // scriem datele in fisierul corespunzator
        // nu este nevoie de mutex-uri, intrucat datele
        // vor fi scrise in fisiere separate de thread-uri diferite
        for (unsigned int i = 0; i < (*dt->wordsInfo)[poz].size(); i++) {
                (*dt->files)[poz]  <<  (*dt->wordsInfo)[poz][i].first << ":[";
                // sortam si vectorul mai mic, intern, care ne spune fisierele in care apare un cuvant
                // folosind comparator-ul default
                sort((*dt->wordsInfo)[poz][i].second.begin(), (*dt->wordsInfo)[poz][i].second.end());
                for (unsigned int j = 0;j < (*dt->wordsInfo)[poz][i].second.size();j++) {
                    if (j != (*dt->wordsInfo)[poz][i].second.size() -1) {
                        (*dt->files)[poz] << (*dt->wordsInfo)[poz][i].second[j] << " ";
                    } else {
                        (*dt->files)[poz] << (*dt->wordsInfo)[poz][i].second[j] << "]";
                    }
                }
            (*dt->files)[poz] << endl;
        }
    }

    return NULL;
}

void *mapFunction(void *argc) {
    DATA *dt = (DATA *)argc;

    string line;
    int reading = 1;
    while (reading) {
        // continui sa citesc cat timp am un fisier
        // de citit
        ifstream fin(dt->currentFileName);
        unordered_map<string,vector<int>> internMap;
        // citesc linie cu linie, pentru a nu face foarte multe call-uri
        // catre fin
        while (getline(fin, line)) {

            stringstream stream(line);
            string word;
            // folosesc stream-uri pentru a separa linia prin spatii
            while (stream >> word) {
                string word_copy;
                // elimin orice nu este litera mica, si fac literele mari mici
                for (unsigned i = 0; i < word.size(); i++) {
                    if(word[i] >= 'a' && word[i] <= 'z') {
                        word_copy.push_back(word[i]);
                    } else if(word[i] >= 'A' && word[i] <= 'Z') {
                        word_copy.push_back(word[i] + 32);
                    }
                }
                // daca nu era nici o litera a alfabetului, ignor cuvantul
                // trec la urmatorul
                if(word_copy.size() == 0) {
                    continue;
                }
                word = word_copy;
                // adaug cuvantul gasit, precum si id-ul fisierului in map-ul intern.
                // nu este nevoie sa ma asigur ca nu a mai aparut inca o data, intrucat
                // dupa fiecare fisier adaug intr-un map comun.
                internMap[word] = {dt->currentFileID};
            }

        }
        // scriem datele in map-ul comun
        pthread_mutex_lock(dt->sharedDataMap);
        for (auto it = internMap.begin(); it != internMap.end(); it++) {
            auto found = (*dt->allData).find(it->first);
            // daca cuvantul a fost deja gasit in map-ul nostru
            if (found != (*dt->allData).end()) {
                // atunci va trebui sa comasam vectorii continuti de cuvinte
                for (unsigned i = 0; i < it->second.size(); i++) {
                    (*dt->allData)[it->first].push_back(it->second[i]);
                }

            } else {
                // daca nu a fost gasit, doar il adaugam
                (*dt->allData).insert({it->first, it->second});
            }
        }
        internMap.clear();
        pthread_mutex_unlock(dt->sharedDataMap);

        pthread_mutex_lock(dt->sharedFileVector);
        // incer sa iau un alt fisier din care sa incep sa citesc
        if ((*dt->fileNames).size() != 0) {
           dt->currentFileName = (*dt->fileNames)[0].first;
           dt->currentFileID = (*dt->fileNames)[0].second;
           (*dt->fileNames).erase((*dt->fileNames).begin());
           reading = 1;
        } else {
          reading = 0;
        }
        // daca au fost deja luate toate, ma opresc
        // inchid fisierul pe care l-am citit
        fin.close();
        pthread_mutex_unlock(dt->sharedFileVector);
    }

    pthread_barrier_wait(dt->waitForMap);
    // acum, pentru a putea sorta datele, voi muta datele intr-un vector
    // intrucat nu pot sorta un unordered map ( pe care l-am folosit pentru eficienta sporita in cautare O(log n))
    int start = dt->threadID * (float ((*dt->allData).size())/dt->nrThreads);
    int ending = min((dt->threadID + 1) * (float ((*dt->allData).size())/dt->nrThreads), (*dt->allData).size());
    auto startIt =(*dt->allData).begin();
    advance(startIt, start);
    auto endIt = (*dt->allData).begin();
    advance(endIt, ending);

    pthread_mutex_lock(dt->sharedDataMap);
    for (auto i = startIt; i != endIt; i++) {
        int index = i->first[0] - 97;
        (*dt->wordsInfo)[index].push_back(make_pair(i->first, i->second));
    }
    pthread_mutex_unlock(dt->sharedDataMap);
    // astept sa se termine toti mappers inainte sa inceapa reducers
    pthread_barrier_wait(dt->waitMappersFinish);
    return NULL;
}

void initMutexes(pthread_mutex_t *m1, pthread_mutex_t *m2, pthread_mutex_t *m3);
void initBarriers(pthread_barrier_t *m1, pthread_barrier_t *m2, pthread_barrier_t *m3, int map, int red);
void initFileNames(vector<pair<string,int>> *fileNames, int *noFiles, ifstream *fin, vector <ofstream> *files);

int main(int argv, char *argc[])
{
    // sanity check
    if (argv < 4) {
        cout << "Proper usage: ./program N_MAPPERS N_REDUCERS INPUT_FILE" << endl;
        exit(1);
    }

    pthread_mutex_t sharedFileVector,sharedDataMap, pickNextFile;
    int noFiles, numMaps = atoi(argc[1]), poz = 0, numReducers = atoi(argc[2]);

    // sanity check
    if (numMaps * numReducers == 0) {
        cout << "Improper usage, the N_MAPPERS and N_REDUCERS should be of type int bigger than 0" << endl;
        exit(1);
    }

    vector<int> indices;
    for (int i = 0; i < 26; i++) {
        indices.push_back(i);
    }

    DATA data[numMaps];
    REDUCERS dataReducers[numReducers];
    ifstream fin(argc[3]);
    pthread_t threads[numMaps + numReducers + 1];

    vector<pair<string,int>> fileNames;
    unordered_map<string,vector<int>> allData;
    vector<vector<pair<string,vector<int>>>> wordsInfo(26);

    pthread_barrier_t waitMappersFinish,waitForMap,waitForReducers;

    vector <ofstream> files;

    // init data
    initMutexes(&sharedFileVector, &sharedDataMap, &pickNextFile);
    initBarriers(&waitMappersFinish, &waitForMap, &waitForReducers, numMaps, numReducers);
    initFileNames(&fileNames, &noFiles, &fin, &files);

    // crearea thread-urilor si asignarea resurselor partajate
    for (int i = 0; i < numMaps + numReducers; i++) {
        if (i < numMaps) {
            data[i].currentFileName = fileNames[0].first;
            data[i].currentFileID = fileNames[0].second;
            data[i].sharedFileVector = &sharedFileVector;
            data[i].sharedDataMap = &sharedDataMap;
            data[i].threadID = i;
            fileNames.erase(fileNames.begin());
            data[i].fileNames = &fileNames;
            data[i].waitMappersFinish = &waitMappersFinish;
            data[i].allData = &allData;
            data[i].wordsInfo = &wordsInfo;
            data[i].waitForMap = &waitForMap;
            data[i].nrThreads = numMaps;
            int r = pthread_create(&threads[i], NULL, mapFunction, &data[i]);
            if (r != 0) {
                cout << "Failed to create thread: " << i << endl;
                exit(1);
            }
        } else {
            dataReducers[poz].waitMappersFinish = &waitMappersFinish;
            dataReducers[poz].threadID = poz;
            dataReducers[poz].wordsInfo = &wordsInfo;
            dataReducers[poz].nrThreads = numReducers;
            dataReducers[poz].waitForReducers = &waitForReducers;
            dataReducers[poz].files = &files;
            dataReducers[poz].indices = &indices;
            dataReducers[poz].pickNextFile = &pickNextFile;
            int r = pthread_create(&threads[i], NULL, reducerFunction, &dataReducers[poz]);
            if (r != 0) {
                cout << "Failed to create thread: " << poz << endl;
                exit(1);
            }
            poz++;
        }

    }

    void *status;
    // asteptam ca thread-urile sa isi termine executia
    for (int id = 0; id < numMaps + numReducers; id++) {
        int r = pthread_join(threads[id], &status);
        if (r != 0) {
            cout << "Failed to sync threads" << endl;
            exit(1);
        }

    }

    // eliberam toate resursele folosite.
    pthread_mutex_destroy(&sharedFileVector);
    pthread_mutex_destroy(&sharedDataMap);
    pthread_barrier_destroy(&waitMappersFinish);
    pthread_barrier_destroy(&waitForReducers);
    pthread_barrier_destroy(&waitForMap);
    pthread_mutex_destroy(&pickNextFile);
    for (int i = 0; i < 26; i++) {
        files[i].close();
    }

    return 0;
}

void initMutexes(pthread_mutex_t *m1, pthread_mutex_t *m2, pthread_mutex_t *m3) {
    pthread_mutex_init(m1, NULL);
    pthread_mutex_init(m2, NULL);
    pthread_mutex_init(m3, NULL);
}

void initBarriers(pthread_barrier_t *m1, pthread_barrier_t *m2, pthread_barrier_t *m3, int map, int red) {
    pthread_barrier_init(m1, NULL, map + red);
    pthread_barrier_init(m2, NULL, map);
    pthread_barrier_init(m3, NULL, red);
}

void initFileNames(vector<pair<string,int>> *fileNames, int *noFiles, ifstream *fin, vector <ofstream> *files) {
    *fin >> *noFiles;
    for (int i = 0; i < *noFiles; i++) {
        string myString;
        *fin >> myString;
        (*fileNames).push_back(make_pair(myString, i + 1));

    }
    (*fin).close();
    string fileName("a.txt");
    char myName = 'a';
    for (int i = 0; i < 26; i++) {
        // can't use push_back for an ofstream vector
        // as the ofstream type CANNOT be copied
        (*files).emplace_back(fileName);
        myName++;
        fileName[0] = myName;
    }
}