#include <unistd.h>
#include <fcntl.h>
#include <stdlib.h>
#include <string.h>
#include <stdio.h>

#include "Suffle.h"
#include "Errors.h"


// Global variables

/* In-memory buffer used to accumulate lines before sorting. Lines read from the map files are temporarily stored here.
   When the buffer reaches its maximum capacity (DMAX_LINES), the lines are sorted and written to a temporary batch file. */

char *buffer[DMAX_REDUCERS][DMAX_LINES];    /* Array of pointers to dynamically allocated lines */
int nlines[DMAX_REDUCERS];                  /* Number of lines currently stored in the buffer */
int batch_count[DMAX_REDUCERS];             /* Counter used to assign unique names to temporary batch files generated during the external sorting phase */
int num_reducers;
static char line_pool[DMAX_REDUCERS][DMAX_LINES][DMAX_LINE];           
static char outbuf[DBUFFER_OUT_SIZE];
static int outpos = 0;


/** 
 *  Comparator used by qsort to sort the lines stored in memory using utf
 */

int cmp_lines(const void *a, const void *b)
{
    const char *sa = *(const char **)a;
    const char *sb = *(const char **)b;
    return strcoll(sa, sb);
}


/** 
 *  Function to write a line to the output buffer
 */
void write_line(int fd, const char *line)
{
    int len = strlen(line);

    if (outpos + len + 1 >= DBUFFER_OUT_SIZE)
        flush_batch_out(fd);

    memcpy(outbuf + outpos, line, len);
    outpos += len;
    outbuf[outpos++] = '\n';
}


/** 
 *  Function to write buffer to file
 */

void flush_batch_out(int fd)
{
    if (outpos > 0)
    {
        if (write(fd, outbuf, outpos)<0)
            Error("[Suffle:write_line] Error writting suffle result (1).\n");
        outpos = 0;
    }
}



/** 
 *  Function to read a complete line from a map file.
 */

int read_line_map(MapFile *mf, char *line)
{
    int i = 0;

    while (1)
    {
        if (mf->pos >= mf->end)
        {
            int r = read(mf->fd, mf->buf, DBUFFER_IN_SIZE);

            if (r <= 0)
                return 0;

            mf->pos = 0;
            mf->end = r;
        }

        char c = mf->buf[mf->pos++];

        if (c == '\n')
        {
            line[i] = '\0';
            return 1;
        }

        if (i < DMAX_LINE - 1)
            line[i++] = c;
    }
}



/** 
 *  Function convert the lines accumulated in memory into an ordered batch andf write it on disk.
 */

void flush_batch(int r)
{
    if (nlines[r] == 0)
        return;

    qsort(buffer[r], nlines[r], sizeof(char *), cmp_lines);

    char name[64];
    sprintf(name, "%s%s_r%d_%d.tmp", DAuxiliarDiretory, DSuffleBatchFilePrefix, r, batch_count[r]++);

    int fd = open(name, O_WRONLY | O_CREAT | O_TRUNC, 0644);

    for (int i = 0; i < nlines[r]; i++)
        write_line(fd, buffer[r][i]);
    flush_batch_out(fd);

    close(fd);

    nlines[r] = 0;
}


/** 
 *  Function to add a read line to the buffer in memory.
 */

void add_line(const char *line)
{
    char word[DMAX_LINE];

    extract_word(line, word);

    int r = hash_word(word) % num_reducers;

    //buffer[r][nlines[r]++] = strdup(line);
    strcpy(line_pool[r][nlines[r]], line);
    buffer[r][nlines[r]] = line_pool[r][nlines[r]];
    nlines[r]++;

    if (nlines[r] >= DMAX_LINES)
        flush_batch(r);
}


/** 
 *  Function to hash function for partitioning (suffle with multiple reducers)
 */

unsigned int hash_word(const char *str)
{
    unsigned int h = 5381;

    while (*str)
        h = ((h << 5) + h) + *str++;

    return h;
}



/** 
 *  Function to extract word from "word 1" tupel string line 
 */

void extract_word(const char *line, char *word)
{
    int i = 0;
    while (line[i] && line[i] != ' ') {
        word[i] = line[i];
        i++;
    }
    word[i] = '\0';
}


/** 
 *  Function to read a line from a temporal batch file.
 */

int read_line_batch(BatchFile *bf)
{
    int i = 0;

    while (1)
    {
        if (bf->pos >= bf->end)
        {
            int r = read(bf->fd, bf->buf, DBUFFER_IN_SIZE);
            if (r<0)
                Error("[Suffle:read_line_batch] Error reading batch file.\n");
            if (r <= 0) return 0;

            bf->pos = 0;
            bf->end = r;
        }

        char c = bf->buf[bf->pos++];

        if (c == '\n')
        {
            bf->line[i] = '\0';
            return 1;
        }

        if (i < DMAX_LINE - 1)
            bf->line[i++] = c;
    }
}


/* ===============================
   HEAP PARA MERGE
================================= */


int heap_cmp(BatchFile *b, int a, int c)
{
    return strcoll(b[a].line, b[c].line);
}

void heap_swap(HeapNode *h, int i, int j)
{
    HeapNode tmp = h[i];
    h[i] = h[j];
    h[j] = tmp;
}

void heapify_down(HeapNode *h, int size, int i, BatchFile *b)
{
    int smallest = i;
    int l = 2*i + 1, r = 2*i + 2;

    if (l < size && heap_cmp(b, h[l].idx, h[smallest].idx) < 0)
        smallest = l;
    if (r < size && heap_cmp(b, h[r].idx, h[smallest].idx) < 0)
        smallest = r;

    if (smallest != i)
    {
        heap_swap(h, i, smallest);
        heapify_down(h, size, smallest, b);
    }
}


/** 
 *  Function read and merge all sorted batch files into a single final sorted file.
 */

void merge_batchs(int r)
{
    BatchFile *batchs;
    HeapNode *heap;
    int heap_size = 0;

    if (DEBUG) {
        sprintf(MsgDebug,"[Suffle:merge_batchs] merge_batchs %d\n", r);
        printDebug(MsgDebug);
    }

    batchs = malloc(sizeof(BatchFile) * DMAX_BATCHS);
    heap = malloc(sizeof(HeapNode) * DMAX_BATCHS);
    if (!batchs || !heap)
        Error("[Suffle:merge_batchs] Out of memory in merge_batchs");

    for (int i = 0; i < batch_count[r]; i++)
    {
        char name[64];

        if (batch_count[r] > DMAX_BATCHS)
            Error("[Suffle:merge_batchs] Too many batch files");

        sprintf(name, "%s%s_r%d_%d.tmp", DAuxiliarDiretory, DSuffleBatchFilePrefix, r, i);

        batchs[i].fd = open(name, O_RDONLY);
        if (batchs[i].fd<0)
            Error("[Suffle:merge_batchs] Error opening input batch file.\n");

        batchs[i].pos = batchs[i].end = 0;

        if (read_line_batch(&batchs[i]))
        {
            batchs[i].active = 1;
            heap[heap_size++].idx = i;
        }
        else batchs[i].active = 0;
    }

    /* build heap */
    for (int i = heap_size/2 - 1; i >= 0; i--)
        heapify_down(heap, heap_size, i, batchs);

    char outname[64];
    sprintf(outname, "%s%s_%d%s", DAuxiliarDiretory, DSuffleFilePrefix, r, DAuxiliarFilesExtension);
    
    int fd_out = open(outname, O_WRONLY | O_CREAT | O_TRUNC, 0644);
    if (fd_out<0)
        Error("[Suffle:merge_batchs] Error opening output batch file.\n");

    while (heap_size > 0)
    {
        int min_idx = heap[0].idx;

        write_line(fd_out, batchs[min_idx].line);

        if (read_line_batch(&batchs[min_idx]))
        {
            heap[0].idx = min_idx;
        }
        else
        {
            heap[0] = heap[--heap_size];
        }

        heapify_down(heap, heap_size, 0, batchs);
    }

    flush_batch_out(fd_out);
    close(fd_out);

    for (int i = 0; i < batch_count[r]; i++)
        close(batchs[i].fd);
}


/** 
 *  Function to process and suffle map output files, followowing the next steps:
 *    1. Read map output files, flushing batch files when bufers is full
 *    2. Flushing last batch files.
 *    3. Merge batch files.
 */
 
int suffle_process_files(int num_maps, char *map_files[], int nreducers)
{
    int total_tuples=0;
    MapFile maps[DMAX_FILES];

    num_reducers = nreducers;

    for (int i = 0; i < num_reducers; i++)
    {
        nlines[i] = 0;
        batch_count[i] = 0;
    }

    for (int i = 0; i < num_maps; i++)
    {
        maps[i].fd = open(map_files[i], O_RDONLY);
        if (maps[i].fd<0) {
            sprintf(MsgError,"[Suffle:suffle_process_files] Error opening suffle input file %s.\n", map_files[i]);
            Error(MsgError);
        }
        maps[i].pos = 0;
        maps[i].end = 0;
        maps[i].active = 1;
        maps[i].mapfile = i;
        maps[i].tuples = 0;

    }

    int active = num_maps;

    char line[DMAX_LINE];

    while (active > 0)
    {
        for (int i = 0; i < num_maps; i++)
        {
            if (!maps[i].active)
                continue;

            if (read_line_map(&maps[i], line))
            {
                add_line(line);
                maps[i].tuples++;
            }
            else
            {
                maps[i].active = 0;
                active--;
                total_tuples +=  maps[i].tuples;

                if (VERBOSE) 
                {
                    sprintf(Msg,"\t\tSuflle processed %d tuples from %s file.\n",  maps[i].tuples, map_files[maps[i].mapfile]);
                    printMessage(Msg, COLOR_GREEN_L);
                }
            }
        }
    }

    /* flush remaining buffers */
    for (int r = 0; r < num_reducers; r++)
        flush_batch(r);

    /* merge batchs per reducer */
    for (int r = 0; r < num_reducers; r++)
        merge_batchs(r);

    for (int i = 0; i < num_maps; i++)
        close(maps[i].fd);

    return total_tuples;
}



/** 
 *  Function to perform the suffle operation for wordcount application.
 */

void wordcount_suffle(int nmappers, int nreducers)
{
    char **maps_files_names;
    struct timespec start, finish;
    float time;
    int tuples;


    if (VERBOSE) {
        sprintf(Msg,"[Suffle Stage]\tStart Suffle to process %d input files and generate %d output files.\n", nmappers, nreducers);
        print(Msg);
    }

    clock_gettime(CLOCK_MONOTONIC, &start);

    maps_files_names = malloc(sizeof(char*)*nmappers);
    if (nmappers==1)
    {
        maps_files_names[0] = malloc(sizeof(char) * 64);
        sprintf(maps_files_names[0], "%s", DMapOutputFileName);
    }
    else {
        for (int i = 0; i < nmappers; i++) {
            maps_files_names[i] = malloc(sizeof(char) * 64);
            sprintf(maps_files_names[i], "%s%s_%d%s", DAuxiliarDiretory, DMapFilePrefix, i, DAuxiliarFilesExtension);
        }
    }
    tuples = suffle_process_files(nmappers, (char**) maps_files_names, nreducers);

    if (VERBOSE) 
    {
        clock_gettime(CLOCK_MONOTONIC, &finish);
        time =  (finish.tv_sec - start.tv_sec)+((finish.tv_nsec - start.tv_nsec) / 1000000000.0);
        sprintf(Msg,"[Suffle Stage]\tEnd Suffle: Processed %d tupples in %.3f secs.\n", tuples, time);
        print(Msg);
    }
}
