// Czy w aplikacji wystepuja wyscigi?
// Czy mozesz poprawić skalowalność aplikacji?

#include <memory>
#include <thread>
#include <mutex>
#include <vector>
#include <iostream>
#include <random>
#include <algorithm>
#include <iterator>
#include <iostream>
#include <atomic>
#include <chrono>

using namespace std;

using Data = int;  //zakladamy, ze Data jest to duzy obiekt
const Data EMPTY = -1; //reprezentuje brak danych

bool isProper(const Data& d) {
    //bardzo dlugi algorytm lokalny
    std::this_thread::sleep_for( chrono::milliseconds(10));
    //  cout << "is proper " << d << endl;
    return d == 0; //symuluje rozne wartosci
}

struct InData {
    InData(const vector<Data>& r) : records_(r), counter_(0) {}
    const vector<Data>& records_;
    mutex m_;
    int counter_;
};

struct Thread {
    Thread(InData& d) : data_(d), currentIdx_(0) {}

    void setIdx(int i) { currentIdx_ = i; }
    void operator()() {
        std::lock_guard<std::mutex> lock( data_.m_ );
        if(isProper(data_.records_[currentIdx_]) ) {
            data_.counter_ += 1;
        }
    }
private:
    int currentIdx_;
    InData& data_;
};

int countProper(const vector<Data>& records) {
    InData input(records);

    const int NUM_THREADS = 3;
    std::vector<Thread> vec_thread(NUM_THREADS, Thread(input) );

    for(int i = 0, epoch_size = 0; i < records.size(); i += epoch_size) {
        int still_not_used = records.size() - i;
        epoch_size = std::min(NUM_THREADS, still_not_used); //w tym obiegu tyle bedzie przetwarzac

        vector<thread> thrds;
        for(int j = 0; j < epoch_size; ++j) {
            vec_thread[j].setIdx(i+j);
            thrds.push_back( thread(ref(vec_thread[j]) ) );
        }
        for(int j = 0; j < epoch_size; ++j) {
            thrds[j].join();
        }
    }
    return input.counter_;
}

struct ThreadBetter {
    ThreadBetter(atomic<int>& c) : counter_(c), data_(nullptr) {}

    void setData(const Data* d) { data_ = d; }
    void operator()() {
        if(data_ && isProper(*data_) ) {
            counter_ += 1;
        }
    }
private:
    atomic<int>& counter_;
    const Data* data_;
};

int countProperBetter(const vector<Data>& records) {
    atomic<int> counter(0);

    const int NUM_THREADS = 3;
    std::vector<ThreadBetter> vec_thread(NUM_THREADS, counter );

    for(int i = 0, epoch_size = 0; i < records.size(); i += epoch_size) {
        int still_not_used = records.size() - i;
        epoch_size = std::min(NUM_THREADS, still_not_used); //w tym obiegu tyle bedzie przetwarzac

        vector<thread> thrds;
        for(int j = 0; j < epoch_size; ++j) {
            vec_thread[j].setData(&records[i+j]);
            thrds.push_back( thread(ref(vec_thread[j]) ) );
        }
        for(int j = 0; j < epoch_size; ++j) {
            thrds[j].join();
        }
    }
    return counter.load();
}

const int N = 500;
int main () {
    vector<Data> v(N); //wektor N liczb
    std::mt19937_64 gen(0); //generator losowy (Mersanne Twister, default seed)
    std::uniform_int_distribution<int> distr(-9, 9);
    generate( v.begin(), v.end(), [&](){ return distr(gen);}); //N liczb losowych od 1*10^12 do 9*10^12
    cout << "v size:" << v.size() << endl;
    copy( v.begin(), v.end(), ostream_iterator<int>(cout, " ") );
    cout << endl;
    auto start1 = std::chrono::high_resolution_clock::now();
    cout << countProper(v) << endl;
    auto end1 = std::chrono::high_resolution_clock::now();
    std::cout << "Ver1 time "
              << std::chrono::duration_cast<std::chrono::milliseconds>(end1 - start1).count() << "ms"
              << std::endl;

    auto start2 = std::chrono::high_resolution_clock::now();
    cout << countProperBetter(v) << endl;
    auto end2 = std::chrono::high_resolution_clock::now();
    std::cout << "Ver2 time "
              << std::chrono::duration_cast<std::chrono::milliseconds>(end2 - start2).count() << "ms"
              << std::endl;

    return 0;

}