itgrecv.cpp

D-ITG2.4源代码

	
	}

	
	if (type != TSP_SENDER_DOWN)
		
		if ( sendAck(newSockSignaling, TSP_ACK_RELEASE) < 0)
			reportErrorAndExit("signalManager","sendAck",
				"Cannot send ack release");
#ifdef DEBUG
	printf("Signal Manager : sent TSP_ACK_RELEASE(7) message\n");
#endif

	if (logRemote == 2)  {
		 
		signaling signalingLog;
		signalingLog.stop = true;


		if (logHost.ai_family == PF_INET)
			((struct sockaddr_in*)(logHost.ai_addr))->sin_port=htons(DEFAULT_LOG_PORT_SIGNALING);
		else if (logHost.ai_family == PF_INET6)
			((struct sockaddr_in6*)(logHost.ai_addr))->sin6_port=htons(DEFAULT_LOG_PORT_SIGNALING);

		if ( sendto(logSockSignaling, (char *) &signalingLog, sizeof(signalingLog), 0,
			logHost.ai_addr, logHost.ai_addrlen) < 0)
			reportErrorAndExit("typeParser","send stop to LOG_SERVER",
			"Cannot send stop logServer");
#ifdef DEBUG
    		printf("Sent infos to LogServer\n");
#endif

		if ( closeSock(logSock) < 0)  {
				reportErrorAndExit("signalManager","closeSock",
						"Cannot close socket logSock after sending ack release");
					
		}
		if ( closeSock(logSockSignaling) < 0)  {
					reportErrorAndExit("signalManager","closeSock",
						"Cannot close socket logSockSignaling after sending ack release");
				
		}
	}


	
	
	if ( closeSock(newSockSignaling) < 0)
			reportErrorAndExit("signalManager","closeSock",
				"Cannot close socket newSockSignaling after sending ack release");
	
	if (closePipe(rPipe) < 0)
			reportErrorAndExit("signalManager","closePipe",
				"Cannot close pipe ");


#ifdef DEBUG
	printf("Closed Signaling Channel\n");
#endif
	
	exitThread();
	return 0;
}




void printHelpAndExit()
{
	printf(" Name\n");
	printf("     ITGRecv - Receiver component of D-ITG platform\n\n");
	printf(" Synopsis\n");
#ifdef LINUX_OS
	printf("	 ./ITGRecv   [-l <logfile>] \n");
#endif
#ifdef WIN32
	printf("     ITGRecv.exe [-P] [[-l <logfile>]\n");
#endif
	printf("                 [-L [<log_server_addr>] [<protocol_type>]]\n");
	printf(" Options\n");
#ifdef WIN32
	printf("     -P                                   Enable thread high priority\n");
#endif
	printf("     -l <logfile>                         Generate the log file\n");
	printf("                                          (Default : <%s> <%s>)\n",DEFAULT_LOG_IP,invFindL4Proto(DEFAULT_PROTOCOL_TX_LOG));
	printf("     -L <log_server_addr> <protocol_type> Remote log file\n");
	printf("                                          (Default : <%s> <%s>)\n",DEFAULT_LOG_IP,invFindL4Proto(DEFAULT_PROTOCOL_TX_LOG));
	printf(" \n For more information consult help file in the same directory\n");
	exit(1);
}



void recvInit()
{
#ifdef WIN32
	if (InitializeWinsock(MAKEWORD(1,1)) != 1)	{
		flagTerm = ERROR_TERMINATE;
		reportErrorAndExit("recvInit","InitializeWinsock","Cannot initialize WinSocket");
	}
#endif
	
	userId = USER_ID();
	
	if ( MUTEX_THREAD_INIT(mutexLog) < 0)
		reportErrorAndExit("recvInit","CreateMutex","Cannot initialize Mutex");
	
	strcpy(logFile, DEFAULT_LOG_FILE);
	
	l4ProtoLog = DEFAULT_PROTOCOL_TX_LOG;
	
	if ( getaddrinfo(DEFAULT_LOG_IP, NULL, NULL, & globaleLogHost) != 0)
		reportErrorAndExit("recvInit","getaddrinfo","Cannot set the default IP log address");
	
	SET_PORT(globaleLogHost, htons(DEFAULT_LOG_PORT));
#ifdef WIN32
	
	LARGE_INTEGER temp;
	if ( QueryPerformanceFrequency(&temp) == 0)
		reportErrorAndExit("recvInit","QueryPerformaceFrequency",
			"Cannot use system's high performace counter");
	if ( QueryPerformanceCounter(&temp) == 0)
		reportErrorAndExit("recvInit","QueryPerformaceCounter",
			"Cannot use system's high performace counter");
#ifdef DEBUG
	printf("High performance system counter PRESENT\n");
#endif
#endif
}



void createSignalingLogChannel(struct addrinfo logHost, char logFile[DIM_LOG_FILE], BYTE protocolLog, int & logSockSignaling)
{
	
	signaling signal;
	
	char buffer[1];

	
	logSockSignaling = socket(logHost.ai_family, SOCK_STREAM, 0);
	if (logSockSignaling < 0)
		reportErrorAndExit("createSignalingLogChannel","socket","Cannot create socket logSockSignaling");
	
	if (logHost.ai_family == PF_INET)
		((struct sockaddr_in*)(logHost.ai_addr))->sin_port=htons(DEFAULT_LOG_PORT_SIGNALING);
	else if (logHost.ai_family == PF_INET6)
		((struct sockaddr_in6*)(logHost.ai_addr))->sin6_port=htons(DEFAULT_LOG_PORT_SIGNALING);

	
	signal.protocol = protocolLog;
	
	strcpy(signal.logFile, logFile);
	
	if (connect(logSockSignaling, logHost.ai_addr, logHost.ai_addrlen) < 0)
		reportErrorAndExit("createSignalingLogChannel","connect","Error into connect - Start first LogServer!");
	
	do {
		if (recv(logSockSignaling, (char *) &buffer, sizeof(BYTE), 0) < 0)
		reportErrorAndExit("createSignalingLogChannel","recv","Cannot receive logSockSignaling1 data");
	
  	} while (*(BYTE *) buffer != LOG_CONNECT);
#ifdef DEBUG
	printf("Log Signal Manager : Received LOG_CONNECT Message\n");
#endif
	
	if (send(logSockSignaling, (char *) &signal, sizeof(signal), 0) < 0)
		reportErrorAndExit("createSignalingLogChannel","send","Cannot send infos to LogServer");
#ifdef DEBUG
	printf("Log Signal Manager : Sent to LogServer infos for creating signaling channel\n");
#endif
}



void createRemoteLogFile(struct addrinfo &logHost, char logFile[DIM_LOG_FILE], BYTE protocolLog, int & logSockSignaling, int & logSock)
{
	
	int sizeMsg = sizeof(int);
	
	char buffer[sizeMsg];

	

	createSignalingLogChannel(logHost, logFile, protocolLog, logSockSignaling);

	

	switch(protocolLog) {
		case L4_PROTO_UDP :
            
			logSock = socket(logHost.ai_family, SOCK_DGRAM, 0);
			if (logSock < 0)
				reportErrorAndExit("createRemoteLogFile","socket DATAGRAM","Cannot create socket logSock");
			break;
		case L4_PROTO_TCP :
            
			logSock = socket(logHost.ai_family, SOCK_STREAM, 0);
			if (logSock < 0)
				reportErrorAndExit("createRemoteLogFile","socket STREAM","Cannot create socket logSock");
			break;
		default :
			break;
	}
	
	if ( recv(logSockSignaling, (char *) buffer, sizeof(buffer), 0) < 0)
		reportErrorAndExit("createRemoteLogFile","recv","Cannot receive logSockSignaling2 data");
#ifdef DEBUG
	printf("Port received for log : %d\n",*(unsigned int *) buffer);
#endif
	
       
	if (logHost.ai_family == PF_INET)
		((struct sockaddr_in*)(logHost.ai_addr))->sin_port=htons(*(unsigned int *) buffer);
	else if (logHost.ai_family == PF_INET6)
		((struct sockaddr_in6*)(logHost.ai_addr))->sin6_port=htons(*(unsigned int *) buffer);

	
	if (connect(logSock, logHost.ai_addr, logHost.ai_addrlen) < 0)
		reportErrorAndExit("createRemoteLogFile","connect","Error into connect with logsender!");
}




void recvFlowLog(int newSockSignaling, struct addrinfo & logHost, BYTE &protocolLog, char logFile[])
{
	

	
	int sizeMsg = 2 * sizeof(int) + sizeof(in_addr_t) + sizeof(in_port_t) + sizeof(BYTE)+DIM_LOG_FILE;
	
	char buffer[sizeMsg];
	
	char *ptrIPVer = buffer;
	
	char *ptrIP = ptrIPVer + sizeof(int);
	
	char *ptrPort = ptrIP + sizeof(in_addr_t);
	
	char *logProtocol= ptrPort + sizeof(in_port_t);
	
	char *fileName= logProtocol + sizeof(BYTE);


	
	

	
	if ( recv(newSockSignaling, (char *) buffer, sizeof(buffer), 0) < 0)
		reportErrorAndExit("recvFlowLog","recv - type = 12",
			"Cannot receive data on newSockSignaling");

	
	logHost.ai_family = *((int *) ptrIPVer);
	
	if (*((int *) ptrIPVer) == PF_INET)
		ptrPort = ptrIP + sizeof(in_addr_t);
	else if (*((int *) ptrIPVer) == PF_INET6)
		ptrPort = ptrIP + 4 * sizeof(in_addr_t);
	
	logProtocol = ptrPort + sizeof(in_port_t);
	
	fileName = logProtocol + sizeof(BYTE);
	
	protocolLog = *(BYTE *) logProtocol;

	
	if (logHost.ai_family == PF_INET) {
		
		logHost.ai_addrlen = sizeof(struct sockaddr_in);
		
		logHost.ai_addr =
		  (struct sockaddr *) malloc(sizeof(struct sockaddr_in));
		((struct sockaddr_in *) logHost.ai_addr)->sin_family = AF_INET;
		((struct sockaddr_in *) logHost.ai_addr)->sin_addr.s_addr =
		    *(in_addr_t *) ptrIP;
		
		((struct sockaddr_in *) logHost.ai_addr)->sin_port =
		    *(in_port_t *) ptrPort;
	
 	} else if (logHost.ai_family == PF_INET6) {
		
		logHost.ai_addrlen = sizeof(struct sockaddr_in6);
		
		logHost.ai_addr =
		    (struct sockaddr *) malloc(sizeof(struct sockaddr_in6));
		((struct sockaddr_in6 *) logHost.ai_addr)->sin6_family = AF_INET6;
		memcpy(((struct sockaddr_in6 *) logHost.ai_addr)->sin6_addr.
		    s6_addr32, ptrIP, 4 * sizeof(in_addr_t));
		
		((struct sockaddr_in6 *) logHost.ai_addr)->sin6_port =
		    *(in_port_t *) ptrPort;
	}
	
	strcpy(logFile, fileName);
}





void recvNameLog(char nameFile[DIM_LOG_FILE],int newSockSignaling)
{
	

	
	int sizeMsg = DIM_LOG_FILE;
	
	char buffer[sizeMsg];

	int size_r = 0;

	
	if ((size_r = recv(newSockSignaling, (char *) buffer, sizeof(buffer), 0)) < 0)
		reportErrorAndExit("sendName","recv - type = 12",
				"Cannot receive data on newSockSignaling");
	strncpy(nameFile, buffer, size_r);
}





#ifdef LINUX_OS
void pipeParser(int newSockSignaling, int &numFlow, int rPipe[], memChannel flowIdNum[],
		   paramThread paraThread[], pthread_t hThr[])
#endif
#ifdef WIN32
void pipeParser(int newSockSignaling, int &numFlow, HANDLE rPipe[], memChannel flowIdNum[],
		   paramThread paraThread[], pthread_t hThr[])
#endif
{
				struct pipeMsg msg;
				
				if (recvPipeMsg(rPipe, &msg) < 0) {
					reportErrorAndExit("pipeParser","",
					"Cannot receive message into pipe");
				}
				switch (msg.code) {
				case MSG_FT_ERR1:
#ifdef DEBUG
					printf("Error into bind - flow %d \n",msg.flowId);
#endif
					
					if (sendAckFlow(newSockSignaling, TSP_ERR_MSG_2, msg.flowId) < 0)
						reportErrorAndExit("pipeParser","",
						"Cannot send TSP_ERR_MSG_2 message to sender");
					int k;
					k=0;
					
					while (msg.flowId != flowIdNum[k].flowId) {
						k++;
					}

					
					if (terminateThread(hThr[k]) < 0)
						reportErrorAndExit("pipeParser","",
						"Cannot terminate thread");

					
					free(paraThread[k].addressInfos);
					
					flowIdNum[k].flowId = -1;
					
					numFlow--;
				break;
				case MSG_FT_ERR2:
#ifdef DEBUG
					printf("Error: you must be root to generate ICMP traffic - %d \n",msg.flowId);
#endif
					
					if (sendAckFlow(newSockSignaling, TSP_ERR_MSG_3, msg.flowId) < 0)
						reportErrorAndExit("pipeParser","",
						"Cannot send TSP_ERR_MSG_3 message to sender");
					int j;
					j=0;
					
					while (msg.flowId != flowIdNum[j].flowId) {
						j++;
					}

					
					if (terminateThread(hThr[j]) < 0)
						reportErrorAndExit("pipeParser","",
						"Cannot terminate thread");

					
					free(paraThread[j].addressInfos);
					
					flowIdNum[j].flowId = -1;
					
					numFlow--;
				break;
				case MSG_FT_OK:
#ifdef DEBUG
					printf("signal manager received MSG_FT_OK - %d \n",msg.flowId);
#endif					
					if (sendAckFlow(newSockSignaling, TSP_ACK_SEND_FLOW, msg.flowId) == -1)
						reportErrorAndExit("pipeParser","",
						"Cannot send TSP_ACK_SEND_FLOW message to sender");

#ifdef DEBUG
					printf("TCP Pkt Receiver : Sent TSP_ACK_SEND_FLOW(5) message\n");
#endif
				break;
				default:
					printf("undefined message received from signal manager\n");
				break;
				}
}



#ifdef WIN32
int typeParser(BYTE type, int &numFlow, int newSockSignaling , memChannel flowIdNum[],
		  paramThread paraThread[], pthread_t hThr[], HANDLE rPipe[], char *fileLog, int logSock, int logSockSignaling,
          struct addrinfo *logHost)
#endif
#ifdef LINUX_OS
int typeParser(BYTE type, int &numFlow, int newSockSignaling , memChannel flowIdNum[],
		   paramThread paraThread[], pthread_t hThr[], int rPipe[],  char *fileLog, int logSock, int logSockSignaling,
          struct addrinfo *logHost)

#endif
{
        if (type == TSP_SEND_FLOW) {
#ifdef DEBUG
			printf("Signal Manager : Received TSP_SEND_FLOW(3) message\n");
#endif
			
			int flowPosition;
			
			for (flowPosition = 0; flowPosition < MAX_NUM_THREAD; flowPosition++) {
				if (flowIdNum[flowPosition].flowId == -1) {
					break;
				}
			}
#ifdef DEBUG
			printf("flowPosition for TSP_SEND_FLOW = %d\n",flowPosition);
#endif
			
			if (flowPosition < MAX_NUM_THREAD) {
				
				 int sizeMsg6 = 4 * sizeof(in_addr_t);
				 int sizeMsg4 = sizeof(in_addr_t);
				 int sizeMsg = 3 * sizeof(type) + 2* sizeof(int) + sizeof(in_port_t)+ DIM_NAME_SERIAL_INTERFACE;

				
				char buffer[sizeMsg + sizeMsg6];

        	        	
				char *ptrProtocol = buffer;
				
				char *ptrUpProtocol = ptrProtocol + sizeof(type);
				
				char *ptrMeter = ptrUpProtocol + sizeof(type);
				
				char *ptrIPVer = ptrMeter + sizeof(type);
				
				char *ptrFlowId = ptrIPVer + sizeof(int);

				
				char *ptrSerialReceiver = ptrFlowId + sizeof(int);

				
				char *ptrPort = ptrSerialReceiver + DIM_NAME_SERIAL_INTERFACE;

				
				char *ptrIP = ptrPort + sizeof(in_port_t);


				
				if (recv(newSockSignaling, (char *) buffer, sizeMsg, 0) < 0)
					reportErrorAndExit("typeParser","recv - type = TSP_SEND_FLOW(3",
					"Cannot receive data on newSockSignaling");

				if (*((int *) ptrIPVer) == PF_INET){
				    
					if (recv(newSockSignaling,  ptrIP, sizeMsg4, 0) < 0)
					reportErrorAndExit("typeParser","recv - type = TSP_SEND_FLOW(3",
					"Cannot receive data on newSockSignaling");
				}
				else if (*((int *) ptrIPVer) == PF_INET6) {
					if (recv(newSockSignaling, ptrIP, sizeMsg6, 0) < 0)
					reportErrorAndExit("typeParser","recv - type = TSP_SEND_FLOW(3",
					"Cannot receive data on newSockSignaling");
				}
#ifdef DEBUG
				printf("Buffer with TYPE = TSP_SEND_FLOW(3) received\n");
#endif

				paraThread[flowPosition].destHost.ai_family = *((int *) ptrIPVer);

				
				if (paraThread[flowPosition].destHost.ai_family == PF_INET) {
					
					paraThread[flowPosition].destHost.ai_addrlen =
					    sizeof(struct sockaddr_in);
					
					paraThread[flowPosition].destHost.ai_addr =
					    (struct sockaddr *) malloc(sizeof(struct sockaddr_in));
					
					((struct sockaddr_in *) paraThread[flowPosition].destHost.
					    ai_addr)->sin_family = AF_INET;
					
					((struct sockaddr_in *) paraThread[flowPosition].destHost.
					    ai_addr)->sin_addr.s_addr = *(in_addr_t *) ptrIP;
					
					((struct sockaddr_in *) paraThread[flowPosition].destHost.
					    ai_addr)->sin_port = *(in_port_t *) ptrPort;
					
				} else if (paraThread[flowPosition].destHost.ai_family == PF_INET6) {
				    	
					paraThread[flowPosition].destHost.ai_addrlen =
					    sizeof(struct sockaddr_in6);
					
					paraThread[flowPosition].destHost.ai_addr =
					    (struct sockaddr *) malloc(sizeof(struct sockaddr_in6));
					
					((struct sockaddr_in6 *) paraThread[flowPosition].destHost.
					    ai_addr)->sin6_family = AF_INET6;
					
					memcpy(((struct sockaddr_in6 *) paraThread[flowPosition].
						destHost.ai_addr)->sin6_addr.s6_addr32, ptrIP,
					    4 * sizeof(in_addr_t));
					
					((struct sockaddr_in6 *) paraThread[flowPosition].destHost.
					    ai_addr)->sin6_port = *(in_port_t *) ptrPort;
				}




				paraThread[flowPosition].rPipe[0] = rPipe[0];
				paraThread[flowPosition].rPipe[1] = rPipe[1];

				
				flowIdNum[flowPosition].flowId = *(unsigned int *) ptrFlowId;

				
				flowIdNum[flowPosition].l4Proto = *(unsigned int *) ptrProtocol;
#ifdef DEBUG
				printf("Signal Manager : Level 4 Protocol Received : %s\n",
					invFindL4Proto(flowIdNum[flowPosition].l4Proto));
#endif
				

				
				paraThread[flowPosition].l7Proto = *(unsigned int *) ptrUpProtocol;
#ifdef DEBUG
				printf("Signal Manager : Level 7 Protocol Received : %s\n",
					invFindL7Proto(paraThread[flowPosition].l7Proto));
#endif
				
				paraThread[flowPosition].flowId = *(unsigned int *) ptrFlowId;