itgsend.cpp

D-ITG2.4源代码

				CauchyRV = new Cauchy;
				flows[id].IntArriv = new SumRandom(a * (*CauchyRV) + b);
				h += 3;
				argc -= 3;
				break;
			case 'N':
				if ((argc < 3) || (argv[h + 2] <= 0))	
					ReportErrorAndExit("Protocol Parser",
					    "Invalid pkts per sec", programName, id);
				flows[id].IntArrivDistro = pdNormal;
				b = strtod(argv[h + 1], NULL); 	
				a = strtod(argv[h + 2], NULL); 	
                		delete flows[id].IntArriv;
				NormalRV = new Normal;
				flows[id].IntArriv = new SumRandom(a * (*NormalRV) + b);
				h += 3;
				argc -= 3;
				break;
			case 'O':
				if ((argc < 2) || (strtod(argv[h + 1], NULL) <= 0))	
					ReportErrorAndExit("Protocol Parser",
					    "Invalid pkts per sec", programName, id);
				a = strtod(argv[h + 1], NULL);
				delete flows[id].IntArriv;
				
				PoissonRV = new Poisson(a);
				flows[id].IntArrivDistro = pdPoisson;
				flows[id].IntArriv = new SumRandom(1000.0 / (*PoissonRV));
				h += 2;
				argc -= 2;
				break;
			case 'G':
				if ((argc < 3) || (strtod(argv[h + 1], NULL) <= 0) || (strtod(argv[h + 2], NULL) <= 0))	
					ReportErrorAndExit("Protocol Parser",
					    "Invalid Gamma Distribution parameter values",
					    programName, id);
				a = (Real) strtod(argv[h + 1], NULL);
				b = (Real) strtod(argv[h + 2], NULL);
    				delete flows[id].IntArriv;
				
				GammaRV = new Gamma(a);
				flows[id].IntArrivDistro = pdGamma;
				flows[id].IntArriv = new SumRandom(b * (*GammaRV));
				h += 3;
				argc -= 3;
				break;
			case 'c':
				if ((argc < 2) || (atoi(argv[h + 1]) < 1))
					ReportErrorAndExit("Protocol Parser", "Invalid pkt size",
					    programName, id);
				delete flows[id].PktSize;
				flows[id].PktSizeDistro = pdConstant;
				ConstantRV = new Constant(1);
				flows[id].PktSize = new SumRandom(atoi(argv[h + 1]) * (*ConstantRV));
				h += 2;
				argc -= 2;
				break;
			case 'u':
				if ((argc < 3) || (atoi(argv[h + 1]) < 1)
				    || (atoi(argv[h + 2]) <= atoi(argv[h + 1])))
					ReportErrorAndExit("Protocol Parser", "Invalid pkt size",
					    programName, id);
				delete flows[id].PktSize;
				flows[id].PktSizeDistro = pdUniform;
				b = atoi(argv[h + 1]);
				a = atoi(argv[h + 2]) - b;
				UniformRV = new Uniform;
				flows[id].PktSize = new SumRandom(a * (*UniformRV) + b);
				argc -= 3;
				h += 3;
				break;
			case 'e':
				if ((argc < 2) || (atoi(argv[h + 1]) < 1))
					ReportErrorAndExit("Protocol Parser", "Invalid pkt size",
					    programName, id);
				delete flows[id].PktSize;
				flows[id].PktSizeDistro = pdExponential;
				ExponentialRV = new Exponential;
				flows[id].PktSize = new SumRandom(atoi(argv[h + 1]) * (*ExponentialRV));
				h += 2;
				argc -= 2;
				break;
			case 'v':
				if ((argc < 3) || (strtod(argv[h + 2], NULL) <= 0))	
					ReportErrorAndExit("Protocol Parser",
					    "Invalid Pareto Distribution parameter values",
					    programName, id);
				a = (Real) strtod(argv[h + 1], NULL);
				b = (Real) strtod(argv[h + 2], NULL);
				delete flows[id].PktSize;
				ParetoRV = new Pareto(a);
				flows[id].PktSizeDistro = pdPareto;
				flows[id].PktSize = new SumRandom(b * (*ParetoRV));
				argc -= 3;
				h += 3;
				break;
			case 'y':
				if ((argc < 3) || (atoi(argv[h + 2]) <= 0))	
					ReportErrorAndExit("Protocol Parser",
					    "Invalid pkts per sec", programName, id);
				delete flows[id].PktSize;
				flows[id].PktSizeDistro = pdCauchy;
				b = (Real) atoi(argv[h + 1]);	
				a = (Real) atoi(argv[h + 2]);	
				CauchyRV = new Cauchy;
				flows[id].PktSize = new SumRandom(a * (*CauchyRV) + b);
				h += 3;
				argc -= 3;
				break;
			case 'n':
				if ((argc < 3) || (argv[h + 2] <= 0))	
					ReportErrorAndExit("Protocol Parser", "Invalid pkt size",
					    programName, id);
				delete flows[id].PktSize;
				flows[id].PktSizeDistro = pdNormal;
				b = (Real) atoi(argv[h + 1]);	
				a = (Real) atoi(argv[h + 2]);	
				NormalRV = new Normal;
				flows[id].PktSize = new SumRandom(a * (*NormalRV) + b);
				h += 3;
				argc -= 3;
				break;
			case 'o':
				if ((argc < 2) || (atoi(argv[h + 1]) < 0))	
					ReportErrorAndExit("Protocol Parser", "Invalid pkt size",
					    programName, id);
				a = atoi(argv[h + 1]);
				delete flows[id].PktSize;
				
				PoissonRV = new Poisson(a);
				flows[id].PktSizeDistro = pdPoisson;
				flows[id].PktSize = new SumRandom(1 * (*PoissonRV));
				h += 2;
				argc -= 2;
				break;
			case 'g':
				if ((argc < 3) || (strtod(argv[h + 2], NULL) <= 0))
					ReportErrorAndExit("Protocol Parser",
					    "Invalid Gamma Distribution parameter values",
					    programName, id);
				delete flows[id].PktSize;
				a = (Real) strtod(argv[h + 1], NULL);
				b = (Real) strtod(argv[h + 2], NULL);
				
				GammaRV = new Gamma(a);
				flows[id].PktSizeDistro = pdGamma;
				flows[id].PktSize = new SumRandom(b * (*GammaRV));
				h += 3;
				argc -= 3;
				break;
			case 'D':
				
				
#ifdef DEBUG
				printf("Nagle algorithm disabled\n");
#endif
				flows[id].Nagle=false;
				h += 1;
				argc -= 1;
				break;
#ifdef WIN32				
			case 'P':
				
				
				if (SetPriorityClass(GetCurrentProcess(), HIGH_PRIORITY_CLASS) == 0){
				        printf("Error - Impossible set priority class - %d \n", GetLastError());
			    }
#ifdef DEBUG
				printf(" Enabled thread priority \n");
#endif
				setPriority = true;
				h += 1;
				argc -= 1;
				break;
#endif
			default:
				char temp[sizeof("What is  ?") + sizeof(argv[h])];
				ReportErrorAndExit("General parser", strcat(strcat(strcpy(temp,
						"What is "), argv[h]), " ?"), programName, id);
				break;
			} 
		} else {
			flows[id].l7Proto = findL7Proto(argv[h]);
			switch (flows[id].l7Proto) {
			case L7_PROTO_TELNET:

				telnetParser(&flows[id].IntArriv, &flows[id].PktSize, flows[id].IntArrivDistro, flows[id].PktSizeDistro);
				flows[id].l4Proto = L4_PROTO_TCP;
				h++;
				argc--;
				break;
			case L7_PROTO_VOIP:

				voIPParser(h, argv, argc, flows[id].id, &flows[id].IntArriv, &flows[id].PktSize,
				    flows[id].IntArrivDistro, flows[id].PktSizeDistro);
				flows[id].l4Proto = L4_PROTO_UDP;
				break;
			case L7_PROTO_DNS:

				dnsParser(&flows[id].IntArriv, &flows[id].PktSize, flows[id].IntArrivDistro, flows[id].PktSizeDistro);
				h++;
				argc--;
				break;
			case LX_PROTO_NONE:
				char temp[sizeof("What is  ?") + sizeof(argv[h])];
				ReportErrorAndExit("General parser", strcat(strcat(strcpy(temp,
						"What is "), argv[h]), " ?"), programName, id);
				break;
			} 
		} 
	} 
#ifdef DEBUG
	printf("Terminate Parser flow %d\n",id);
#endif

#ifdef WIN32
	if ((flows[id].l4Proto == L4_PROTO_ICMP) && (flows[id].DestHost->ai_family == PF_INET6)) {
		printf("Error: traffic ICMP with protocol IPv6 is not supported \n");
		memClean();
	exit(1);
	}
#endif
	
	if (seed == 0.0) {
#ifdef WIN32
		_ftime(&tstruct);
		seed = 0.49 * sin(tstruct.millitm) + 0.50;	
#endif
#ifdef LINUX_OS
		struct timeval tv;	
		gettimeofday(&tv, NULL);
		seed = 0.49 * sin(tv.tv_usec) + 0.50;	
#endif
	}

	
	flows[id].IntArriv->Set(seed);
	flows[id].PktSize->Set(seed);


	
	if (createNewPipe(flows[id].parserPipe) < 0) {
		printf("Error in flowParser() trying to create a new pipe. [flow %d]\n", id);
		exitThread();
	}

	
	
	if (Delay) {
#ifdef LINUX_OS
		timeout.tv_sec = Delay / 1000;
		timeout.tv_usec = (Delay % 1000) * 1000;
		long int result;
		do
			result = (long int) (select(0, NULL, NULL, NULL, &timeout));
		while (result == -1L && errno == EINTR);
#endif
#ifdef WIN32
		Sleep(Delay);
#endif
	}

	
	
	
 	
	int rit = identifySignalManager(id, &chanId, flows[id].DestHost);
	if (rit == -1) {
		printf("Error into function identifySignalManager() n");
		exitThread();
	}

	if ((namelogReceiver == 1) || (logServer == 1)) {
		if (signalChannels[chanId].errorLog == true){
			printf("Error log file specified is already open \n");
  			isChannelClosable(chanId);
			if (multiFlows)
	   			exitThread();
			else return 0;
		}
	}


	
	msg.code = MSG_SM_NEWFLOW;
	msg.flowId = id;
	if (sendPipeMsg(signalChannels[chanId].pipe, &msg) < 0) {
		perror("flowParser sending msg");
		exitThread();
	}

	
	if (recvPipeMsg(flows[id].parserPipe, &msg) < 0) {
		perror("flowParser receiving msg");
		exitThread();
	}

#ifdef DEBUG
	printf("msg received from signal manager %d\n",msg.code);
#endif

	switch (msg.code) {
	case MSG_FP_END:
		break;
	case MSG_FP_ERR1:
		printf("Error at Receiver side \n");
		break;
	case MSG_FP_ERR2:
		printf("Error - FlowSender interrupted by an error\n");
		break;
	default:
		printf("Error undefined message received from signal manager\n");
		break;
	}

	

	if (managerMode) {
		
		int length = strlen(ManagerMsg);
		memmove(ManagerMsg + 2 * sizeof(int), ManagerMsg, length);
		((int *) ManagerMsg)[0] = MNG_FLOWEND;	
		((int *) ManagerMsg)[1] = length;	
		sendto(managerSock, ManagerMsg, length + sizeof(int) * 2, 0, (struct sockaddr *) &ManagerIP,
		    ManagerIPslen);
#ifdef DEBUG
		printf("Notify ITGManager about the end of the generation \n");
#endif

	}

	
   	closePipe(flows[id].parserPipe);


    

	if (multiFlows)
	   exitThread();
    return 0;
}

void *signalManager(void *id)
{
	int	chanId, sock, size;
	BYTE	type;
	unsigned int	flowId;
	char	buffer[4];
	struct pipeMsg msg;
#ifdef LINUX_OS
	int fd, maxfd;
	fd_set	rset;
#endif
#ifdef WIN32
	HANDLE fd, namedPipe;
	HANDLE	events[2];
	DWORD available = 0;
	DWORD waited;
	unsigned long pending;
#endif

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

	
	chanId = *(int *)id;
	sock = signalChannels[chanId].socket;
	fd = signalChannels[chanId].pipe[0];
#ifdef WIN32
	events[0] = WSACreateEvent();
	WSAEventSelect(sock, events[0], FD_READ);
	events[1] = signalChannels[chanId].pipe[1];
	namedPipe = signalChannels[chanId].pipe[2];
#endif

#ifdef LINUX_OS
	maxfd = max(fd, sock) + 1;
	FD_ZERO(&rset);
#endif
    
	
	for (;;) {
		
#ifdef LINUX_OS
		FD_SET(sock, &rset);
		FD_SET(fd, &rset);
		if (select(maxfd, &rset, NULL, NULL, NULL) == -1) {
			if (errno == EINTR)
				continue;
			printf("error during select in signalManager %d\n", chanId);
			exitThread();
		}
		
		if (FD_ISSET(sock, &rset)) {
#endif

#ifdef WIN32
#ifdef DEBUG
		printf("before waitformultiple...\n");
#endif
		waited = WaitForMultipleObjects(2, (const HANDLE *)events, FALSE, INFINITE);
#ifdef DEBUG
		printf("out of waitformultiple...\n");
#endif
		
		
		ResetEvent(events[0]);
		
		pending = 0;
		ioctlsocket(sock, FIONREAD, &pending);
#ifdef DEBUG
		printf("pending: %d\n", pending);
#endif
		while (pending > 0) {
#endif
			

			
			size = recv(sock, (char *) &type, sizeof(type), 0);
#ifdef DEBUG
			printf("received type %d on socket\n", type);
#endif
			switch (type) {
			case TSP_ACK_CLOSED_FLOW:

				
				recv(sock, (char *) buffer, sizeof(buffer), 0);
				flowId = *(unsigned int *) buffer;
#ifdef DEBUG
				printf("received ack closed flow %d\n", flowId);
#endif
				

				
				if ( isChannelClosable(chanId))  {
					
					msg.code = MSG_FP_END;