#include <pthread.h>    //pthread_create(), pthread_join(), pthread_attr_destroy(), pthread_cond_wait(), pthread_cond_signal(), pthread_mutex_lock(), pthread_mutex_unlock()
#include <stdio.h>      //printf(), perror(), fflush()
#include <stdlib.h>     //exit(), atoi()
#include <sys/types.h>  //socket(), bind(), accept(), recv(), connect(), fork(), open()
#include <sys/socket.h> //socket(), bind(), accept(), recv(), connect()
#include <netinet/in.h> //sockaddr_in, htons()
#include <netdb.h>      //gethostbyname() 
#include <string.h>     //strcpy(), memcpy()
#include <unistd.h>     //gethostname(), close(), write(), sleep(), fork()
#include <sys/stat.h>   //open()
#include <fcntl.h>      //open()

#define MTU             1500
#define min(a,b)        (((a) < (b)) ? (a) : (b))

//argv[0]: throttle_tcp
//argv[1]: puerto local
//argv[2]: [host remoto]
//argv[3]: puerto remoto

int bucketLevelUp = 0;                                          // Variable global con el espacio usado del balde de subida.
int bucketLevelDown = 0;                                        // Variable global con el espacio usado del balde de bajada.
int bucketSize = 15000;                                         // Variable global para controlar el tamano maximo de rafaga.
int leakingRate = 10000;                                        // Variable global para controlar la tasa de vaciado del balde.
int subida=0;                                                   // Variable global para almacenar la tasa de subida (graphics).
int bajada=0;                                                   // Variable global para almacenar la tasa de bajada (graphics).
pthread_mutex_t subidaLock = PTHREAD_MUTEX_INITIALIZER;         // Candado de exclusion mutua para la tasa de subida.
pthread_mutex_t bajadaLock = PTHREAD_MUTEX_INITIALIZER;         // Candado de exclusion mutua para la tasa de bajada.
pthread_mutex_t bucketLockUp = PTHREAD_MUTEX_INITIALIZER;       // Candado de exclusion mutua para el balde de subida.
pthread_mutex_t bucketLockDown = PTHREAD_MUTEX_INITIALIZER;     // Candado de exclusion mutua para el balde de bajada.
pthread_cond_t bucketChangeUp = PTHREAD_COND_INITIALIZER;       // Variable de condicion de cambio para el balde de subida.
pthread_cond_t bucketChangeDown = PTHREAD_COND_INITIALIZER;     // Variable de condicion de cambio para el balde de bajada.
pthread_rwlock_t bucketSizeLock = PTHREAD_RWLOCK_INITIALIZER;   // Candado de exclusion mutua para modificacion de tamano del balde.
pthread_rwlock_t leakingRateLock = PTHREAD_RWLOCK_INITIALIZER;  // Candado de exclusion mutua para modificacion de tasa de vaciado.

void * leak(void * arg);
void * graphics(void * arg);
void graphicsCleanup(void * arg);

// Funcion que controla la direccion de subida.
void *upload(void *args) {
    char buf[MTU];                                                          // Buffer de transferencia.
    int n;                                                                  // Cantidad total de elementos a transferir.
    int i;                                                                  // Posicion del buffer donde comienzan los elementos que faltan por transmitir.
    int j;                                                                  // Cantidad parcial de elementos a transferir en un ciclo.
    while ((n = recv(*((int *)(args)), buf, sizeof(buf), 0)) > 0)           // Se bloquea hasta recepcion de bytes desde el cliente.
    {
        for(i=0;i<n;i+=j)
        {
            pthread_mutex_lock(&bucketLockUp);                              // Toma el candado de acceso al balde de subida.
            pthread_rwlock_rdlock(&bucketSizeLock);                         // Toma el candado para garantizar lectura valida del tamano del balde.
            while (bucketLevelUp >= bucketSize)                             // Mientra el balde esta lleno.
            {
                pthread_rwlock_unlock(&bucketSizeLock);                     // Libera candado de lectura de tamano de balde.
                pthread_cond_wait(&bucketChangeUp, &bucketLockUp);          // Se duerme en la variable de condicion del balde de subida. (Wait tipo Hoare)
                pthread_rwlock_rdlock(&bucketSizeLock);                     // Toma el candado para garantizar lectura valida del tamano del balde en la revision del while.
            }
            j = min(n-i,bucketSize-bucketLevelUp);                          // Cantidad permitida de elementos a transferir. (n-i): elementos restantes por enviar, (bucketSize-bucketLevelUp): maxima rafaga permitida
            pthread_rwlock_unlock(&bucketSizeLock);                         // Libera candado de lectura de tamano de balde.
            pthread_mutex_lock(&subidaLock);                                // Toma el candado de acceso a la tasa de subida para el grafico.
            subida+=j;                                                      // Suma la cantidad de bytes que transmitira al servidor.
            pthread_mutex_unlock(&subidaLock);                              // Libera el candado de acceso a la tasa de subida para el grafico.
            bucketLevelUp+=j;                                               // Aumenta el uso del balde de subida en el mayor posible.
            pthread_mutex_unlock(&bucketLockUp);                            // Libera el candado de acceso al balde de subida.
            if (send(*((int *)(args) + 1), buf+i, j, 0) < 0)                // Envia j bytes recibidos del cliente al servidor.
                {perror("send failed"); exit(1);}
        }
    }
}

// Funcion que controla la direccion de bajada.
void *download(void *args) {
    char buf[MTU];                                                          // Buffer de transferencia.
    int n;                                                                  // Cantidad de elementos a transferir.
    int i;                                                                  // Posicion del buffer donde comienzan los elementos que faltan por transmitir.
    int j;                                                                  // Cantidad parcial de elementos a transferir en un ciclo.
    while ((n = recv(*((int *)(args) + 1), buf, sizeof(buf), 0)) > 0)       // Se bloquea hasta recepcion de bytes desde el servidor.
    {
        for(i=0;i<n;i+=j)
        {
            pthread_mutex_lock(&bucketLockDown);                            // Toma el candado de acceso al balde de subida.
            pthread_rwlock_rdlock(&bucketSizeLock);                         // Toma el candado para garantizar lectura valida del tamano del balde.
            while (bucketLevelDown >= bucketSize)                           // Mientra el balde esta lleno.
            {
                pthread_rwlock_unlock(&bucketSizeLock);                     // Libera candado de lectura de tamano de balde.
                pthread_cond_wait(&bucketChangeDown, &bucketLockDown);      // Se duerme en la variable de condicion del balde de subida. (Wait tipo Hoare)      
                pthread_rwlock_rdlock(&bucketSizeLock);                     // Toma el candado para garantizar lectura valida del tamano del balde en la revision del while.
            }
            j = min(n-i,bucketSize-bucketLevelDown);                        // Cantidad permitida de elementos a transferir. (n-i): elementos restantes por enviar, (bucketSize-bucketLevelUp): maxima rafaga permitida
            pthread_rwlock_unlock(&bucketSizeLock);                         // Libera candado de lectura de tamano de balde.
            pthread_mutex_lock(&bajadaLock);                                // Toma el candado de acceso a la tasa de bajada para el grafico.
            bajada+=j;                                                      // Suma la cantidad de bytes que transmitira al cliente.
            pthread_mutex_unlock(&bajadaLock);                              // Libera el candado de acceso a la tasa de bajada para el grafico.
            bucketLevelDown+=j;                                             // Aumenta el uso del balde de subida en el mayor posible.
            pthread_mutex_unlock(&bucketLockDown);                          // Libera el candado de acceso al balde de subida.
            if (send(*((int *)(args)), buf+i, j, 0) < 0)                    // Envia los j bytes recibidos del servidor al cliente.
                {perror("send failed"); exit(1);}
        }
    }
}                      

int main(int argc, char *argv[]) {
    int s_srv;                      // Descriptor referenciando al socket de la conexion con el programa cliente.
    int s_cli;                      // Descriptor referenciando al socket de la conexion con el programa servidor.
    int ns;                         // Descriptor de socket servidor abierto.
    int len_srv;                    // Tamano de la estructura sockaddr_in para el extremo servidor.
    int len_cli;                    // Tamano de la estructura sockaddr_in para el extremo cliente.
    char hostname[64];              // Nombre del host remoto.
    pthread_t up;                   // Hebra para control de upload.
    pthread_t down;                 // Hebra para control de download.
    pthread_t leakyB;               // Hebra para control de vaciamiento de los baldes.
    pthread_t graph;                // Hebra para desplegar los graficos (Parte Opcional).
    struct sockaddr_in name_srv;    // Caracteristicas del socket servidor.
    struct sockaddr_in name_cli;    // Caracteristicas del socket cliente.
    struct hostent *hp;             // Puntero a la estructura con la informacion sobre la maquina remota.
    int args[2];                    // Arreglo para pasar mas de un argumento con pthread_create.

    switch(argc) {
        case 3:
            if(gethostname(hostname, sizeof(hostname)) < 0)                     // Si se omite el host_remoto se entiende que es la misma maquina (localhost).
                {perror("gethostname failed"); exit(-1);}
            break;
        case 4:
            strcpy(hostname, argv[2]);                                          // Extrae nombre de host del argumento.
            break;
        default:
            printf("Uso: %s puerto [host_remoto] puerto_remoto \n",argv[0]);
            exit(-1);
    }	// En hostname queda el nombre de la maquina destino.

//////////////////////////////////////////////////////////
//  Establecimiento de conexion cliente <-> servidor    //
//////////////////////////////////////////////////////////
    if ((s_srv = socket(AF_INET, SOCK_STREAM, 0)) < 0)                          // Crea socket IPv4, Conexion TCP.
        {perror("socket failed"); exit(-1);}
    name_srv.sin_family = AF_INET;                                              // Address Family Internet.
    name_srv.sin_port = htons(atoi(argv[1]));                                   // Asigna la puerta de servicio de throttle_tcp, la cual es pasada en el primer argumento.
    name_srv.sin_addr.s_addr = htonl(INADDR_ANY);                               // Se atienden requerimientos entrantes por cualquier interfaz.
    len_srv = sizeof(struct sockaddr_in);                                       // Tamano de la estructura sockaddr_in.
    if(bind(s_srv, (struct sockaddr *) &name_srv, len_srv))                     // Asocia el socket con su nombre (caracteristicas).
        {perror("bind failed"); exit(-1);}
    listen(s_srv, 5);                                                           // Escucha por conexiones en el socket s_srv.
    if ((ns = accept(s_srv, (struct sockaddr *) &name_srv, &len_srv)) < 0)      // Acepta conexiones en el socket s_srv de caracteristicas name_srv.
        {perror("accept failed"); exit(-1);}
    if((hp = gethostbyname(hostname)) == NULL)                                  // Obtiene informacion de la maquina remota.
        {perror("gethostbyname failed"); exit(-1);}
    if ((s_cli = socket(AF_INET, SOCK_STREAM, 0)) < 0)                          // Crea socket IPv4, Conexion TCP.
        {perror("socket failed"); exit(-1);}
    name_cli.sin_family = AF_INET;                                              // Address Family Internet.
    name_cli.sin_port = htons(atoi(argv[argc-1]));                              // Asigna la puerta de servicio del host remoto, la cual es pasada en el tercer argumento.
    memcpy(&name_cli.sin_addr, hp->h_addr_list[0], hp->h_length);               // Guarda la direccion del host remoto en name_cli.
    len_cli = sizeof(struct sockaddr_in);                                       // Tamano de la estructura sockaddr_in.
    if((connect(s_cli, (struct sockaddr *) &name_cli, len_cli)) < 0 )           // Se conecta con el host remoto.
        {perror("connect failed"); exit(-1);}

//////////////////////////////////////////////////////////
//  Manejo de las conexiones                            //
//////////////////////////////////////////////////////////
    args[0]=ns;                                                 // Guarda en el arreglo de argumentos el descriptor del socket servidor abierto (conectado con el programa cliente).
    args[1]=s_cli;                                              // Guarda en el arreglo de argumentos el descriptor del socket cliente (conectado con el programa servidor).

    if(pthread_create(&leakyB, NULL, leak, NULL))               // Crea la hebra que controlara el vaciamiento de los baldes.
        {perror("pthread_create failed"); exit(-1);}
    if(pthread_create(&up, NULL, upload,(void *) args))         // Crea la hebra que manejara el upload.
        {perror("pthread_create failed"); exit(-1);}
    if(pthread_create(&down, NULL, download,(void *) args))     // Crea la hebra que manejara el download.
        {perror("pthread_create failed"); exit(-1);}
    if(pthread_create(&graph, NULL, graphics,NULL))
        {perror("pthread_create failed"); exit(-1);}

    printf("Para terminar fije un tamano invalido (p.ej. -1).\n");
    while(1)                                                    // Prompt de cambio de los parametros de leaky bucket.
    {
        int temp=0;                                             // Almacenara temporalmente el nuevo valor de parametro.
        printf("Nuevo tamano del balde [B]: "); scanf("%d",&temp);
        if(temp<0)
        {
            printf("Tamano invalido, cerrando...\n");
            break;
        }
        else
        {
            pthread_rwlock_wrlock(&bucketSizeLock);             // Toma candado de exclusion mutua para cambiar el tamano del balde.
            bucketSize = temp;                                  // Cambia el tamano del balde.
            pthread_rwlock_unlock(&bucketSizeLock);             // Libera candado de modificacion de tamano del balde.
            pthread_cond_signal(&bucketChangeUp);               // Senala a la variable de condicion del balde de subida que hubo un cambio.
            pthread_cond_signal(&bucketChangeDown);             // Senala a la variable de condicion del balde de bajada que hubo un cambio.
        }
        printf("Nueva tasa de vaciado [Bps]: "); scanf("%d",&temp);
        if((temp<0)||(temp>bucketSize))
        {
            printf("Tamano invalido, cerrando...\n");
            break;
        }
        else
        {
            pthread_rwlock_wrlock(&leakingRateLock);            // Toma candado de exclusion mutua para cambiar tasa de vaciado.
            leakingRate = temp;                                 // Cambia la tasa de vaciado del balde.
            pthread_rwlock_unlock(&leakingRateLock);            // Libera candado de modificacion de tasa de vaciado.
        }
    }

    if(pthread_cancel(graph))                                   // Cancela la hebra que grafica.
        {perror("pthread_cancel failed"); exit(-1);}
    if(pthread_cancel(up))                                      // Cancela la hebra de subida.
        {perror("pthread_cancel failed"); exit(-1);}
    if(pthread_cancel(down))                                    // Cancela la hebra de bajada.
        {perror("pthread_cancel failed"); exit(-1);}
    if(pthread_cancel(leakyB))                                  // Cancela la hebra de actualizacion de tasa de vaciado.
        {perror("pthread_cancel failed"); exit(-1);}


    if(pthread_join(graph, NULL))                               // Espera por el retorno de las hebras canceladas.
        {perror("pthread_join failed"); exit(-1);}
    if(pthread_join(up, NULL))
        {perror("pthread_join failed"); exit(-1);}
    if(pthread_join(down, NULL))
        {perror("pthread_join failed"); exit(-1);}
    if(pthread_join(leakyB, NULL))
        {perror("pthread_join failed"); exit(-1);}

    close(ns);      // Cierra descriptor de socket servidor abierto.
    close(s_srv);   // Cierra descriptor referenciando al socket de la conexion con el cliente.
    close(s_cli);   // Cierra descriptor referenciando al socket de la conexion con el servidor.
    return 0;
}

// Funcion que controla el vaciamiento de los baldes.
void * leak(void * arg)
{
    while (1)
    {
        sleep(1);
        pthread_mutex_lock(&bucketLockUp);              // Toma el candado de acceso al balde de subida.
        if (bucketLevelUp>0)                            // Si el balde de subida no esta vacio.
        {
            pthread_rwlock_rdlock(&leakingRateLock);    // Toma el candado de lectura de tasa de vaciado.
            bucketLevelUp-=leakingRate;                 // Libera una cantidad dada por la tasa de salida.
            pthread_rwlock_unlock(&leakingRateLock);    // Libera el candado de lectura de tasa de vaciado.
        }
        pthread_mutex_unlock(&bucketLockUp);            // Libera el candado de acceso al balde de subida.
        pthread_cond_signal(&bucketChangeUp);           // Senala a la variable de condicion del balde de subida que hubo un cambio.

        pthread_mutex_lock(&bucketLockDown);            // Toma el candado de acceso al balde de bajada.
        if (bucketLevelDown>0)                          // Si el balde de bajada no esta vacio.
        {
            pthread_rwlock_rdlock(&leakingRateLock);    // Toma el candado de lectura de tasa de vaciado.
            bucketLevelDown-=leakingRate;               // Libera una cantidad dada por la tasa de salida.
            pthread_rwlock_unlock(&leakingRateLock);    // Libera el candado de lectura de tasa de vaciado.
        }
        pthread_mutex_unlock(&bucketLockDown);          // Libera el candado de acceso al balde de bajada.
        pthread_cond_signal(&bucketChangeDown);	        // Senala a la variable de condicion del balde de bajada que hubo un cambio.
    }
}

// Funcion que grafica las tasas en Matlab
void * graphics(void *arg)
{
    pid_t pid;                                          // Identificador de proceso para el hijo.
    int pfd[2];                                         // Descriptor de archivo para la pipa.
    FILE *sd;                                           // Descriptor de archivo para formatear la salida a la pipa.
    int args[3];
    
    if (pipe(pfd) < 0)                                  // Crea la pipa
        {perror("pipe");pthread_exit((void *)1);}
	
    if ((pid = fork()) < 0)
        {perror("fork");pthread_exit((void *)1);}
	
    if (pid == 0)					                    //HIJO
    {
        int junk;                                       // Variable que apunta a tierra
        dup2(pfd[0], 0);	                            // Pipa matlab: Standard Input del hijo lee de la pipa
        close(pfd[1]);		                            // Pipa matlab: Hijo cierra lado de escritura de la pipa pues no la utiliza
        junk = open("/dev/null", O_WRONLY);             // Apunta a tierra
        dup2(junk, 1);
        dup2(junk, 2);                                  // Descartar cualquier mensaje o error enviado a la pantalla
        execlp("matlab", "matlab", "-nodesktop", NULL); // Ejecuta matlab en modo consola
        perror("exec");
        pthread_exit((void *)1);
    }
    close(pfd[0]);                                      // Pipa matlab: Padre no lee de la pipa, entonces cierra lado de lectura
    sd = fdopen(pfd[1], "w");	                        // Permite escribir a la pipa con fprintf mediante sd

    args[0]=(int)sd;
    args[1]=(int)pfd[1];
    args[2]=(int)pid;
    pthread_cleanup_push(graphicsCleanup,args);         // Establece la rutina de limpieza que se invoca al hacer un pthread_cancel() a esta hebra.

    fprintf(sd, "x= [];\n");                            // Crea los 2 arreglos q contendran los valores de las tasas en matlab
    fprintf(sd, "y= [];\n");
    while(1)
    {
        pthread_mutex_lock(&subidaLock);        
        fprintf(sd, "if(length(x)<30) x=[x,%d]; else x=[x([2:end]),%d]; end\n",subida/2,subida/2);  // Llena el buffer hasta que hayan 30 elementos, luego los desplaza. subida va dividida en 2 porque plotea cada 2 segundos.
        subida=0;                                                                                   // Resetea subida
        pthread_mutex_unlock(&subidaLock);
        pthread_mutex_lock(&bajadaLock);
        fprintf(sd, "if(length(y)<30) y=[y,%d]; else y=[y([2:end]),%d]; end\n",bajada/2,bajada/2);
        bajada=0;
        pthread_mutex_unlock(&bajadaLock);   
        fprintf(sd,"subplot(2,1,1);\n");
        fprintf(sd,"plot(0:2:2*(length(y)-1),y);\n");
        fprintf(sd,"axis([ 0 60 0 %d ]);\n",bucketSize);
        fprintf(sd,"axis 'auto y';\n");
        fprintf(sd,"xlabel('Tiempo de la muestra [s]');\n");
        fprintf(sd,"ylabel('Tasa [Bps]');\n");
        fprintf(sd,"title('Tasa de bajada');\n");
        fprintf(sd,"subplot(2,1,2);\n");
        fprintf(sd,"plot(0:2:2*(length(x)-1),x);\n");
        fprintf(sd,"axis([ 0 60 0 %d ]);\n",bucketSize);
        fprintf(sd,"axis 'auto y';\n");
        fprintf(sd,"xlabel('Tiempo de la muestra [s]');\n");
        fprintf(sd,"ylabel('Tasa [Bps]');\n");
        fprintf(sd,"title('Tasa de subida');\n");
	    fflush(sd);
        sleep(2);           // Es un punto de cancelacion (Cancelation point) Equivalente a pthread_testcancel(),
    }                       // se invoca el handler de cancelacion si se le ejecuta un pthread_cancel() a esta hebra.

    pthread_cleanup_pop(0);
}

// Rutina de limpieza de graphics
void graphicsCleanup(void *args)
{
    FILE *sd=(FILE *)(*((int *)(args)));    // Descriptor de la pipa hacia matlab.
    fprintf(sd,"exit\n");fflush(sd);        // Cierra matlab.
    close(*((int *)(args)+1));              // Cierra el extremo de escritura de la pipa de matlab.
    waitpid(*((int *)(args)+2), NULL, 0);   // Espera por el termino del proceso hijo (matlab).
}
