#include <unistd.h>
#include <fcntl.h>
#include <ctype.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <wchar.h>
#include <ctype.h>
#include <wctype.h>
#include <time.h>

#include "Reduce.h"
#include "Errors.h"


/** 
 *  Compare to words using strcoll to take into account the special characteristics of the current locale premises
 */

int cmp_words(const char *a, const char *b)
{
    return strcoll(a, b);
}


/** 
 *  Reducer process suffle (sorted by key/word) tuples file a writting result on output file.
 */

int reduce_process_file(int fd_in, int fd_out) 
{
    char buffer[DBUFFER_IN_SIZE], line[DMAX_LINE], prev[128]="", word[128];
    int pos = 0, sum = 0, total_words=0;
    ssize_t n;

    /* read tuples file input */ 
    while ((n = read(fd_in, buffer, DBUFFER_IN_SIZE)) > 0) {

        for (int i = 0; i < n; i++) {

            if (buffer[i] != '\n') 
            {
                line[pos++] = buffer[i];
            } 
            else 
            {
                int count;
                
                // End line found -> process (word,count) tuple
                line[pos] = '\0';

                sscanf(line,"%s %d",word,&count);
                if (prev[0]=='\0' || cmp_words(prev,word)!=0) 
                {
                    /* if the word is different of previous one write acumulate result (word,total_count) to result file */
                    if (sum>0) {
                        char out[256];
                        int len=sprintf(out,"%s %d\n",prev,sum);
                        if (write(fd_out,out,len)<0)
                            Error("[Reduce::reduce_process_file] Error writting reduce result.\n");
                        total_words++;
                    }

                    strcpy(prev,word);
                    sum=count;

                } else
                    sum+=count;

                pos=0;
            }
        }
    }

    // Write las key/word result
    if (sum>0) {
        char out[256];
        int len=sprintf(out,"%s %d\n",prev,sum);
        if (write(fd_out,out,len)<0)
            Error("[Reduce::reduce_process_file] Error writting reduce result (final).\n");
        total_words++;
    }

    return total_words;
}



/** 
 *  Function to perfom reducer stage. 
 *  Process all suffle files using r reducers and generating the result in only one output file
 */

void wordcount_reducer(char *output_file, int n_reducers)
{
    char outname[64];
    int fd_in, fd_out, words=0, res=0;
    struct timespec start, finish;
    float time;

    if (VERBOSE) {
        sprintf(Msg,"[Reduce Stage\tStart Reduceer to process %d input files.\n", n_reducers);
        print(Msg);
    }

    if (DEBUG) {
        sprintf(MsgDebug,"[Reduce::wordcount_reducer] wordcount_reducer %s %d\n", output_file, n_reducers);
        printDebug(MsgDebug);
    } 

    clock_gettime(CLOCK_MONOTONIC, &start);

    fd_out = open(output_file, O_WRONLY | O_CREAT | O_TRUNC, 0644);
    if (fd_out<0)
    {
        sprintf(MsgError,"[Reduce::wordcount_reducer] Error opening reducer output file %s\n",output_file);
        Error(MsgError);
    }

    // Open reducer input file.
    if (n_reducers==1)
    {
        // Open one reducer input file.
        sprintf(outname, "%s%s_%d%s", DAuxiliarDiretory, DSuffleFilePrefix, 0, DAuxiliarFilesExtension);
        fd_in = open(outname, O_RDONLY);
        if (fd_in<0)
        {
            sprintf(MsgError,"[Reduce::wordcount_reducer] Error opening reducer input file %s\n",outname);
            Error(MsgError);
        }

        res = reduce_process_file(fd_in, fd_out);
        words +=res;
        if (VERBOSE) 
        {
            sprintf(Msg,"\t\tReducer processed %d words from %s file.\n", res, outname);
            printMessage(Msg, COLOR_GREEN_L);
        }

        close(fd_in);
        close(fd_out);
    }
    else if (n_reducers>1)
    {
        // Process input reduce files
        for (int i = 0; i < n_reducers; i++) 
        {   
            // Open i reducer input file.
            sprintf(outname, "%s%s_%d%s", DAuxiliarDiretory, DSuffleFilePrefix, i, DAuxiliarFilesExtension);
            fd_in = open(outname, O_RDONLY);
            if (fd_in<0)
            {
                sprintf(MsgError,"[Reduce::wordcount_reducer] Error opening reducer input file %s\n",outname);
                Error(MsgError);
            }
            res = reduce_process_file(fd_in, fd_out);     
            words +=res;
            if (VERBOSE) 
            {
                sprintf(Msg,"\t\tReducer processed %d words from %s file.\n", res, outname);
                printMessage(Msg, COLOR_GREEN_L);
            }

            close(fd_in);
        }
        close(fd_out);
    }    

    if (VERBOSE) 
    {
        clock_gettime(CLOCK_MONOTONIC, &finish);
        time =  (finish.tv_sec - start.tv_sec)+((finish.tv_nsec - start.tv_nsec) / 1000000000.0);
        sprintf(Msg,"[Reduce Stage]\tReducer Done. Processed %d diferent words with %d reducers in %.3f secs.\n", words, n_reducers, time);
        print(Msg);
    }
}


