/**
 * @file parallel_accumulate.cpp
 * @brief лۼ㷨
*/

#include <iostream> 
#include <thread>
#include <vector>
#include <numeric>
#include <algorithm>
#include <functional>
#include <future>
#include <execution>

using namespace std;

class join_threads
{
    std::vector<std::thread>& threads;
public:
    explicit join_threads(std::vector<std::thread>& threads_) :
        threads(threads_)
    {}
    ~join_threads()
    {
        for (unsigned long i = 0; i < threads.size(); ++i)
        {
            if (threads[i].joinable())
                threads[i].join();
        }
    }
};

template<typename Iterator, typename T>
struct accumulate_block
{
    void operator()(Iterator first, Iterator last, T& result)
    {
        result = accumulate(first, last, result);
    }
};

template<typename Iterator, typename T>
struct accumulate_block_future
{
    T operator()(Iterator first, Iterator last)
    {
        return accumulate(first, last, T());
    }
};

template<typename Iterator, typename T>
T parallel_accumulate(Iterator first, Iterator last, T init)
{
    auto length = distance(first, last);
    if (!length) return init;

    unsigned concurrent_count = thread::hardware_concurrency();
    auto num_threads = concurrent_count != 0 ? concurrent_count : 2;
    auto block_size = length / num_threads;

    vector<T> results(num_threads);
    vector<thread> threads(num_threads - 1);

    Iterator block_start = first;
    for (int i = 0; i < (num_threads - 1); ++i)
    {
        Iterator block_end = block_start;
        advance(block_end, block_size);
        threads[i] = std::thread( 
            accumulate_block<Iterator, T>(),
            block_start, block_end, std::ref(results[i]));
        block_start = block_end;
    }
    accumulate_block<Iterator, T>()(block_start, last, results[num_threads - 1]);
    for_each(threads.begin(), threads.end(), mem_fn(&thread::join));
    return accumulate(results.begin(), results.end(), init);
}

template<typename Iterator, typename T>
T parallel_accumulate_future(Iterator first, Iterator last, T init)
{
    auto length = distance(first, last);
    if (!length) return init;

    unsigned concurrent_count = thread::hardware_concurrency();
    auto num_threads = concurrent_count != 0 ? concurrent_count : 2;
    auto block_size = length / num_threads;

    vector<future<T>> futures(num_threads - 1);
    vector<thread> threads(num_threads - 1);
    join_threads joiner(threads);

    Iterator block_start = first;  
    for (int i = 0; i < (num_threads - 1); ++i)
    {
        Iterator block_end = block_start;
        advance(block_end, block_size);
        packaged_task<T(Iterator, Iterator)> task{ accumulate_block<Iterator, T>() };
        futures[i] = task.get_future();
        threads[i] = std::thread(move(task), block_start, block_end);
        block_start = block_end;
    }
    T last_result = accumulate_block<Iterator, T>()(block_start, last);
    for_each(threads.begin(), threads.end(), mem_fn(&thread::join));
    T result = init; 
    for (unsigned long i = 0; i < (num_threads - 1); ++i)
        result += futures[i].get();
    result += last_result; 
    return result;
}

template<typename Iterator, typename T>
T parallel_accumulate_async(Iterator first, Iterator last, T init)
{
    auto length = distance(first, last);
    if (!length) return init;

    unsigned long const max_chunk_size = 25;
    if (length <= max_chunk_size)
    {
        return accumulate(first, last, init); 
    }
    else
    {
        Iterator mid_point = first;
        advance(mid_point, length / 2); 
        future<T> first_half_result =
                async(parallel_accumulate<Iterator, T>, 
                first, mid_point, init);
        T second_half_result = parallel_accumulate(mid_point, last, T()); 
        return first_half_result.get() + second_half_result;
    }
}

int main()
{
    vector<int> vi(10e7, 10);

    auto start_time = chrono::steady_clock::now();

    double sum = accumulate(vi.begin(), vi.end(), 0);

    //double sum = parallel_accumulate(vi.begin(), vi.end(), 0);  

    //double sum = reduce(execution::par, vi.begin(), vi.end(), 0);

    auto end_time = chrono::steady_clock::now();
    auto ms = chrono::duration_cast<chrono::milliseconds>(end_time - start_time).count();
    cout << ms << " ms consumed, Result: " << sum << endl;

	return 0;
}