/*
.vS
.ps +2
.vs +4
/*
NAME:       TCPCleint 
SYNOPSIS:    TCPCleint hostid portnumber 
DESCRIPTION:    The program creates a stream socket in the inet domain, 
                Connect to TCPServer, 
                Get messsages typed by a user and 
                Send them to TCPServer running on hostid
                Then it waits for a reply from  the TCPServer
                and show it back to the user, with a message
                indicating if there is an error during the round trip. 
*/
#include <strings.h>
#include <string.h>
#include <unistd.h>

#include <signal.h>
#include <pthread.h>
#include <stdlib.h>

#include <stdio.h>
#include <errno.h>
#include <sys/time.h>

#include <sys/param.h>
#include <sys/socket.h>
#include <sys/file.h>

#include <netinet/in_systm.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <netinet/ip_icmp.h>
#include <netdb.h>
#define MAXHOSTNAME 80
#define BUFSIZE 1024

#define MAX_BURST 10
#define INITIAL_LEAKY_RATE 100 //10 KB/s


char program[25];
char l_port[10];
char r_port[10];
char host_remoto[80];
unsigned long sent=0,
	      recived=0;
	      
int   sd;
int sd_T;
int ns, puerto_remoto, n, na;
char buf[BUFSIZE];
char rbuf[BUFSIZE];

int bucketLevel=0,
    bucketLevel2=0;
int bucketSize= MAX_BURST,
    bucketSize2= MAX_BURST;
int leakingRate= INITIAL_LEAKY_RATE,
    leakingRate2= INITIAL_LEAKY_RATE; /* 1 Carry return per second */

pthread_mutex_t bucketLock = PTHREAD_MUTEX_INITIALIZER,
                bucketLock2 = PTHREAD_MUTEX_INITIALIZER,
				infoLock = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t bucketChange = PTHREAD_COND_INITIALIZER,
               bucketChange2 = PTHREAD_COND_INITIALIZER; 


void reusePort(int s)
{
  int one=1;

  if ( setsockopt(s,SOL_SOCKET,SO_REUSEADDR,(char *) &one,sizeof(one)) == -1 )
  {
    printf("error in setsockopt,SO_REUSEPORT \n");
    exit(-1);
  }
}  

/****** leak Function for rcv data *******/
void * leak(void * arg)
{
  while (1) {
    usleep(100*1000); //100ms
    pthread_mutex_lock(&bucketLock);
    if (bucketLevel>0)
      bucketLevel-=leakingRate; 
    pthread_mutex_unlock(&bucketLock);
    pthread_cond_signal(&bucketChange);
  }
}

/****** END OF leak Function *******/

/****** leak Function for snd data ******/
void * leak2(void * arg)
{
  while (1) {
    usleep(100*1000); //100ms
    pthread_mutex_lock(&bucketLock2);
    if (bucketLevel2>0)
      bucketLevel2-=leakingRate2; 
    pthread_mutex_unlock(&bucketLock2);
    pthread_cond_signal(&bucketChange2);
  }
}

/****** END OF leak Function *******/




/****   Read from the server   *****/
void * R_Server(void * arg) {

// The space we be able to use  
  int rcv_space=0; 
// Extras
   int i,cc;

   	while(1){

		cleanup(rbuf);
	  /***  leakyBucket implementation   ***/
    		pthread_mutex_lock(&bucketLock);

    		while ( bucketLevel >= bucketSize) /* while bucket is full */
      		pthread_cond_wait(&bucketChange, &bucketLock);

			rcv_space = bucketSize - bucketLevel;
                if(rcv_space>sizeof(rbuf)) rcv_space= sizeof(rbuf); // then recive min(buffsize,rcv_space)

    		pthread_mutex_unlock(&bucketLock);


         /** ready to recive **/

          cc=read(sd, rbuf, rcv_space*sizeof(char));
    	  if ( cc <= 0){
	        close(sd_T); close(sd);
		execlp(program,program,l_port,host_remoto,r_port,(char *)0); // Reset Program
      	      } else if(cc>0) 
			 { 
                         
			  rcv_space=send(ns, rbuf, cc*sizeof(char), 0);
			  if (rcv_space< 0) break;

			   pthread_mutex_lock(&bucketLock);
    		   	   bucketLevel+=rcv_space;
			   recived+=rcv_space; // infolock not required
			   pthread_mutex_unlock(&bucketLock); //Level increased
			  }
    	}

 }

/****** End of Server's Reader ************/


/****   Send to server     *****/
void * S_Server(void * arg) {
  int rc,snd_space;

	for(;;) {
		 cleanup(buf);

	  /***  leakyBucket implementation   ***/
    		pthread_mutex_lock(&bucketLock2);
 
    		while ( bucketLevel2 >= bucketSize2) /* while bucket is full */
      		pthread_cond_wait(&bucketChange2, &bucketLock2);

		snd_space = bucketSize2 - bucketLevel2;
                if(snd_space>sizeof(buf)) snd_space= sizeof(buf); // then send min(buffsize,snd_space)

    		pthread_mutex_unlock(&bucketLock2);


         		/** ready to send **/
                rc=read(ns, buf, snd_space*sizeof(char));
                if(rc<=0) {  //Reset Program
                            close(sd_T); close(sd); 
                            execlp(program,program,l_port,host_remoto,r_port,(char *)0);
                           }
		snd_space= send(sd, buf, rc, 0); 
    		if (snd_space <0 ){
    		perror("sending stream message");}
                else{
			pthread_mutex_lock(&bucketLock2); 
	     		bucketLevel2+=snd_space;
			sent+=snd_space; // infolock not required
	    		pthread_mutex_unlock(&bucketLock2); //Level increased
		     }

  	 }

}

/****** End of Server's Sender   *****/


/******** Show the INFO   *******/

void * info(void * arg)
{
	while(1) 
	{
	 /** Conection Test  ** /         sleep(3);
	 if(eof(sd_T) || eof(sd))
         { 
           printf("%s %s %s %s\n",program,l_port,host_remoto,r_port);
           close(sd_T); close(sd); sleep(1);
           execlp(program,program,l_port,host_remoto,r_port,(char *)0);
          }*/

         pthread_mutex_lock(&infoLock);
 	 system("clear");
 	 printf("************************************************************************************\n");
 	 printf("Info:\n");
 	 printf("Up Speed: %3.0f[KB/s]\tDown Speed: %3.0f[KB/s]\tSent: %3ld[B]\tRecived: %3ld[B]\t\n",(double)(leakingRate2/10),(double)(leakingRate/10),sent,recived);
 	 printf("Max Up Burst: %3d[B]\tMax Down Burst: %3d[B]\n",bucketSize2,bucketSize);
 	 printf("\nConfiguration:\n");
 	 printf("s Simetric Speed Config | u Up Speed Config | d Down Speed Config |\n");
	 printf("b Simetric Burst Config | v Up Burst Config | c Down Burst Config | x Exit\n");
 	 printf("************************************************************************************\n");
         fflush(stdout);
         pthread_mutex_unlock(&infoLock);
         usleep(700*1000); //Refresh rate 700ms 


	}

}

/********** END OF INFO *****/


main( argc, argv )
int argc;
char *argv[];
{
  struct   sockaddr_in server;
  struct sockaddr_in server_T;
  struct  hostent *hp, *gethostbyname();
  struct  servent *sp;
  struct sockaddr_in from;
  struct sockaddr_in addr;
  int fromlen;
  int length;
  int len, puerto;
  char ThisHost[80];
  char opt;
  char buffer_r[BUFSIZE];
  char buffer_s[BUFSIZE];

/* New var definition for throttle_tcp function */

/* 
We have 6 threads: 
Send to sever | Recive from server | lekyBucket for send | lekyBucket for recive  | Show info |Parameters manager
*/
pthread_t pth_snd, pth_rcv, pth_LB_rcv, pth_LB_snd, pth_info; 

printf("Starting up....\n");
//sleep(2);

/*************** TCP SERVER    ***************/
if(argc == 4){
	  puerto=atoi(argv[1]);
	  strcpy(host_remoto,argv[2]);
	  puerto_remoto=atoi(argv[3]);
          strcpy(r_port,argv[3]);
	  //printf("%s\n",host_remoto);
  }
  else if(argc == 3){
	  puerto=atoi(argv[1]);	
	  puerto_remoto=atoi(argv[2]);
	  gethostname(host_remoto, sizeof(host_remoto));
	  strcpy(r_port,argv[2]);;

	  //printf("%s\n",host_remoto);  
  }
  else {printf("throttle puerto [host_remoto] puerto_remoto  \n");exit(0);}

    strcpy(program,argv[0]);
    strcpy(l_port,argv[1]);

    sd_T = socket(PF_INET, SOCK_STREAM, 0);

    /* Create the address of the server. */
    server_T.sin_family = AF_INET;
    server_T.sin_port = htons(puerto);
    server_T.sin_addr.s_addr = htonl(INADDR_ANY); /* Use the wildcard address.*/
    len = sizeof(struct sockaddr_in);

 /* Find an abailable socket */
  while(1){
    /* Bind the socket to the address. */
    if(! bind(sd_T, (struct sockaddr *) &server_T, len))break;
	 server_T.sin_port = htons(++puerto);	
	}

	if ( getsockname (sd_T, (struct sockaddr *) &server_T, &len) ) {
		perror("getting socket name");
		exit(0);
	}
	
	printf("Server Port is: %d\nwating a client ...\n", ntohs(server_T.sin_port)); fflush(stdout);
    
    /* Listen for connections.  */
    listen(sd_T, 5);

    /* Accept a connection.  */
    ns = accept(sd_T, (struct sockaddr *) &server_T, &len);

/*********   END OF TCP SERVER    ********/


/**********  TCP Client   *************/

  sp = getservbyname("echo", "tcp");
  /* get TCPClient Host information, NAME and INET ADDRESS */

  gethostname(ThisHost, MAXHOSTNAME);

  printf("----TCP/Client running at host NAME: %s\n", ThisHost);
  if  ( (hp = gethostbyname(ThisHost)) == NULL ) {
    fprintf(stderr, "Can't find host %s\n", ThisHost);
    return -1;
  }
  bcopy ( hp->h_addr, &(server.sin_addr), hp->h_length);
  printf("    (TCP/Cleint INET ADDRESS is: %s )\n",(char *) inet_ntoa(server.sin_addr));

  /* get TCP/Server Host information, NAME and INET ADDRESS */

  if  ( (hp = gethostbyname(argv[2])) == NULL ) {
    addr.sin_addr.s_addr = inet_addr(argv[2]);
    if ((hp = gethostbyaddr((void *)&addr.sin_addr.s_addr,
        sizeof(addr.sin_addr.s_addr),AF_INET)) == NULL) {
      fprintf(stderr, "Can't find host %s\n", argv[2]);
      return -1;
    }
  }
  printf("----TCP/Server running at host NAME: %s\n", hp->h_name);
  bcopy ( hp->h_addr, &(server.sin_addr), hp->h_length);
  printf("    (TCP/Server INET ADDRESS is: %s )\n",(char *) inet_ntoa(server.sin_addr));

  /* Construct name of socket to send to. */
  server.sin_family = hp->h_addrtype;
  /* server.sin_family = AF_INET; */

  server.sin_port = htons(atoi(argv[3]));
  /*OR    server.sin_port = sp->s_port; */

  /*   Create socket on which to send  and receive */

  sd = socket (hp->h_addrtype,SOCK_STREAM,0);
  /*OR  sd = socket (PF_INET,SOCK_STREAM,0); */

  if (sd<0) {
    perror("opening stream socket");
    return -1;
  }
  /* Connect to TCP/SERVER */
  if ( connect(sd, (struct sockaddr *)&server, sizeof(server)) < 0 ) {
    close(sd);
    perror("connecting stream socket");
    return 0;
  }
  fromlen = sizeof(from);
  if (getpeername(sd, (struct sockaddr *)&from,&fromlen)<0){
    perror("could't get peername\n");
    return 1;
  }
  printf("Connected to TCP/Server:");
  printf("%s:%d\n", (char *)inet_ntoa(from.sin_addr),
      ntohs(from.sin_port));
  if ((hp = gethostbyaddr((void *)&from.sin_addr.s_addr,
      sizeof(from.sin_addr.s_addr),AF_INET)) == NULL)
    fprintf(stderr, "Can't find host %s\n", (char *)inet_ntoa(from.sin_addr));
  else
    printf("(Name is : %s)\n", hp->h_name);


  /*     get data from USER, send it SERVER,
      receive it from SERVER, display info to USER  */


sleep(2);
system("clear"); 


/****  Starting the trottle controls  *****/
 if (pthread_create(&pth_LB_rcv, NULL, leak, NULL) != 0){
    printf("can't create leak thread \n");
    return -1;
  }

 if (pthread_create(&pth_LB_snd, NULL, leak2, NULL) != 0){
    printf("can't create leak thread \n");
    return -1;
  }


   
/*  One thread for read from the server and other for send to server  */

 if (pthread_create(&pth_snd, NULL, R_Server, NULL) != 0){
    printf("can't create R_Server thread \n");
    return -1;
  }

 if (pthread_create(&pth_rcv, NULL, S_Server, NULL) != 0){
    printf("can't create S_Server thread \n");
    return -1;
  }


 if (pthread_create(&pth_info, NULL, info, NULL) != 0){
    printf("can't create info thread \n");
    return -1;
  }


while(1) 
{

 opt=getchar();
 

 pthread_mutex_lock(&infoLock);
 putchar('\n');

 if(opt=='s' || opt=='S') 
   {
    printf("Please type the simetric Speed in KB/s\n"); fflush(stdout);
    scanf("%d",&leakingRate);
    leakingRate= leakingRate*10;
    leakingRate2= leakingRate;
   }
 else if(opt=='u' || opt=='U')
   {
    printf("Please type the UP Speed in KB/s\n"); fflush(stdout);
    scanf("%d",&leakingRate2);
    leakingRate2= leakingRate2*10;
   }
 else if(opt=='d' || opt=='D')
   {
    printf("Please type the Down Speed in KB/s\n"); fflush(stdout);
    scanf("%d",&leakingRate);
    leakingRate= leakingRate*10;
   }
 if(opt=='b' || opt=='B') 
   {
    printf("Please type the simetric Burst size in B\n"); fflush(stdout);
    scanf("%d",&bucketSize);
    bucketSize2= bucketSize;
   }
 else if(opt=='v' || opt=='V')
   {
    printf("Please type the UP Speed Burst size in B\n"); fflush(stdout);
    scanf("%d",&bucketSize2);
   }
 else if(opt=='c' || opt=='C')
   {
    printf("Please type the Down Speed Burst size in B\n"); fflush(stdout);
    scanf("%d",&bucketSize);
   }
 else if(opt=='x' || opt=='X') 
  {
   printf("Good bye\n");
   close(sd); 
   return 0;
  }
  else 
  {
   printf("Invalid Option\n"); 
   fflush(stdout);
  }
 
 pthread_mutex_unlock(&infoLock);

}

  		
 close(sd);
 return 0; 
}


cleanup(buf)
char *buf;
{
  int i;
  for(i=0; i<BUFSIZE; i++) buf[i]='\0';
}    
