PeerStreamer

libml_a_SOURCES = BUGS.txt ml.c ml_log.c util/stun.c \

	util/udpSocket.c util/rateLimiter.c util/queueManagement.c

#        ${COMPILER} -o echoServer.o echoServer.c ml.h ml.c transmissionHandler.h transmissionHandler.c util/stun.h util/stun.c util/udpSocket.h util/udpSocket.c ${EVENT} ${MATH} ${GDB}

  * Configure the verbosity of messages

  * @param log_level [0-4] the lower the less messages are printed out

void mlSetVerbosity(int log_level);

/**

  * Configure the parameters for output rate control.

  * @param bucketsize The size of the bucket in kbytes

  * @param drainrate The amount of kbytes draining in a second. If drainrate is <=0, then rateControl is completely disabled (all packets are passed).

  * @param maxQueueSize In kbytes. Max data stored while limiting the output rate. If 0 packets limitted by drainrate are dropped.

  * @param maxQueueSizeRTX In kbytes. Max data waiting for the retransmission if needed.

  * @param maxTimeToHold. Time for which sent packets are stored in RTX queue in seconds.

*/

void mlSetRateLimiterParams(int bucketsize, int drainrate, int maxQueueSize, int maxQueueSizeRTX, double maxTimeToHold);

#include <ml_all.h>

#define PMTU_TIMEOUT 500000 // in usec

#ifdef RTX

/*

 * default timeout value for a packet reception

 */

#define PKT_RECV_TIMEOUT_DEFAULT { 0, 50000 } // 50 ms

/*

 * default timeout value for a packet reception

 */

#define LAST_PKT_RECV_TIMEOUT_DEFAULT { 1, 700000 }

/*

 * default fraction of RECV_TIMEOUT_DEFAULT for a last packet(s) reception timeout

 */

#define LAST_PKT_RECV_TIMEOUT_FRACTION 0.7

#endif

#ifndef WIN32

#endif

#ifdef RTX

//*********Counters**********

struct Counters {

	unsigned int receivedCompleteMsgCounter;

	unsigned int receivedIncompleteMsgCounter;

	unsigned int receivedDataPktCounter;

	unsigned int receivedRTXDataPktCounter;

	unsigned int receivedNACK1PktCounter;

	unsigned int receivedNACKMorePktCounter;

	unsigned int sentDataPktCounter;

	unsigned int sentRTXDataPktCtr;

	unsigned int sentNACK1PktCounter;

	unsigned int sentNACKMorePktCounter;

} counters;

extern unsigned int sentRTXDataPktCounter;

/*

 * receive timeout for a packet

 */

static struct timeval pkt_recv_timeout = PKT_RECV_TIMEOUT_DEFAULT;

static struct timeval last_pkt_recv_timeout = LAST_PKT_RECV_TIMEOUT_DEFAULT;

void mlShowCounters() {

	counters.sentRTXDataPktCtr = sentRTXDataPktCounter;

	fprintf(stderr, "\nreceivedCompleteMsgCounter: %d\nreceivedIncompleteMsgCounter: %d\nreceivedDataPktCounter: %d\nreceivedRTXDataPktCounter: %d\nreceivedNACK1PktCounter: %d\nreceivedNACKMorePktCounter: %d\nsentDataPktCounter: %d\nsentRTXDataPktCtr: %d\nsentNACK1PktCounter: %d\nsentNACKMorePktCounter: %d\n", counters.receivedCompleteMsgCounter, counters.receivedIncompleteMsgCounter, counters.receivedDataPktCounter, counters.receivedRTXDataPktCounter, counters.receivedNACK1PktCounter, counters.receivedNACKMorePktCounter, counters.sentDataPktCounter, counters.sentRTXDataPktCtr, counters.sentNACK1PktCounter, counters.sentNACKMorePktCounter);

	return;

}

void recv_nack_msg(struct msg_header *msg_h, char *msgbuf, int msg_size)

{

	struct nack_msg *nackmsg;

	msgbuf += msg_h->len_mon_data_hdr;

	msg_size -= msg_h->len_mon_data_hdr;

	nackmsg = (struct nack_msg*) msgbuf;

	unsigned int gapSize = nackmsg->offsetTo - nackmsg->offsetFrom;

	//if (gapSize == 1349) counters.receivedNACK1PktCounter++;

	//else counters.receivedNACKMorePktCounter++;

	rtxPacketsFromTo(nackmsg->con_id, nackmsg->msg_seq_num, nackmsg->offsetFrom, nackmsg->offsetTo);	

}

void send_msg(int con_id, int msg_type, void* msg, int msg_len, bool truncable, send_params * sParams);

void pkt_recv_timeout_cb(int fd, short event, void *arg){

	int recv_id = (long) arg;

	debug("ML: recv_timeout_cb called. Timeout for id:%d\n",recv_id);

	//check if message still exists	

	if (recvdatabuf[recv_id] == NULL) return;

	//check if gap was filled in the meantime

	if (recvdatabuf[recv_id]->gapArray[recvdatabuf[recv_id]->firstGap].offsetFrom == recvdatabuf[recv_id]->gapArray[recvdatabuf[recv_id]->firstGap].offsetTo) {

		recvdatabuf[recv_id]->firstGap++;

		return;	

	}

	struct nack_msg nackmsg;

	nackmsg.con_id = recvdatabuf[recv_id]->txConnectionID;

	nackmsg.msg_seq_num = recvdatabuf[recv_id]->seqnr;

	nackmsg.offsetFrom = recvdatabuf[recv_id]->gapArray[recvdatabuf[recv_id]->firstGap].offsetFrom;

	nackmsg.offsetTo = recvdatabuf[recv_id]->gapArray[recvdatabuf[recv_id]->firstGap].offsetTo;

	recvdatabuf[recv_id]->firstGap++;

	unsigned int gapSize = nackmsg.offsetTo - nackmsg.offsetFrom;

	send_msg(recvdatabuf[recv_id]->connectionID, ML_NACK_MSG, (char *) &nackmsg, sizeof(struct nack_msg), true, &(connectbuf[recvdatabuf[recv_id]->connectionID]->defaultSendParams));	

}

void last_pkt_recv_timeout_cb(int fd, short event, void *arg){

	int recv_id = (long) arg;

	debug("ML: recv_timeout_cb called. Timeout for id:%d\n",recv_id);

	if (recvdatabuf[recv_id] == NULL) {

		return;

	}

	if (recvdatabuf[recv_id]->expectedOffset == recvdatabuf[recv_id]->bufsize - recvdatabuf[recv_id]->monitoringDataHeaderLen) return;

	struct nack_msg nackmsg;

	nackmsg.con_id = recvdatabuf[recv_id]->txConnectionID;

	nackmsg.msg_seq_num = recvdatabuf[recv_id]->seqnr;

	nackmsg.offsetFrom = recvdatabuf[recv_id]->expectedOffset;

	nackmsg.offsetTo = recvdatabuf[recv_id]->bufsize - recvdatabuf[recv_id]->monitoringDataHeaderLen;

	unsigned int gapSize = nackmsg.offsetTo - nackmsg.offsetFrom;

	send_msg(recvdatabuf[recv_id]->connectionID, ML_NACK_MSG, &nackmsg, sizeof(struct nack_msg), true, &(connectbuf[recvdatabuf[recv_id]->connectionID]->defaultSendParams));	

}

#endif

void send_msg(int con_id, int msg_type, void* msg, int msg_len, bool truncable, send_params * sParams) {

			int priority = 0; 

			if ((msg_type == ML_CON_MSG)

#ifdef RTX

 || (msg_type == ML_NACK_MSG)

#endif

) priority = HP;

			//fprintf(stderr,"*******************************ML.C: Sending packet: msg_h.offset: %d msg_h.msg_seq_num: %d\n",ntohl(msg_h.offset),ntohl(msg_h.msg_seq_num));

			switch(queueOrSendPacket(socketfd, iov, 4, &udpgen.udpaddr,priority)) {

                                        debug("THROTTLE on output"); 

#ifdef RTX

					if (msg_type < 127) counters.sentDataPktCounter++;

#endif

	//fprintf(stderr, "sentDataPktCounter after msg_seq_num = %d: %d\n", msg_h.msg_seq_num, counters.sentDataPktCounter);

	//fprintf(stderr, "sentRTXDataPktCounter after msg_seq_num = %d: %d\n", msg_h.msg_seq_num, counters.sentRTXDataPktCtr);

int sendPacket(const int udpSocket, struct iovec *iov, int len, struct sockaddr_in *socketaddr) {

	//monitoring layer hook

debug("SENDP1");

	if(get_Send_pkt_inf_cb != NULL && iov[1].iov_len) {

		mon_pkt_inf pkt_info;	

		struct msg_header *msg_h  = (struct msg_header *) iov[0].iov_base;

		memset(iov[1].iov_base,0,iov[1].iov_len);

		pkt_info.remote_socketID = &(connectbuf[ntohl(msg_h->local_con_id)]->external_socketID);

		pkt_info.buffer = iov[3].iov_base;

		pkt_info.bufSize = iov[3].iov_len;

		pkt_info.msgtype = msg_h->msg_type;

		pkt_info.dataID = ntohl(msg_h->msg_seq_num);

		pkt_info.offset = ntohl(msg_h->offset);

		pkt_info.datasize = ntohl(msg_h->msg_length);

		pkt_info.monitoringHeaderLen = iov[1].iov_len;

		pkt_info.monitoringHeader = iov[1].iov_base;

		pkt_info.ttl = -1;

		memset(&(pkt_info.arrival_time),0,sizeof(struct timeval));

		(get_Send_pkt_inf_cb) ((void *) &pkt_info);

	}

 	//struct msg_header *msg_h;

    //msg_h = (struct msg_header *) iov[0].iov_base;        

	//fprintf(stderr,"*** Sending packet - msgSeqNum: %d offset: %d\n",ntohl(msg_h->msg_seq_num),ntohl(msg_h->offset));

debug("SENDP2");

	return sendPacketFinal(udpSocket, iov, len, socketaddr);

}

  {

                        char buf[SOCKETID_STRING_SIZE];

                        mlSocketIDToString(&((struct conn_msg*)connectbuf[con_id]->ctrl_msg_buf)->sock_id,buf,sizeof(buf));

                        debug("Local socket_address sent in INVITE: %s, sizeof msg %ld\n", buf, sizeof(struct conn_msg));

   }

	struct timeval tout = {0,0};

	tout.tv_usec = PMTU_TIMEOUT * (1.0+ 0.1 *((double)rand()/(double)RAND_MAX-0.5));

		//Workaround to support reuse of socketID

				connectbuf[free_con_id]->external_socketID.internal_addr.udpaddr.sin_family=AF_INET;

				connectbuf[free_con_id]->external_socketID.external_addr.udpaddr.sin_family=AF_INET;

		{

                	char buf[SOCKETID_STRING_SIZE];

                	mlSocketIDToString(&local_socketID,buf,sizeof(buf));

 			debug("received local socket_address: %s\n", buf);

		}

#ifdef RTX

		counters.receivedIncompleteMsgCounter++;

		//mlShowCounters();

		//fprintf(stderr,"******Cleaning slot for inclomplete msg_seq_num: %d\n", recvdatabuf[recv_id]->seqnr);		

#endif

 		//(receive_data_callback) (recvdatabuf[recv_id]->recvbuf + recvdatabuf[recv_id]->monitoringDataHeaderLen, recvdatabuf[recv_id]->bufsize - recvdatabuf[recv_id]->monitoringDataHeaderLen, recvdatabuf[recv_id]->msgtype, &rParams);

	debug("ML: received packet of size %d with rconID:%d lconID:%d type:%d offset:%d inlength: %d\n",bufsize,msg_h->remote_con_id,msg_h->local_con_id,msg_h->msg_type,msg_h->offset, msg_h->msg_length);

#ifdef RTX

	counters.receivedDataPktCounter++;

#endif	

	for (recv_id = 0; recv_id < RECVDATABUFSIZE; recv_id++) {

  }

		debug(" recv id not found (free found: %d)\n", free_recv_id);

#ifdef RTX

		recvdatabuf[recv_id]->txConnectionID = msg_h->local_con_id;

		recvdatabuf[recv_id]->expectedOffset = 0;

		recvdatabuf[recv_id]->gapCounter = 0;

		recvdatabuf[recv_id]->firstGap = 0;

		recvdatabuf[recv_id]->last_pkt_timeout_event = NULL;

#endif

		//fprintf(stderr,"Hoooooray!! We have first packet of some message!!\n");

	recvdatabuf[recv_id]->arrivedBytes += bufsize; 

	//fprintf(stderr,"Arrived bytes: %d Offset: %d Expected offset: %d\n",recvdatabuf[recv_id]->arrivedBytes/1349,msg_h->offset/1349,recvdatabuf[recv_id]->expectedOffset/1349);

#ifdef RTX

	// detecting a new gap	

	if (msg_h->offset > recvdatabuf[recv_id]->expectedOffset) {

		recvdatabuf[recv_id]->gapArray[recvdatabuf[recv_id]->gapCounter].offsetFrom = recvdatabuf[recv_id]->expectedOffset;

		recvdatabuf[recv_id]->gapArray[recvdatabuf[recv_id]->gapCounter].offsetTo = msg_h->offset;

		if (recvdatabuf[recv_id]->gapCounter < RTX_MAX_GAPS - 1) recvdatabuf[recv_id]->gapCounter++;

		evtimer_add(event_new(base, -1, EV_TIMEOUT, &pkt_recv_timeout_cb, (void *) (long)recv_id), &pkt_recv_timeout);

	}

	//filling the gap by delayed packets

	if (msg_h->offset < recvdatabuf[recv_id]->expectedOffset){

		counters.receivedRTXDataPktCounter++;

		//skip retransmitted packets

		if (recvdatabuf[recv_id]->firstGap < recvdatabuf[recv_id]->gapCounter && msg_h->offset >= recvdatabuf[recv_id]->gapArray[recvdatabuf[recv_id]->firstGap].offsetFrom) {

			int i;

			//fprintf(stderr,"firstGap: %d	gapCounter: %d\n", recvdatabuf[recv_id]->firstGap, recvdatabuf[recv_id]->gapCounter);

			for (i = recvdatabuf[recv_id]->firstGap; i < recvdatabuf[recv_id]->gapCounter; i++){

				if (msg_h->offset == recvdatabuf[recv_id]->gapArray[i].offsetFrom) {

					recvdatabuf[recv_id]->gapArray[i].offsetFrom += bufsize;

					break;

				}

				if (msg_h->offset == (recvdatabuf[recv_id]->gapArray[i].offsetTo - bufsize)) {

					recvdatabuf[recv_id]->gapArray[i].offsetTo -= bufsize;

					break;

				}

			}

		} else {//fprintf(stderr,"Skipping retransmitted packets in filling the gap.\n"); 

			//counters.receivedRTXDataPktCounter++;

			}

	}

	//updating the expectedOffset	

	if (msg_h->offset >= recvdatabuf[recv_id]->expectedOffset) recvdatabuf[recv_id]->expectedOffset = msg_h->offset + bufsize;

#endif

#ifdef RTX

				counters.receivedCompleteMsgCounter++;

				//mlShowCounters();

#endif

#ifdef RTX

			if (recvdatabuf[recv_id]->last_pkt_timeout_event) {

				debug("ML: freeing last packet timeout for %d",recv_id);

				event_del(recvdatabuf[recv_id]->last_pkt_timeout_event);

				event_free(recvdatabuf[recv_id]->last_pkt_timeout_event);

				recvdatabuf[recv_id]->last_pkt_timeout_event = NULL;

			}

			//fprintf(stderr,"******Cleaning slot for clomplete msg_seq_num: %d\n", recvdatabuf[recv_id]->seqnr);	

#endif

#ifdef RTX

				recvdatabuf[recv_id]->last_pkt_timeout_event = event_new(base, -1, EV_TIMEOUT, &last_pkt_recv_timeout_cb, (void *) (long)recv_id);

				evtimer_add(recvdatabuf[recv_id]->last_pkt_timeout_event, &last_pkt_recv_timeout);

#endif

		struct timeval tout = {0,0};

		tout.tv_usec = PMTU_TIMEOUT * (1.0+ 0.1 *((double)rand()/(double)RAND_MAX-0.5));

	if (connectbuf[con_id]->pmtusize == P_ERROR) {

		struct timeval tout = {0,0};

		tout.tv_usec = PMTU_TIMEOUT * (1.0+ 0.1 *((double)rand()/(double)RAND_MAX-0.5));

		return P_ERROR;

    if (new_pmtusize == P_ERROR) {

		debug("ML: recv_pkg: parse stun message called on %d bytes\n", recvSize);

        uint32_t inlen = ntohl(msg_h->msg_length);

        if(inlen > 0x20000 || inlen < ntohl(msg_h->offset) || inlen == 0) {

            warn("ML: BAD PACKET received from: %s:%d (len: %d < %d [=%08X] o:%d)", 

                  inet_ntoa(recv_addr.sin_addr), recv_addr.sin_port,

                               recvSize, inlen, inlen, ntohl(msg_h->offset));

 warn("ML: received %d: %02X %02X %02X %02X %02X %02X - %02X %02X %02X %02X %02X %02X", recvSize,

            msgbuf[0], msgbuf[1],msgbuf[2],msgbuf[3],msgbuf[4],msgbuf[5],

            msgbuf[6],msgbuf[7],msgbuf[8],msgbuf[9],msgbuf[10],msgbuf[11]);

            return;

       }

                else if(msg_h->remote_con_id < 0 || 

                       msg_h->remote_con_id >= CONNECTBUFSIZE || 

                       connectbuf[msg_h->remote_con_id] == NULL) {

#ifdef RTX

		case ML_NACK_MSG:

			debug("ML: received nack pkg\n");

			recv_nack_msg(msg_h, bufptr, msg_size);

			break;

#endif

debug("X. NatTrTo %d\n", NAT_traversal);

debug("X. NatTrTo\n");

        debug("X. create_socket %s, %d\n", ipaddr, port);

/*X*/ //  fprintf(stderr,"MLINIT1 %s, %d, %s, %d\n", ipaddr, port, stun_ipaddr, stun_port);

/*X*/ //  fprintf(stderr,"MLINIT1\n");

void mlSetRateLimiterParams(int bucketsize, int drainrate, int maxQueueSize, int maxQueueSizeRTX, double maxTimeToHold) {

	setQueuesParams (maxQueueSize, maxQueueSizeRTX, maxTimeToHold);

void mlSetVerbosity (int log_level) {

	setLogLevel(log_level);

}

#ifdef RTX

	unsigned int total_usec = recv_timeout.tv_sec * 1000000 + recv_timeout.tv_usec;

	total_usec = total_usec * LAST_PKT_RECV_TIMEOUT_FRACTION;

	last_pkt_recv_timeout.tv_sec = total_usec / 1000000;

	last_pkt_recv_timeout.tv_usec = total_usec - last_pkt_recv_timeout.tv_sec * 1000000;

	fprintf(stderr,"Timeout for receiving message: %d : %d\n", recv_timeout.tv_sec, recv_timeout.tv_usec);	

	fprintf(stderr,"Timeout for last pkt: %d : %d\n", last_pkt_recv_timeout.tv_sec, last_pkt_recv_timeout.tv_usec);	

#endif

#include <stdlib.h>

#include <unistd.h>

#include <stdio.h>

#include <stddef.h>

#include <stdint.h>

#include <string.h>

#include <sys/types.h>

#include <time.h>

#include <math.h>

#include <assert.h>

#include <errno.h>

#ifndef WIN32

#include <arpa/inet.h>

#include <netdb.h>

#include <netinet/in.h>

#include <sys/socket.h>

#include <fcntl.h>

#else

#include <winsock2.h>

#include <ws2tcpip.h>

#endif

#include "util/udpSocket.h"

#include "util/stun.h"

#include "transmissionHandler.h"

#include "util/rateLimiter.h"

#include "util/queueManagement.h"

#define LOG_MODULE "[ml] "

#include "ml_log.h"

/* Set defualt log level */

int ml_log_level = 2; //2 -> WARNING

void setLogLevel(int ll) {

	ml_log_level = ll;

}

extern int ml_log_level;

extern void setLogLevel(int ll);

#define DPRINT(ll, format, ... )  {struct timeval tnow; if(ll <= ml_log_level) {gettimeofday(&tnow,NULL); fprintf(stderr, "%ld.%03ld "format, tnow.tv_sec, tnow.tv_usec/1000, ##__VA_ARGS__ );fprintf(stderr,format[strlen(format)-1] == '\n'?"":"\n"); fflush(stderr);}}

#define debug(format, ... ) DPRINT(4 ,format, ##__VA_ARGS__ )

#define info(format, ... )  DPRINT(3, format, ##__VA_ARGS__ )

#define warn(format, ... )  DPRINT(2, format, ##__VA_ARGS__ )

#define error(format, ... )  DPRINT(1, format, ##__VA_ARGS__ )

#define fatal(format, ... )  { DPRINT(0, format, ##__VA_ARGS__ ); exit(-1); }

#ifndef WIN32

#include <arpa/inet.h>

#include <sys/uio.h>

#endif

#ifndef WIN32

#endif

#ifdef RTX

/**

 * This is the maximum number of gaps RTX can keep track of inside one message

 */

#define RTX_MAX_GAPS 25

#define ML_NACK_MSG 128

#endif

typedef enum {MAX = 1472, DSL = 1464, DSLMEDIUM = 1422, DSLSLIM = 1372, BELOWDSL = 1172, MIN = 472, P_ERROR = -1} pmtu;

#ifdef RTX

struct gap {

	int offsetFrom;

	int offsetTo;

};

#endif

#ifdef RTX

  struct event* last_pkt_timeout_event;

  int txConnectionID;

  int expectedOffset;

  int gapCounter; //index of the first "free slot"

  int firstGap;	//first gap which hasn't been handled yet (for which the NACK hasn't been sent yet)

  struct gap gapArray[RTX_MAX_GAPS];

#endif

#ifdef RTX

/************modifications-START************/

struct nack_msg {

	int32_t con_id;		///local connectionID of the transmitter

	int32_t msg_seq_num;

	uint32_t offsetFrom;

	uint32_t offsetTo;

} __attribute__((packed));

/************modifications-END**************/

#endif

int sendPacket(const int udpSocket, struct iovec *iov, int len, struct sockaddr_in *socketaddr);

/*

 *          

 *

 *

 * 

 *	Agnieszka Witczak & Szymon Kuc

 *     

 */

#include "../ml_all.h"

FILE *file = NULL;

PacketQueue TXqueue;

PacketQueue RTXqueue;

unsigned int sentRTXDataPktCounter = 0;

int TXmaxSize = 6000*1500;			//bytes

int RTXmaxSize = 6000*1500;			//bytes

struct timeval maxTimeToHold = {5,0};

PacketContainer* createPacketContainer(const int uSoc,struct iovec *ioVector,int iovlen,struct sockaddr_in *sockAddress, unsigned char prior) {

	PacketContainer *packet = malloc(sizeof(PacketContainer));

	packet->udpSocket = uSoc;

	packet->iovlen = iovlen;

	packet->next = NULL;

	packet->pktLen = ioVector[0].iov_len + ioVector[1].iov_len + ioVector[2].iov_len + ioVector[3].iov_len;

	packet->priority = prior;

 	int i;

        packet->iov = malloc(sizeof(struct iovec) * iovlen);

        for (i=0; i<iovlen; i++){

                packet->iov[i].iov_len = ioVector[i].iov_len;

                packet->iov[i].iov_base = malloc(packet->iov[i].iov_len);

                memcpy(packet->iov[i].iov_base, ioVector[i].iov_base, packet->iov[i].iov_len);

        }

        packet->socketaddr = malloc(sizeof(struct sockaddr_in));

        memcpy(packet->socketaddr, sockAddress, sizeof(struct sockaddr_in));

	int packet_len = ioVector[0].iov_len + ioVector[1].iov_len + ioVector[2].iov_len + ioVector[3].iov_len;

	return packet;

}

void destroyPacketContainer(PacketContainer* pktContainer){

	if (pktContainer != NULL){

	        int i;

        	for (i=0; i < pktContainer->iovlen; i++) free(pktContainer->iov[i].iov_base);

        	free(pktContainer->iov);

        	free(pktContainer->socketaddr);

        	free(pktContainer);

	}

}

int addPacketTXqueue(PacketContainer *packet) {

	if ((TXqueue.size + packet->pktLen) > TXmaxSize) {

		fprintf(stderr,"[queueManagement::addPacketTXqueue] -- Sorry! Max size for TX queue. Packet will be discarded. \n");

		destroyPacketContainer(packet);

		return THROTTLE;

	}

	TXqueue.size += packet->pktLen;	

	if (TXqueue.head == NULL) {			//adding first element

		TXqueue.head = packet;

		TXqueue.tail = packet;		

	} else {					//adding at the end of the queue

		TXqueue.tail->next = packet;

		TXqueue.tail = packet;

	}

	return OK;

}

PacketContainer* takePacketToSend() {			//returns pointer to packet or NULL if queue is empty

	if (TXqueue.head != NULL) {

		PacketContainer *packet = TXqueue.head;

		TXqueue.size -= packet->pktLen;

		TXqueue.head = TXqueue.head->next;

		packet->next = NULL;

		if (TXqueue.head == NULL) TXqueue.tail = NULL;					

		return packet;

	}

	else return NULL;	

}

#ifdef RTX

void addPacketRTXqueue(PacketContainer *packet) {

	//removing old packets - because of maxTimeToHold

	struct timeval now, age;

	gettimeofday(&now, NULL);

	PacketContainer *tmp = RTXqueue.head;

	while (tmp != NULL) {

		age.tv_sec = now.tv_sec - tmp->timeStamp.tv_sec;

		age.tv_usec = now.tv_usec - tmp->timeStamp.tv_usec;

		if (age.tv_sec > maxTimeToHold.tv_sec) {

			tmp = tmp->next;

			removeOldestPacket();

		}

		else break;

	}

	while ((RTXqueue.size + packet->pktLen) > RTXmaxSize) {

		removeOldestPacket();

	}

	//adding timeStamp

	gettimeofday(&(packet->timeStamp), NULL);

	//finally - adding packet

	RTXqueue.size += packet->pktLen;

	packet->next = NULL;

	if (RTXqueue.head == NULL) {			//adding first element

		RTXqueue.head = packet;

		RTXqueue.tail = packet;

		return;	

	} else {					//adding at the end of the queue

		RTXqueue.tail->next = packet;

		RTXqueue.tail = packet;

		return;

	}

}

int removeOldestPacket() {			//return 0 if success, else (queue empty) return 1

	if (RTXqueue.head != NULL) {

		RTXqueue.size -= RTXqueue.head->pktLen;

		PacketContainer *pointer = RTXqueue.head;		

		RTXqueue.head = RTXqueue.head->next;

		destroyPacketContainer(pointer);

		return 0;

	}

	return 1;	

}

#endif

void setQueuesParams (int TXsize, int RTXsize, double maxTTHold) { //in bytes, bytes, sexonds

	TXmaxSize = TXsize;

	RTXmaxSize = RTXsize;

	maxTimeToHold.tv_sec = (int)floor(maxTTHold);

	maxTimeToHold.tv_usec = (int)(1000000.0 * fmod(maxTTHold, 1.0));

}

int isQueueEmpty() {

	if (TXqueue.head == NULL) return 1;

	return 0;

}

int getFirstPacketSize() {

	if (TXqueue.head != NULL) return TXqueue.head->pktLen;

	return 0;

}

#ifdef RTX

PacketContainer* searchPacketInRTX(int connID, int msgSeqNum, int offset) {

	connID = ntohl(connID);

	msgSeqNum = ntohl(msgSeqNum);

	offset = ntohl(offset);

	//fprintf(stderr,"***************Searching for packet... connID: %d msgSeqNum: %d offset: %d\n",connID,ntohl(msgSeqNum),ntohl(offset));

	PacketContainer *tmp = RTXqueue.head;

	PacketContainer *prev = NULL;

	while (tmp != NULL) {

		struct msg_header *msg_h;

		msg_h = (struct msg_header *) tmp->iov[0].iov_base;

		//fprintf(stderr,"^^^^^^^^^Searching^^^^^^^^^^^^ connID: %d msgSeqNum: %d offset: %d\n",connID,ntohl(msgSeqNum),offset);

		//fprintf(stderr,"^^^^^^^^^In queue^^^^^^^^^^^^ connID: %d msgSeqNum: %d offset: %d\n",msg_h->local_con_id,ntohl(msg_h->msg_seq_num),ntohl(msg_h->offset));

		if ((msg_h->local_con_id == connID) && (msg_h->msg_seq_num == msgSeqNum) && (msg_h->offset == offset)) {

		//fprintf(stderr,"*****************Packet found: ConID: %d  MsgseqNum: %d Offset: %d \n", ntohl(connID),ntohl(msgSeqNum),ntohl(offset));

			if (tmp == RTXqueue.head) RTXqueue.head = tmp->next;

			else if (tmp == RTXqueue.tail) { RTXqueue.tail = prev; prev->next == NULL; }

			else prev->next = tmp->next;

			RTXqueue.size -= tmp->pktLen;

			return tmp;

		}

		prev = tmp;

		tmp = tmp->next;

	}

return NULL;

}

int rtxPacketsFromTo(int connID, int msgSeqNum, int offsetFrom, int offsetTo) {

        int offset = offsetFrom;

        //fprintf(stderr,"\t\t\t\t Retransmission request From: %d To: %d\n",offsetFrom, offsetTo);

        while (offset < offsetTo) {

                PacketContainer *packetToRTX = searchPacketInRTX(connID,msgSeqNum,offset);

                if (packetToRTX == NULL) return 1;

                //sending packet

                //fprintf(stderr,"\t\t\t\t\t Retransmitting packet: %d of msg_seq_num %d.\n",offset/1349,msgSeqNum);

                sendPacket(packetToRTX->udpSocket, packetToRTX->iov, 4, packetToRTX->socketaddr);

                sentRTXDataPktCounter++;

                offset += packetToRTX->iov[3].iov_len;

		destroyPacketContainer(packetToRTX);

        }

        return 0;

}

#endif

/*

 *          

 *

 *

 * 

 *	Agnieszka Witczak & Szymon Kuc

 *     

 */

#include <stdio.h>

#include <stdlib.h>

#include <unistd.h>

#ifndef WIN32

#include <netinet/in.h>

#else

#include <winsock2.h>

#endif

typedef struct PktContainer {

	int udpSocket; 

	struct iovec *iov; 

	int iovlen; 

	struct sockaddr_in *socketaddr;

	int pktLen;		//kB

	struct timeval timeStamp;

	struct PktContainer *next;

	unsigned char priority;

} PacketContainer;

typedef struct PktQueue { 

	PacketContainer *head;

	PacketContainer *tail;

	int size;		//kB

} PacketQueue;

PacketContainer* createPacketContainer (const int uSoc,struct iovec *ioVector,int iovlen,struct sockaddr_in *sockAddress, unsigned char prior);

int addPacketTXqueue(PacketContainer *packet);

PacketContainer* takePacketToSend();

int removeOldestPacket() ;

int isQueueEmpty();

int getFirstPacketSize();

void setQueuesParams (int TXsize, int RTXsize, double maxTimeToHold); //in  bytes, bytes, seconds

#ifdef RTX

void addPacketRTXqueue(PacketContainer *packet);

int rtxPacketsFromTo(int connID, int msgSeqNum, int offsetFrom, int offsetTo);

#endif

/*

 *          

 *

 *	upgraded rateControl - token bucket

 * 

 *	Agnieszka Witczak & Szymon Kuc

 *     

 */

#include <ml_all.h>

extern struct event_base *base;

static long bucket_size = 0;

static int64_t drain_rate = 0;

static long bytes_in_bucket = 0;

struct timeval bib_then = { 0, 0};

void planFreeSpaceInBucketEvent();

void freeSpaceInBucket_cb (int fd, short event,void *arg) {

	/*struct timeval test;

	gettimeofday(&test,NULL);

	int us;

	if(test.tv_usec > (int) arg)

		us = test.tv_usec - (int) arg;

	else

		us = 1000000 + test.tv_usec - (int) arg;

	fprintf(stderr,"Event scheduled in: %d microseconds\n",us);*/

	while((!isQueueEmpty()) && (outputRateControl(getFirstPacketSize()) == OK)) {	

		PacketContainer* packet = takePacketToSend();

		if (packet == NULL) return;

		struct timeval now;

   		gettimeofday(&now, NULL);

		bib_then = now;

#ifdef RTX

		if (!(packet->priority & NO_RTX)) addPacketRTXqueue(packet);

#endif

		sendPacket(packet->udpSocket, packet->iov, 4, packet->socketaddr);

	}

	if (!isQueueEmpty()) planFreeSpaceInBucketEvent(getFirstPacketSize());

	return;

}

void planFreeSpaceInBucketEvent(int bytes) {		//plan the event for time when there will be free space in bucket (for the first packet from the TXqueue)

	struct timeval TXtime;

	struct event *ev;

	//time needed to send data = firstPacketFromQueue.size, will free space for this packet in the bucket

	TXtime.tv_sec = bytes / drain_rate; //seconds

	TXtime.tv_usec = (bytes - TXtime.tv_sec * drain_rate) * 1000000 / drain_rate; //microseconds

	ev = evtimer_new(base, freeSpaceInBucket_cb, NULL);

	event_add(ev, &TXtime);

}

int queueOrSendPacket(const int udpSocket, struct iovec *iov, int len, struct sockaddr_in *socketaddr, unsigned char priority)

{

	PacketContainer *newPacket = (PacketContainer*) createPacketContainer(udpSocket,iov,len,socketaddr,priority);

debug("QUUEO1");

debug("QUUEO1");

	if(!(priority & HP)) {

		if (!isQueueEmpty()) {						//some packets are already waiting, "I am for sure after them"

			return addPacketTXqueue(newPacket);

		}	

		else if(outputRateControl(newPacket->pktLen) != OK) {			//queue is empty, not enough space in bucket - "I will be first in the queue"

			planFreeSpaceInBucketEvent(newPacket->pktLen);		//when there will be enough space in the bucket for the first packet from the queue

			return addPacketTXqueue(newPacket);

		}

	}

debug("QUUEO1c\n");

#ifdef RTX

	if (!(priority & NO_RTX)) addPacketRTXqueue(newPacket);

#endif

debug("QUUEO2");

	return sendPacket(udpSocket, iov, 4, socketaddr);

}

void setOutputRateParams(int bucketsize, int drainrate) { //given in Bytes and Bits/s

     bucket_size = bucketsize;

     outputRateControl(0);

     drain_rate = drainrate >> 3; //now in bytes/s

}

int outputRateControl(int len) {

	struct timeval now;

	gettimeofday(&now, NULL);

	if(drain_rate <= 0) {

		bytes_in_bucket = 0;

		bib_then = now;

		return OK;

	} else {

		unsigned int leaked;

		int total_drain_secs = bytes_in_bucket / (drain_rate) + 1;

		if(now.tv_sec - bib_then.tv_sec - 1 > total_drain_secs) {

				bytes_in_bucket = 0;

		} else {

			leaked = drain_rate * (now.tv_sec - bib_then.tv_sec);

			leaked += drain_rate * (now.tv_usec - bib_then.tv_usec) / 1000000; 

			if(leaked > bytes_in_bucket) bytes_in_bucket = 0;

					else bytes_in_bucket -= leaked;

		}

		bib_then = now;

		if(bytes_in_bucket + len <= bucket_size) {

			bytes_in_bucket += len;

			return OK;

		} else {

			return THROTTLE;

		}

   }

}

/*

 *          

 *

 *	upgraded rateControl - token bucket

 * 

 *	Agnieszka Witczak & Szymon Kuc

 *     

 */

//#include <sys/time.h>

#include <stdio.h>

#include <stdlib.h>

#include <event2/event.h>

#include <errno.h>

#define HP 1

#define NO_RTX 2

void planFreeSpaceInBucketEvent();

void freeSpaceInBucket_cb (int fd, short event,void *arg);

int queueOrSendPacket(const int udpSocket, struct iovec *iov, int len, struct sockaddr_in *socketaddr, unsigned char priority);

#ifndef WIN32

#include <sys/ioctl.h>

#include <sys/socket.h>