zadatak

// INDEKS IME PREZIME
/**
 * Napisati konkurentni C++ program koji iz ulazne datoteke "ulaz.txt" cita red po red, obradjuje ih po navedenim komandama i potom rezultat smesta u izlazne datoteke "ponovo3.txt" i "ostali.txt".
 *
 * U svakoj ulaznoj datoteci su podaci podeljeni delimiterom ',', i prate
 * sledeci format:
 *     1. broj: ID, unsigned int
 *     2. broj: komanda, unsigned char, odreduje sta treba raditi sa brojevima
 *                koji slede, pritom obratiti paznju da je komanda takodje BROJ
 *     3+. brojevi: brojevi, double, koje treba obraditi po definiciji komande
 *
 * Ulazna datoteka je "ulaz.txt".
 *
 *
 * Komande koje postoje su:
 *     1) Pocevsi od nule svaki prvi razlomljeni broj dodati rezultatu a svaki drugi oduzeti
 *     2) Naci minimum razlomljenih brojeva
 *     3) Podrazumevati da su navedeni razlomljeni brojevi redom brojevi a,b,c,d; izracunati e gde je e = (a+2b+3b+4c) / 10
 *     4) odbaciti brojeve deljive sa 7
 *     5) sortirati uzlazno razlomljene brojeve
 *
 * U svakoj izlaznoj datoteci su podaci nakon transformacije upisani u jedan
 * red, podeljeni delimiterom ',', i prate sledeci format:
 *     1. broj: ID, neoznacen ceo broj
 *     2+. brojevi: razlomljeni broj/evi dobijen/i transformacijom na osnovu
 *                    komande
 *
 * Izlazni podaci se rasporedjuju u izlazne datoteke i to:
 *     "ponovo3.txt" - rezultaati za unose sa neparnim dvocifrenim ID-om
 *     "ostali.txt" - za sve ostale rezultate
 *
 * Treba napraviti jednu citac nit koja ce samo citati redove iz ulaznih
 * datoteka, jednu stampac nit koja ce samo pisati izlazne podatke u izlazne
 * datoteke i 9 niti radnika koji ce transformisati ulazne podatke u izlazne.
 *
 * NAPOMENE:
 * Komanda 0 nikada nece postojati u ulaznim datotekama.
 * Redove sa nepostojecim komandama napisati u datoteku "kanta.txt" tako da budu
 * ispisani id, komanda i potom polazni brojevi podeljeni delimiterom ','.
*/
#include <iostream>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <vector>
#include <queue>
#include <fstream>
#include <sstream>
#include <cmath>
#include <sstream>


#define BROJ_RADNIKA 9
using namespace std;

typedef struct izlazni_podaci {
    // TODO
    unsigned int id;
    unsigned char komanda;
    vector<double> NizBrojeva;
    int KojiFajl;  // 0-Kanta 1- ponovo 2-ostali
}iPodaci;

/** Klasa koja modeluje "postansko sanduce" izmedju citaca i radnika.
*/
template<typename T>
class UlazniPodaci {
private:
    // TODO dodati polja ako je potrebno
    mutex podaci_mx;
    queue<T> redovi;
    condition_variable podaci_spremni;
    bool kraj;
public:
    UlazniPodaci(): kraj(false){}

    void dodaj(T neobradjeni_podaci) {
        // TODO
    unique_lock<mutex> propusnica(podaci_mx);
        redovi.push(neobradjeni_podaci);
    }

    bool preuzmi(T &neobradjeni_podaci) {
        // TODO
    unique_lock<mutex> propusnica(podaci_mx);
        while (daLiCekamPodatke()) {
            podaci_spremni.wait(propusnica);
        }

        if (jeLiKraj()) {
            return false;
        }

        neobradjeni_podaci = redovi.front();
        redovi.pop();
        return true;
    }
       void objaviKraj() {
        unique_lock<mutex> propusnica(podaci_mx);
        kraj = true;
        podaci_spremni.notify_all();
    }
private:
    bool daLiCekamPodatke() {
        return redovi.empty() && !kraj;
    }


    bool jeLiKraj() {
        return redovi.empty() && kraj;
    }
    // TODO dodati jos neku metodu ako je potrebno
};


/** Klasa koja modeluje "postansko sanduce" izmedju radnika i pisaca.
*/
template<typename T>
class IzlazniPodaci {
private:
    mutex podaci_mx;
    queue<T> IzlazniRed;
    condition_variable podaci_spremni;
    bool kraj;
    int br_stvaralaca_podataka;

public:
    IzlazniPodaci(): kraj(false), br_stvaralaca_podataka(0){}

    void dodaj(T obradjeni_podaci) {

        unique_lock<mutex> propusnica(podaci_mx);
        IzlazniRed.push(obradjeni_podaci);
    }

    bool preuzmi(T &obradjeni_podaci) {
        // TODO
        unique_lock<mutex> propusnica(podaci_mx);
        while (daLiCekamPodatke()) {
            podaci_spremni.wait(propusnica);
        }

        if (jeLiKraj()) {
            return false;
        }

        obradjeni_podaci= IzlazniRed.front();
        IzlazniRed.pop();
        return true;
    }
 void prijavaStvaraocaPodataka() {
        unique_lock<mutex> propusnica(podaci_mx);
        br_stvaralaca_podataka++;
    }


    void odjavaStvaraocaPodataka() {
        unique_lock<mutex> propusnica(podaci_mx);
        br_stvaralaca_podataka--;
        if (!br_stvaralaca_podataka) {
            kraj = true;
            podaci_spremni.notify_one();
        }
    }

private:

    bool daLiCekamPodatke() {
        return IzlazniRed.empty() && !kraj;
    }


    bool jeLiKraj() {
        return IzlazniRed.empty() && kraj;
    }
    // TODO dodati jos neku metodu ako je potrebno
};


/**
 * Parsiranje reda iz ulazne datoteke
 *   parametri:
 *     - red_datoteke, sadrzaj jednog reda ulazne datoteke
 *     - id, adresa gde ce biti upisan id procitan iz reda
 *     - komanda, adresa gde ce biti upisan id komande procitan iz reda
 *   vraca: kolekciju brojeva koje radnik treba da transformise po komandi
*/

vector<double> izdvoj_brojeve(const string red_datoteke, unsigned &id, unsigned char &komanda) {
    istringstream red_datoteke_tok(red_datoteke);
    vector<double> brojevi;
    string broj;

    if (getline(red_datoteke_tok, broj, ','))
        id = stoul(broj);

    if (getline(red_datoteke_tok, broj, ','))
        komanda = stoul(broj);

    while (getline(red_datoteke_tok, broj, ','))
        brojevi.push_back(stod(broj));

    return brojevi;
}

/**
 * Logika radnika - niti koje vrse transformaciju ulaznih podataka u izlazne
 * podatke spremne za ispis.
*/
void radnik_nit(UlazniPodaci<string> &ulaz, IzlazniPodaci<struct izlazni_podaci> &izlaz) {
    // TODO
    izlaz.prijavaStvaraocaPodataka();  //

    string red;
    while (ulaz.preuzmi(red)) {
       iPodaci Podatak;
       Podatak.NizBrojeva=izdvoj_brojeve(red,Podatak.id,Podatak.komanda);
       if(Podatak.id%2==1 && Podatak.id>=10)
        Podatak.KojiFajl=1;
       else if(Podatak.id%2==0)
        Podatak.KojiFajl=2;


       switch(Podatak.komanda){

          case 1:
          {
            double rezultat=0;
            for(int i=0;i<Podatak.NizBrojeva.size();i++){
                if(i%2==0)
                    rezultat += Podatak.NizBrojeva[i];
                else
                    rezultat -= Podatak.NizBrojeva[i];

            }
            Podatak.NizBrojeva.clear();
            Podatak.NizBrojeva.push_back(rezultat);
        break;
        }
          case 2:
          {
            double min=Podatak.NizBrojeva[0];
            for(int i=1;i<Podatak.NizBrojeva.size();i++){
                if(Podatak.NizBrojeva[i] < min ){
                    min=Podatak.NizBrojeva[i];
                }
            }
        Podatak.NizBrojeva.clear();
            Podatak.NizBrojeva.push_back(min);
            break;
        }
         case 3:
            {
            double e;

        e = ( Podatak.NizBrojeva[0] + 2 * Podatak.NizBrojeva[1] + 3 * Podatak.NizBrojeva[1] + 4 * Podatak.NizBrojeva[2] ) / 10 ; //ne koristi se nigde d, verovatno greska u postavci
            Podatak.NizBrojeva.clear();
            Podatak.NizBrojeva.push_back(e);

        break;
        }
         case 4:{
        vector<double>noviNiz;
        string num;
        for(int i=0;i<Podatak.NizBrojeva.size();i++){
            if( (int)Podatak.NizBrojeva[i] % 7 == 0)
                noviNiz.push_back(Podatak.NizBrojeva[i]);

        }
        Podatak.NizBrojeva.clear();
        for(int i=0;i<noviNiz.size();i++)
            Podatak.NizBrojeva.push_back(noviNiz[i]);

        break;
        }
         case 5:
         {
            double temp;
        //bubble sort
        for(int i=0;i<Podatak.NizBrojeva.size()-1;i++){
            for(int j=i+1; j<Podatak.NizBrojeva.size();i++)
                if(Podatak.NizBrojeva[j]<Podatak.NizBrojeva[i]){
                    temp = Podatak.NizBrojeva[j];
                    Podatak.NizBrojeva[j]=Podatak.NizBrojeva[i];
                    Podatak.NizBrojeva[i]=temp;
                }

        }
         }
     default:
     {
         Podatak.KojiFajl=0;
         break;
         }

       }


        izlaz.dodaj(Podatak);
    }

    izlaz.odjavaStvaraocaPodataka();


}

/**
 * Logika citaca - nit koja radi citanje iz ulazne datoteke i salje u ulaznu
 * kolekciju za radnike
*/
void citac_nit(UlazniPodaci<string> &ulaz) {
    // TODO
        string ime_ulazne_dat="ulaz.txt";
        ifstream ulazna_dat(ime_ulazne_dat);
        cout << "CITAC: Obradjujem datoteku. . . \"" << ime_ulazne_dat << "\"." << endl;
        if(ulazna_dat.is_open()) {
            string red_datoteke;
            while(getline(ulazna_dat, red_datoteke)) {  // citanje datoteke red po red
                ulaz.dodaj(red_datoteke);
            }

            ulazna_dat.close();
        } else
            cerr << "CITAC: Nisam mogao da otvorim datoteku \"" << ime_ulazne_dat << "\" za citanje!\n";


    ulaz.objaviKraj();
}

//serijalizacija
string ToString(iPodaci podatak){
  string result = to_string(podatak.id)+",";
  string str;
  if(podatak.KojiFajl==0)
    result +=to_string(podatak.komanda)+",";

    if(podatak.NizBrojeva.size()){
    str=podatak.NizBrojeva[0];
    if(str!="0\0"){
        str.erase ( str.find_last_not_of('0') + 1, std::string::npos );
    str.erase ( str.find_last_not_of('.') + 1, std::string::npos );
    }
        result += str;
        }
    for (int i = 1; i < podatak.NizBrojeva.size(); ++i) {
        str=podatak.NizBrojeva[i];
        str.erase ( str.find_last_not_of('0') + 1, std::string::npos );
    str.erase ( str.find_last_not_of('.') + 1, std::string::npos );
        result += "," + str;
    }
    result+='\n';
 return result;

    }
/**
 * Logika stampaca - nit koja radi pisanje u izlaznu datoteku podataka dobijenih
 * od radnika
*/

void stampac_nit(IzlazniPodaci<struct izlazni_podaci> &izlaz) {
    // TODO
    ofstream parni_dat("parni.txt");
    ofstream neparni_dat("neparni.txt");
    ofstream kanta_dat("kanta.txt");
    if(parni_dat.is_open()&&neparni_dat.is_open()&&kanta_dat.is_open()){
    iPodaci podatak;
        while (izlaz.preuzmi(podatak)) {
           if(podatak.KojiFajl==2)
            parni_dat<<ToString(podatak);
           else if (podatak.KojiFajl==1)
            neparni_dat<<ToString(podatak);
           else
            kanta_dat<<ToString(podatak);
        }

        parni_dat.close();
        neparni_dat.close();
        kanta_dat.close();
    } else
        cerr << "Greska prilikom otvaranja datoteke \n";
}




int main() {
    // TODO
    UlazniPodaci<string> Redovi;
    IzlazniPodaci<iPodaci> PripremljeniPodaci;
    thread th_reader(citac_nit, ref(Redovi));
    thread th_writer(stampac_nit, ref(PripremljeniPodaci));
    thread th_workers[BROJ_RADNIKA];

    for(auto &th: th_workers){
        th = thread(radnik_nit, ref(Redovi), ref(PripremljeniPodaci));
    }

    th_reader.join();
    for(auto &th: th_workers) {
        th.join();
    }
    th_writer.join();
    exit(0);
}