
/*
 *
 * ingress_thread.c -- Traffic Squeezer Ingress Traffic Handling Thread Code
 *
 * Author: Kiran K (2006-2008)
 *         kiran_cyberpro at yahoo dot com
 *
 *
 * Traffic Squeezer is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the
 * Free Software Foundation; either version 2, or (at your option) any
 * later version.
 *
 * Traffic Squeezer is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with GNU Zebra; see the file COPYING.  If not, write to the Free
 * Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA
 * 02111-1307, USA.
 */


#include "ingress_thread.h"
#include "bucket.h"

extern int compress_enabled;
extern int coalescing_enabled;

extern ts_bucket_t ingress_bucket;


void *ingress_thread_start(void *a)
{
        BYTE ingress_buff[ETH2_MAX_MTU];
        size_t ingress_buff_size=0;
        BYTE egress_buff[ETH2_MAX_MTU];
        size_t egress_buff_size=0;
        mymsgbuf ingress_msg_buff;

	//To Support Coalescing
	BYTE contains_coalesed_PDU = FALSE;
	BYTE new_ingress_buff[ETH2_MAX_MTU];
        size_t new_ingress_buff_size=0;

        ingress_msg_buff.msgType=MSG_INGRESS_STATS;

        init_stats(ingress_port_stats);
	clean_bucket(&ingress_bucket);

        sleep(3);

        while(TRUE) /* transfer frame from in-port to out-port */
        {

                if( rx_frame( ingress_port_sock_fd, ingress_buff, &ingress_buff_size)) /* Rx from ingress port */
                {
			/*
                         * If the packet is meant for the trafficsqueezer box, just discard the same.
                         */
                        /*if(memcmp(ingress_port_mac, ingress_buff, 6))
                              continue;
			*/

                        ingress_port_stats.rx++;
                        ingress_port_stats.rx_bytes+=(BYTEx16)ingress_buff_size;

			if((coalescing_enabled==TRUE) && (compress_enabled == FALSE)) //Only Coalescing enabled
			{
				if(add_bucket_PDU( ingress_buff, ingress_buff_size, &ingress_bucket)==FALSE)
				{
					if(ingress_bucket.pdu_cnt==0) //in the sense the new packet received is near to MTU
					{
						//so send it as it is
						ingress_buff[ingress_buff_size++] = PYLD_PROSD_NONE;
                             			/* Tx to the egress port  */
			                        if(tx_frame(egress_port_sock_fd, ingress_buff, ingress_buff_size))
                             			{
			                                /*printf("in-port -> out-port\n");*/
                             			}
			                        ingress_port_stats.tx++;
                             			ingress_port_stats.tx_bytes+=(BYTEx16)ingress_buff_size;
						continue;


					}
					else //The bucket has already contents and this packet looks like its exceeding the MTU
					{	// Flush/send the bucket contents first
						memset(new_ingress_buff, ZERO, ETH2_MAX_MTU);
						new_ingress_buff_size = ZERO;
						get_bucket_PDU(&ingress_bucket, new_ingress_buff, &new_ingress_buff_size);
						new_ingress_buff[new_ingress_buff_size++]=PYLD_PROSD_COAL; /* add the last byte:  */
						//Sending the same
						tx_frame(egress_port_sock_fd, new_ingress_buff, new_ingress_buff_size);
						clean_bucket(&ingress_bucket);
						ingress_port_stats.coalesedPkts++;
						ingress_port_stats.tx++;
                     				ingress_port_stats.tx_bytes+=(BYTEx16)new_ingress_buff_size;

						{ //now try to put this new packet into bucket else send as it is
						  if(add_bucket_PDU( ingress_buff, \
							ingress_buff_size, &ingress_bucket)==FALSE)
                                		  {
							//in the sense the new empty packet received is near to MTU
                                        	  	if(ingress_bucket.pdu_cnt==0)
                                        		{
								//so send it as it is
                                                		ingress_buff[ingress_buff_size++] = PYLD_PROSD_NONE;
                                                		/* Tx to the egress port - */
                                                		if(tx_frame(egress_port_sock_fd, \
									ingress_buff, ingress_buff_size))
                                                		{
                                                        		/*printf("in-port -> out-port\n");*/
                                                		}
                                                		ingress_port_stats.tx++;
                                                		ingress_port_stats.tx_bytes+=(BYTEx16)ingress_buff_size;
                                                		continue;
							}
						  }

 						}
					}

				}
				else //if packet got coalesced then continue receiving next packet before sending the same
				{
					continue;
				}
			#if 0
				{	int i;
					for(i=0;i<ingress_bucket.pdu_cnt;i++)
					{
						printf("pdu[%d] size = %d\n", i, ingress_bucket.pdu_size[i]);
					}
					printf("Total bucket PDUs: = %d\n", ingress_bucket.pdu_cnt);
					printf("Total bucket Size: = %d\n", ingress_bucket.size);
				}
			#endif

			}


			if( (coalescing_enabled == FALSE) && (compress_enabled == TRUE) ) //Only Compression Enabled
                        {
                                unsigned char comp_buff[ETH2_MAX_MTU];
                                size_t comp_buff_size=0;
                                if(compress(ingress_buff+ETH2_FRAME_PAYLOAD_OFFSET, \
					ingress_buff_size-ETH2_FRAME_PAYLOAD_OFFSET, comp_buff, &comp_buff_size)==FALSE)
                                {
				        //printf("error @ compression \n");
					ingress_buff[ingress_buff_size++]=PYLD_PROSD_NONE; /* add the last byte  */

				}
                                else
                                {       //Select the frame only which got its payload compressed
					if((comp_buff_size < (ingress_buff_size-ETH2_FRAME_PAYLOAD_OFFSET))
						&& (ingress_buff_size > 200) /* compress only if frame is >70 bytes added: 28_Feb_08)*/)
					{	memcpy(ingress_buff+ETH2_FRAME_PAYLOAD_OFFSET, comp_buff, comp_buff_size);
                                        	ingress_buff_size = (comp_buff_size+ETH2_FRAME_PAYLOAD_OFFSET);
						ingress_buff[ingress_buff_size++]=PYLD_PROSD_COMP; /* add the last byte:  */
						ingress_port_stats.comprsedPkts++;
					}
					else
                                        {	ingress_buff[ingress_buff_size++]=PYLD_PROSD_NONE; /* add the last byte: */
					}

                                }
				/* Tx to the egress port - after compression */
                     		if(tx_frame(egress_port_sock_fd, ingress_buff, ingress_buff_size))
                     		{
                        		/*printf("in-port -> out-port(after compression)\n");*/
                     		}
				ingress_port_stats.tx++;
                     		ingress_port_stats.tx_bytes+=(BYTEx16)ingress_buff_size;

                        }

			// Both enabled however do first coalescing and do compression on the same.
                           // or alternatively do compression then coalescing and again compression
			if( (compress_enabled == TRUE) && (coalescing_enabled==TRUE) )
                        {
				// First Coalescing all the frames
				if(add_bucket_PDU( ingress_buff, ingress_buff_size, &ingress_bucket)==FALSE)
				{
					if(ingress_bucket.pdu_cnt==0) //in the sense the new packet received is near to MTU
					{
						//So compress non coalesced frame and send
						unsigned char comp_buff[ETH2_MAX_MTU];
                                		size_t comp_buff_size=0;
                                		if(compress(ingress_buff+ETH2_FRAME_PAYLOAD_OFFSET, \
							ingress_buff_size-ETH2_FRAME_PAYLOAD_OFFSET, comp_buff, &comp_buff_size)==FALSE)
                                		{
				        		//frame got neither coalesced or compressed so sending as it is
							ingress_buff[ingress_buff_size++]=PYLD_PROSD_NONE; /* add the last byte  */

						}
                                		else
                                		{       //Select the frame only which got its payload compressed
							if((comp_buff_size < (ingress_buff_size-ETH2_FRAME_PAYLOAD_OFFSET)) \
								&& (ingress_buff_size > 200))
							{	memcpy(ingress_buff+ETH2_FRAME_PAYLOAD_OFFSET, comp_buff, comp_buff_size);
                                        			ingress_buff_size = (comp_buff_size+ETH2_FRAME_PAYLOAD_OFFSET);
								ingress_buff[ingress_buff_size++]=PYLD_PROSD_COMP; /* add the last byte:  */
								ingress_port_stats.comprsedPkts++;
							}
							else
                                        		{	ingress_buff[ingress_buff_size++]=PYLD_PROSD_NONE; /* add the last byte: */
							}

                                		}

						/* Tx to the egress port  */
			                        if(tx_frame(egress_port_sock_fd, ingress_buff, ingress_buff_size))
                             			{
			                                /*printf("in-port -> out-port\n");*/
                             			}
						ingress_port_stats.tx++;
                             			ingress_port_stats.tx_bytes+=(BYTEx16)ingress_buff_size;
						continue;
					}
					else //The bucket has already contents and this packet looks like its exceeding the MTU
					{	// Flush/send the bucket contents first
						memset(new_ingress_buff, ZERO, ETH2_MAX_MTU);
						new_ingress_buff_size = ZERO;
						get_bucket_PDU(&ingress_bucket, new_ingress_buff, &new_ingress_buff_size);
						//now compress the coalesced frame if possible
						{
							unsigned char comp_buff[ETH2_MAX_MTU];
                                			size_t comp_buff_size=0;
                                			if(compress(new_ingress_buff+ETH2_FRAME_PAYLOAD_OFFSET, \
							new_ingress_buff_size-ETH2_FRAME_PAYLOAD_OFFSET, comp_buff, &comp_buff_size)==FALSE)
                                			{
				        			//frame didnt got compressed so sending the coalesced frame
								new_ingress_buff[new_ingress_buff_size++]=PYLD_PROSD_COAL;
							}
                                			else
                                			{       //Select the frame only which got its payload compressed
								if((comp_buff_size < (new_ingress_buff_size-ETH2_FRAME_PAYLOAD_OFFSET)) \
									&& (new_ingress_buff_size > 200))
								{	memcpy(new_ingress_buff+ETH2_FRAME_PAYLOAD_OFFSET, \
									comp_buff, comp_buff_size);
                                        				new_ingress_buff_size = (comp_buff_size+ETH2_FRAME_PAYLOAD_OFFSET);
									new_ingress_buff[new_ingress_buff_size++]=PYLD_PROSD_COAL_COMP;
									ingress_port_stats.comprsedPkts++;
								}
								else
                                        			{	new_ingress_buff[new_ingress_buff_size++]=PYLD_PROSD_COAL;
								}
		                               		}
						}

						//Sending the same
						tx_frame(egress_port_sock_fd, new_ingress_buff, new_ingress_buff_size);
						clean_bucket(&ingress_bucket);
						ingress_port_stats.coalesedPkts++;
						ingress_port_stats.tx++;
                     				ingress_port_stats.tx_bytes+=(BYTEx16)new_ingress_buff_size;

						{ //now try to put this new packet into bucket else send as it is
						  if(add_bucket_PDU( ingress_buff, \
							ingress_buff_size, &ingress_bucket)==FALSE)
                                		  {
							//in the sense the new empty packet received is near to MTU
                                        	  	if(ingress_bucket.pdu_cnt==0)
                                        		{

								//So compress non coalesced frame and send
								unsigned char comp_buff[ETH2_MAX_MTU];
                                				size_t comp_buff_size=0;
                                				if(compress(ingress_buff+ETH2_FRAME_PAYLOAD_OFFSET, \
									ingress_buff_size-ETH2_FRAME_PAYLOAD_OFFSET, \
									comp_buff, &comp_buff_size)==FALSE)
                                				{
				        				//frame got neither coalesced or compressed so sending as it is
									ingress_buff[ingress_buff_size++]=PYLD_PROSD_NONE;
								}
                                				else
                                				{       //Select the frame only which got its payload compressed
									if((comp_buff_size < (ingress_buff_size-ETH2_FRAME_PAYLOAD_OFFSET)) \
										&& (ingress_buff_size > 200))
									{	memcpy(ingress_buff+ETH2_FRAME_PAYLOAD_OFFSET, \
											comp_buff, comp_buff_size);
                                        					ingress_buff_size = \
											 (comp_buff_size+ETH2_FRAME_PAYLOAD_OFFSET);
										ingress_buff[ingress_buff_size++]=PYLD_PROSD_COMP;
										ingress_port_stats.comprsedPkts++;
									}
									else
                                        				{	ingress_buff[ingress_buff_size++]=PYLD_PROSD_NONE;
									}
								}

                                                		ingress_buff[ingress_buff_size++] = PYLD_PROSD_NONE;
                                                		/* Tx to the egress port - */
                                                		if(tx_frame(egress_port_sock_fd, \
									ingress_buff, ingress_buff_size))
                                                		{
                                                        		/*printf("in-port -> out-port\n");*/
                                                		}
                                                		ingress_port_stats.tx++;
                                                		ingress_port_stats.tx_bytes+=(BYTEx16)ingress_buff_size;
                                                		continue;
							}
						  }

 						}
					}

				}
				else //if packet got coalesced then continue receiving next packet before sending the same
				{
					continue;
				}
			}

			if( (compress_enabled == FALSE) && (coalescing_enabled==FALSE) ) // None enabled
			{    // If compression is disabled globally
			     ingress_buff[ingress_buff_size++] = PYLD_PROSD_NONE;
			     /* Tx to the egress port - after compression */
                     	     if(tx_frame(egress_port_sock_fd, ingress_buff, ingress_buff_size))
                     	     {
                        	/*printf("in-port -> out-port(after compression)\n");*/
                     	     }
			     ingress_port_stats.tx++;
                     	     ingress_port_stats.tx_bytes+=(BYTEx16)ingress_buff_size;
			}



                    /*
                     * tx the message (ingress stats)
                     */
                    {
                         memcpy( &ingress_msg_buff.stats, &ingress_port_stats, sizeof(ts_stats_t));
                         ingress_msg_buff.mtype=MESSAGE_TYPE;
                         send_message(stats_msgQ_id, &ingress_msg_buff);
                    }
                }
        }
}



