msg.flowId = flowId;
					if (sendPipeMsg(flows[flowId].parserPipe, &msg) < 0) {
						perror("signalManager sending msg");
						exitThread();
					}
					exitThread();
				} else {
					
					msg.code = MSG_FP_END;
					msg.flowId = flowId;
					if (sendPipeMsg(flows[flowId].parserPipe, &msg) < 0) {
						perror("signalManager sending msg");
						exitThread();
					}
				}
				break;
			case TSP_ACK_SEND_FLOW:
				
				recv(sock, (char *) buffer, sizeof(buffer), 0);
				flowId = *(unsigned int *) buffer;
				
				CREATE_THREAD(&(flows[flowId]), flowSender, NULL, flows[flowId].handle);
#ifdef WIN32
				if (setPriority == true){
					if (SetThreadPriority(flows[flowId].handle, THREAD_PRIORITY_TIME_CRITICAL) == 0)
					printf("Error - Impossible set priority for thread - %d \n", GetLastError());
 				}
#endif
				break;
			case TSP_DISCOVERY:
				sendType(sock, TSP_ACK_DISCOVERY);
				break;
			case TSP_ERR_MSG_2:
				recv(sock, (char *) buffer, sizeof(buffer), 0);
				flowId = *(unsigned int *) buffer;
				printf("Error - Busy port on receiver side\n");
				
				if ( isChannelClosable(chanId))  {
					
					msg.code = MSG_FP_ERR1;
					msg.flowId = flowId;
					if ( sendPipeMsg(flows[flowId].parserPipe, &msg) < 0) {
						perror("signalManager sending msg");
						exitThread();
					}
			 		exitThread();
				} else {
					msg.code = MSG_FP_ERR1;
					msg.flowId = flowId;
					if ( sendPipeMsg(flows[flowId].parserPipe, &msg) < 0) {
						perror("signalManager sending msg");
						exitThread();
				 	}
				}
				break;
			case TSP_ERR_MSG_3:
				recv(sock, (char *) buffer, sizeof(buffer), 0);
				flowId = *(unsigned int *) buffer;
				printf("** To generate ICMP traffic sender and receiver must be root\n");
				
				if (isChannelClosable(chanId))  {
					
					msg.code = MSG_FP_ERR1;
				 	msg.flowId = flowId;
				 	if (sendPipeMsg(flows[flowId].parserPipe, &msg) < 0) {
						perror("signalManager sending msg");
						exitThread();
				 	}
			 		exitThread();
				}else {
					
				 	msg.code = MSG_FP_ERR1;
				 	msg.flowId = flowId;
				 	if (sendPipeMsg(flows[flowId].parserPipe, &msg) < 0) {
						perror("signalManager sending msg");
						exitThread();
				 	}
				}
    				break;
			default:
				printf("Error - Got unknown message from receiver: %d\n",(int) type);
				break;
			} 
#ifdef WIN32
			ioctlsocket(sock, FIONREAD, &pending);
#endif
		} 

#ifdef LINUX_OS
		
		if (FD_ISSET(fd, &rset)) {
#endif
#ifdef WIN32
		
		if (PeekNamedPipe(namedPipe, NULL , 0 , NULL , &available , NULL) == 0) {
			printf("Error in peek named pipe\n");
			exitThread();
		}
#ifdef DEBUG
		printf("available: %d\n", available);
#endif

		while(available > 0) {
#endif
			
			if (recvPipeMsg(signalChannels[chanId].pipe, &msg) < 0) {
				perror("signalManager receiving msg");
				exitThread();
			}
#ifdef DEBUG
			printf("signal manager received msg code: %d\n", msg.code);
#endif
			switch(msg.code) {
			case MSG_SM_NEWFLOW:
				flows[msg.flowId].sigChanId = chanId;
			       	requestToSend(msg.flowId, sock);
				break;
			case MSG_SM_ENDFLOW:
				
#ifdef DEBUG
				printf("send request to close flow %d\n", msg.flowId);
#endif
				closedFlow(msg.flowId, sock);
				break;
			case MSG_SM_ERRFLOW:
				
				msg.code = MSG_FP_ERR2;
				if (sendPipeMsg(flows[msg.flowId].parserPipe, &msg) < 0) {
					perror("signalManager sending msg");
					exitThread();
				}
				closedFlowErr(msg.flowId, sock);
				
				if ( isChannelClosable(chanId))
					exitThread();
				break;
			default:
				printf("Got unknown message from queue\n");
				break;
			}
#ifdef WIN32
			
			if (PeekNamedPipe(fd, NULL , 0 , NULL , &available , NULL) == 0) {
				printf("Error in peek named pipe\n");
				exitThread();
			}
#endif
		} 

	} 
}


int isChannelClosable(int id)
{
	MUTEX_THREAD_LOCK(mutex);
	signalChannels[id].flows--;
	if (signalChannels[id].flows == 0) {
		sendRelease(signalChannels[id].socket);

		if (closeSock(signalChannels[id].socket) < 0)
			printf("error closing socket\n");
		signalChannels[id].socket = -1;
		signalChannels[id].errorLog = false;

		closePipe(signalChannels[id].pipe);

  		MUTEX_THREAD_UNLOCK(mutex);
#ifdef DEBUG
		printf("closing signalling channel\n");
#endif
		return(1);
	}
	MUTEX_THREAD_UNLOCK(mutex);
	return(0);
}


int identifySignalManager(int flowId, int *chanId, struct addrinfo *DestHost)
{
	int signalSock;
	char type =0;
	MUTEX_THREAD_LOCK(mutex);

	
	if (checkDestHostIP(chanId, DestHost) == false) {
		
		createTransportChan(signalSock, DestHost);	
		createSignalChan(signalSock);			

		
		if ((namelogReceiver == 1) && (logServer == 0)){
     		if (sendNameLog(signalSock, logFileReceiver,sizeof(logFileReceiver))<0) {
				printf("sendNameLog() error \n");
				return(-1);
			}
			recv(signalSock, (char *) &type, sizeof(type), 0);
		}
		if (logServer == 1){
			if (sendLog(signalSock, serverLogReceiver, protoTx_ServerLogReceiver, logFileReceiver)<0){
				printf("sendLog() error \n");
				return(-1);
			}
			recv(signalSock, (char  *) &type, sizeof(type), 0);
		}

		
		
		*chanId = 0;
		while (signalChannels[*chanId].socket != -1)
			(*chanId)++;
		
		if (createNewPipe(signalChannels[*chanId].pipe) < 0) {
			printf("signalManager() could not open pipe for flowId %d\n", flowId);
			return(-1);
		}
		
		signalChannels[*chanId].DestAddr = DestHost;
		signalChannels[*chanId].flows = 1;
		signalChannels[*chanId].socket = signalSock;

		if ((logServer ==1 ) || (namelogReceiver ==1)){
			if (type == TSP_ERR_MSG_4){
				printf("signalManager() TSP_ERR_MSG_4\n");
				signalChannels[*chanId].errorLog = true;
			}
		}
		if (signalChannels[*chanId].errorLog == false){
			
			CREATE_THREAD(chanId, signalManager, NULL, signalChannels[*chanId].handle);

			
			signalChanCount++;
		}
		MUTEX_THREAD_UNLOCK(mutex);
		return(1);

	} else {
		MUTEX_THREAD_UNLOCK(mutex);
		return(0);
	}
}







void *flowSender(void *para)
{
	flowDescriptor *param = (flowDescriptor *)para;
	int id = param->id;
   	struct addrinfo *SrcHost =	flows[id].SrcHost;
	struct addrinfo *DestHost =	flows[id].DestHost;
	BYTE meter =	    		flows[id].meter;
	BYTE l4Proto = 		    	flows[id].l4Proto;
	BYTE l7Proto = 		    	flows[id].l7Proto;
	int icmptype = 			flows[id].icmptype;
	int DSByte =			flows[id].DSByte;
	unsigned long Duration = 	flows[id].Duration;
	int TTL = 			flows[id].TTL;
	SumRandom * IntArriv = 		flows[id].IntArriv;
	SumRandom * PktSize = 		flows[id].PktSize;
	int sigChanId =            	flows[id].sigChanId;
	bool Nagle =                	flows[id].Nagle;
	struct info *infos = 		(struct info *) malloc(DIM * sizeof(info));
	bool SrcPortSpecify =       	flows[id].srcPortSpecify;
	unsigned char payload[MAX_PAYLOAD_SIZE];
	unsigned char *ptrSeqNum = payload + sizeof(unsigned int);
	unsigned char *ptrTimeSec = ptrSeqNum + sizeof(unsigned int);
	unsigned char *ptrTimeUsec = ptrTimeSec + sizeof(long int);
	unsigned char *ptrSize = NULL;

	fd_set read_set, active_set;
	int size, size_r = 0;
	long int time, time1, time2;
	struct TTicker Ticker;
	struct timeval start_time, end_time, RcvTime, timeout;

	int socktype;
	int prototype = 0;
	unsigned int seqNum = 1;
	Real wait;
	int count = 0;
	int sockchk = 0;
	struct pipeMsg msg;

	char HelpSrcAddress[INET6_ADDRSTRLEN];
	char HelpDstAddress[INET6_ADDRSTRLEN];
	int tmpPort_SrcPort = 0;
	int tmpPort_DstPort = 0;
#ifdef WIN32
	LARGE_INTEGER _tstart, _tend, _tprec;
	unsigned long secondi = 0, microsecondi = 0;
	int first = 1;
	int first_update = 1;
#endif


#ifdef DEBUG
	printf("flowSender() started\n");
#endif

	
	UniformRV = new Uniform;
	UniformRV->Set(0.5);
	for (int i=0; i<MAX_PAYLOAD_SIZE;i++)
		payload[i]=(unsigned char)(255*UniformRV->Next());

	
	msg.flowId = id;

 	
 	if (SrcPortSpecify == true)
 	    GET_PORT(flows[id].SrcHost, tmpPort_SrcPort);

  	
	SrcHost->ai_family = DestHost->ai_family;
	SrcHost->ai_addrlen = DestHost->ai_addrlen;
	SrcHost->ai_addr = (struct sockaddr *) malloc(SrcHost->ai_addrlen);

	
	switch (l4Proto) {
	case L4_PROTO_UDP:
		socktype = SOCK_DGRAM;
		break;
	case L4_PROTO_ICMP:
		socktype = SOCK_RAW;
		if (meter == METER_OWDM)
			prototype = (DestHost->ai_family == AF_INET) ? IPPROTO_ICMP : IPPROTO_ICMPV6;
		else if (meter == METER_RTTM)
			prototype = IPPROTO_RAW;
		break;
	default:
		socktype = SOCK_STREAM;
		break;
	}

    	
	int sock = socket(DestHost->ai_family, socktype, prototype);
	if (sock < 0) {
		perror("flowSender");
		printf("Could not create a new socket. Flow ID: %d\n", id);
		msg.code = MSG_SM_ERRFLOW;
		
		if (sendPipeMsg(signalChannels[sigChanId].pipe, &msg) < 0) {
			perror("flowSender sending msg");
			exitThread();
		}
		exitThread();
	}

	
#ifdef LINUX_OS
	int reuse = 1;
	int optlen = sizeof(reuse);
	if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char*)&reuse, optlen)<0){
		printf("flowSender sending msg");
	}
#endif

	
	if (SrcPortSpecify == true){
#ifdef DEBUG
	printf("Set Port on Sender Side \n");
#endif

#ifdef LINUX_OS
	
	if ( getaddrinfo("::", NULL, NULL, &SrcHost) < 0 )  {
		
		if (SrcHost) freeaddrinfo(SrcHost);
		if ( getaddrinfo("0.0.0.0", NULL, NULL, &SrcHost) <0)
		perror("flowSender");
		printf("Error into getaddrinfo Flow ID: %d\n", id);
		msg.code = MSG_SM_ERRFLOW;
		
		if (sendPipeMsg(signalChannels[sigChanId].pipe, &msg) < 0) {
			perror("flowSender sending msg");
			exitThread();
		}
		exitThread();
	}
#endif

#ifdef WIN32
#ifdef IPv6RECV
      if (getaddrinfo("::", NULL, NULL, &SrcHost) <0){
#else
      if (getaddrinfo("0.0.0.0", NULL, NULL, &SrcHost)<0){
#endif
	perror("flowSender");
	printf("Error into getaddrinfo Flow ID: %d\n", id);
	msg.code = MSG_SM_ERRFLOW;
	
	if (sendPipeMsg(signalChannels[sigChanId].pipe, &msg) < 0) {
		perror("flowSender sending msg");
		exitThread();
	}
	exitThread();
	}
#endif

    SET_PORT(SrcHost, tmpPort_SrcPort);
    if (bind(sock, SrcHost->ai_addr, SrcHost->ai_addrlen)!=0){
	        perror("flowSender");
		printf("Could not bind a new socket. Flow ID: %d\n", id);
		msg.code = MSG_SM_ERRFLOW;
		
		if (sendPipeMsg(signalChannels[sigChanId].pipe, &msg) < 0) {
			perror("flowSender sending msg");
			exitThread();
		}
		exitThread();
	}
	} 


	
	if ((!(l7Proto == L7_PROTO_TELNET) && (!(l7Proto == L7_PROTO_DNS))) && (socktype == SOCK_STREAM)
        && (Nagle == false)) {
		printf("Nagle algorithm disabled\n");
#ifdef LINUX_OS
		int no_delay = 1;
		if (setsockopt(sock, getprotobyname("TCP")->p_proto, TCP_NODELAY, &no_delay,
			sizeof(no_delay)) < 0)
#endif
#ifdef WIN32
		char *no_delayWin;
		no_delayWin = "1";
		if (setsockopt(sock, getprotobyname("TCP")->p_proto, TCP_NODELAY, no_delayWin,
			sizeof(no_delayWin)) < 0)
#endif
			printf("** WARNING ** Flow %d. Cannot disable Nagle Algorithm\n", id);
	}