/**************************************************************
  File:          div_conquer.cc
  Description:   A program that implements a divide and 
                 conquer function computing the sum of an array. 
                                                         
  Author:        Dana Vrajitoru                          
  Organization:  IUSB, Computer and Information Sciences
  Date:          September 2020
***************************************************************/

#include <cstdlib>
#include <pthread.h>
#include <sys/time.h>
#include <iostream>
using namespace std;
#include "div_conquer.h"

// global variables 
int size, limit=10, nr_thr; 
int counter = 0, global_id = 0, threads_done = 0, global_sum = 0;
int *a = NULL, *result = NULL;
pthread_mutex_t id_mutex, barrier_mutex, cout_mutex, data_mutex;
pthread_cond_t allin_cond;

// Creates the threads calling the Sum_thread function.
void Create_threads(pthread_t *threads, int nr_threads)
{
    for (int i=0; i<nr_threads; i++)
        if (pthread_create(&threads[i], NULL, Sum_thread, NULL) != 0)
            cout << "Pthread_create failed" << endl;
}

// Joins all of the threads so that the program doesn't end before
// they have finished the work.
void Synchronise(pthread_t *threads, int nr_threads)
{
    for (int i=0; i<nr_threads; i++)
        if (pthread_join(threads[i], NULL) != 0)
            cout << "Pthread_join failed" << endl;
}

// Inputs the number of ships and initializes the global variables.
void Init_data(int &nr_threads)
{
    pthread_mutex_init(&id_mutex, NULL);
    pthread_mutex_init(&barrier_mutex, NULL);
    pthread_mutex_init(&cout_mutex, NULL);
    pthread_mutex_init(&data_mutex, NULL);
    pthread_cond_init(&allin_cond, NULL);
    cout << "Enter the size of the array" << endl;
    cin >> size;
    a = new int[size];
    for (int i=0; i<size; i++) {
        a[i] = rand() % limit;
        cout << a[i] << ' ';
    }
    cout << endl;
    cout << "Enter the number of threads" << endl;
    cin >> nr_threads;
    nr_thr = nr_threads;
    result = new int[nr_threads];
}

// A barrier function based on the number of threads implemented using
// a conditional variable. It uses a global counter intialized as
// 0. Each thread coming in increments the counter. If the count is
// nrt equal to the number of threads, it goes into a wait on the
// conditional variable. The thread that finds the counter equal to
// the number of threads broadcasts the condition that releases all of
// them.
void Barrier(int nr_threads) 
{
    static int count = 0;
    pthread_mutex_lock(&barrier_mutex);
    count++;
    if (count == nr_threads) {
        pthread_cond_broadcast(&allin_cond); // last one in lets everyone go
        count = 0;
    }
    else
        pthread_cond_wait(&allin_cond, &barrier_mutex); 
    pthread_mutex_unlock(&barrier_mutex);
}

// The thread function: gets an id number, commputes the partial sum on
// its part of the array, then communicates the results globally using
// one of the report functions.
void * Sum_thread(void *arg) 
{
    int id, i, n, step, first, last, sum=0;
    Get_id(id);
    Global_to_local(n);
    step = n/nr_thr; 
    first = id*step;
    if (id <nr_thr-1)
        last = (id+1)*step-1;
    else
        last = size-1;
    for (i=first; i<=last; i++)
        sum += a[i];
    Report_tree(sum, id);
}

// Copies the value of the global variable size into the local
// variable n.
void Global_to_local(int &n) 
{
    pthread_mutex_lock(&data_mutex);
    n = size;
    pthread_mutex_unlock(&data_mutex);
}

// Copies the value of a global id into the reference parameter and
// increments the global id.
void Get_id(int &id)
{
    pthread_mutex_lock(&id_mutex);
    id = global_id;
    global_id++;
    pthread_mutex_unlock(&id_mutex);
}

// Waits for a given delay specified in miliseconds.
void Wait(int delay)
{
    struct timeval before, after;
    int timing = 0;
    gettimeofday(&before, 0);
    while (timing < delay) {
        gettimeofday(&after, 0);
        // Add the difference in seconds * 1000 and difference 
        // in microseconds divided by 1000
        timing = (after.tv_sec - before.tv_sec)*1000 +
            (after.tv_usec - before.tv_usec)/1000;
    }
}

// Communicating and combining the results of the partial computations
// in an asynchronrus linear way.
void Report_linear(int sum, int id) 
{  
    pthread_mutex_lock(&data_mutex);
    global_sum += sum;
    threads_done++;
    if (threads_done == nr_thr)
        cout << "The sum of the array elements is " << global_sum << endl;
    pthread_mutex_unlock(&data_mutex);
}

// Communicating and combining the results of the partial computations
// in an order based on a binary tree with root = 0 and where at each
// level, the thread id accumulates the results from itself and id+ a
// power of 2 starting from 1 and doubling in each iteration.
void Report_tree(int sum, int id) 
{
    result[id] = sum;
    int step = 1;
    while (step <= nr_thr) {
        Barrier(nr_thr);
        // nodes in the tree for this particular level
        // getting the result from the sibling
        if (id % (2*step) == 0 && id+step < nr_thr) { 
            result[id] += result[id+step]; 
        }
        step *= 2;// the next level has half the nodes
    }
    if (id == 0) // final result will be at index 0
        cout << "The sum of the array elements is " << result[0] << endl;
}