/* 

** Build command: g++ -lpthread exchange.cpp -o exchange

*/

#include <stdio.h>

#include <sys/select.h>

#include <sys/time.h>

#include <sys/types.h>

#include <unistd.h>

#include <sys/socket.h>

#include <arpa/inet.h>

#include <netinet/in.h>

#include <stdlib.h>

#include <memory.h>

#include <stdint.h>

#include <string.h>

#include <pthread.h>

#include <sched.h>

#include <errno.h>

#define NUM_SOCKETS 2

#define MC_SOCKET 0

#define UC_SOCKET 1

#define NUM_PACKETS 200000

#define IF_ADDRESS "1.1.1.18"

#define UC_SERVER_ADDRESS "1.1.1.19"

#define MC_ADDRESS "224.0.1.2"

#define MC_DEST_PORT 15111

#define UC_LOCAL_PORT 15222

#define UC_SERVER_PORT 15333

#define MC_BUFFLEN 200

#define UC_BUFFLEN 12

#define MIN_UC_BUFFLEN 10

#define SLEEP_TIME_USEC 10

#define MAX_PARAM_LENGTH 20

int fd_list[NUM_SOCKETS];

uint64_t tx_pkt_count, delta_usec_quote;

struct timeval tv_quote_start, tv_quote_end;

char if_address[MAX_PARAM_LENGTH] = "NO IF ADDRESS!!!";

int num_packets = NUM_PACKETS;

char mc_address[MAX_PARAM_LENGTH] = MC_ADDRESS;

uint16_t mc_dest_port = MC_DEST_PORT;

uint16_t uc_local_port = UC_LOCAL_PORT;

int mc_bufflen = MC_BUFFLEN;

int uc_bufflen = UC_BUFFLEN;

uint64_t sleep_time_usec = SLEEP_TIME_USEC;

void usage(void)

{

	printf("Usage:\n");

	printf("\t-l\t<MANDATORY! local interface ip address for mc and uc>\n");

	printf("\t[-n]\t<optional num of mc packets before marking for QUOTE. default 200000>\n");

	printf("\t[-m]\t<optional mc address. default - 224.0.1.2>\n");

	printf("\t[-pm]\t<optional mc destination port. default 15111>\n");

	printf("\t[-lp]\t<optional local uc port. default 15222>\n");

	printf("\t[-sm]\t<optional mc massage payload size. default 200>\n");

	printf("\t[-su]\t<optional uc massage payload size. MIN value = 10. default 12>\n");

	printf("\t[-u]\t<optional sleep time in usec between each mc packet send. default 10usec>\n");

}

int prepare_socket()

{

	struct sockaddr_in groupsock;

	struct in_addr localInterface;

	int fd = socket(AF_INET, SOCK_DGRAM, 0);

	if(fd < 0)

	{

			perror("Opening datagram socket error");

			exit(1);

	}

	memset(&groupsock, 0, sizeof(groupsock));

	groupsock.sin_family            = AF_INET;

	groupsock.sin_addr.s_addr       = inet_addr(mc_address);

	groupsock.sin_port              = htons(mc_dest_port);

	/* Disable loopback so you do not receive your own datagrams.*/

	char loopch = 0;

	if(setsockopt(fd, IPPROTO_IP, IP_MULTICAST_LOOP, (char *)&loopch, sizeof(loopch)) < 0)

	{

		perror("Setting IP_MULTICAST_LOOP error");

		close(fd);

		exit(1);

    	}

	/* Set local interface for outbound multicast datagrams. */

	localInterface.s_addr = inet_addr(if_address);

	if(setsockopt(fd, IPPROTO_IP, IP_MULTICAST_IF, (char *)&localInterface, sizeof(localInterface)) < 0)

	{

		perror("Setting local interface error");

 		exit(1);

	}

	printf("Connecting..\n");

	if(connect(fd, (struct sockaddr *) &groupsock, sizeof(struct sockaddr)))

	{

		perror("connect");

		close(fd);

 		exit(1);

	}

        return fd;

}

void* send_mc_loop(void* num)

{

	int ret;

	char databuf[mc_bufflen];

	char quote[] = "QUOTE";

	uint64_t delta_usec, delta_usec_sleep;

	/* Prepare MC socket */

	printf("Opening datagram MC socket\n");

	fd_list[MC_SOCKET] = prepare_socket();

	// Prepare to start measurements

	tx_pkt_count = 0;

	struct timeval tv_start, tv_sleep_start, tv_sleep_end;

	gettimeofday(&tv_start, NULL);

	gettimeofday(&tv_sleep_start, NULL);

	gettimeofday(&tv_sleep_end, NULL);

	while(true)

	{

		delta_usec_sleep = ((tv_sleep_end.tv_sec - tv_sleep_start.tv_sec) * 1000000) + (tv_sleep_end.tv_usec - tv_sleep_start.tv_usec);

		if (delta_usec_sleep > sleep_time_usec)

		{

			ret = send(fd_list[MC_SOCKET], databuf, mc_bufflen, 0);	// Can use send with UDP socket because called connect() before...

			if (ret < 0)

				printf("ERROR on SEND errno = %s\n", strerror(errno));

			tx_pkt_count++;

			tv_sleep_start = tv_sleep_end;

		}

		else

		{

			gettimeofday(&tv_sleep_end, NULL);

		}

		if ((tx_pkt_count != 0) && (tx_pkt_count % num_packets) == 0) {

			struct timeval tv_now;

			gettimeofday(&tv_now, NULL);

			delta_usec = ((tv_now.tv_sec - tv_start.tv_sec) * 1000000) + (tv_now.tv_usec - tv_start.tv_usec);

			tv_start = tv_now;

			double mps = 1000000 * (tx_pkt_count/(double)delta_usec);

			double bwGbps = mps * mc_bufflen * 8/(1024*1024*1024);

			printf("BW(Gbps)=%6.3f, MPS=%10.0f\n", bwGbps, mps);

			tx_pkt_count = 0;

			gettimeofday(&tv_quote_start, NULL);

			ret = send(fd_list[MC_SOCKET], quote, sizeof(quote), 0);

			if (ret < 0)

				printf("ERROR on SEND errno = %s\n", strerror(errno));

		}

	}

	return 0;

}

void * uc_func(void * num)

{

	struct sockaddr_in localSock, servaddr;

	socklen_t servaddrlen = sizeof(struct sockaddr);

	char buf[uc_bufflen], ord_ack[] = "ORD_ACK", ka_ack[] = "KA_ACK";

	int ret, print = 0;

	fd_list[UC_SOCKET] = socket(AF_INET, SOCK_DGRAM, 0);

	if(fd_list[UC_SOCKET] < 0)

	{

			perror("Opening datagram UC socket error");

			exit(1);

	}

	printf("Opening datagram UC socket....OK.\n");

	memset((char *) &localSock, 0, sizeof(localSock));

	localSock.sin_family = AF_INET;

	localSock.sin_addr.s_addr = inet_addr(if_address);

	localSock.sin_port = htons(uc_local_port);

	if(bind(fd_list[UC_SOCKET], (struct sockaddr*)&localSock, sizeof(struct sockaddr)))

	{

		perror("Binding datagram UC socket error");

		close(fd_list[UC_SOCKET]);

		exit(1);

	}

	else

	{

		printf("Binding datagram UC socket...OK.\n");

	}

	while(1)

	{

		ret = recvfrom(fd_list[UC_SOCKET], buf, uc_bufflen, 0, (struct sockaddr *)&servaddr, &servaddrlen);

		gettimeofday(&tv_quote_end, NULL);

		if (ret < 0)

		{

			printf("ERROR on RECV errno = %s \n", strerror(errno));

			printf("errno value = %d\n", errno);

		}

		else

		{

			if (strcmp(buf, "ORD") == 0)

			{

				ret = sendto(fd_list[UC_SOCKET], ord_ack, sizeof(ord_ack), 0, (struct sockaddr *) &servaddr, sizeof(struct sockaddr));

				if (ret < 0)

				{

					printf("ERROR on SEND UC errno = %s \n", strerror(errno));

					printf("errno value = %d\n", errno);

				}

				print = 1;

			}

			else if (strcmp(buf, "KA") == 0){

				ret = sendto(fd_list[UC_SOCKET], ka_ack, sizeof(ka_ack), 0, (struct sockaddr *) &servaddr, sizeof(struct sockaddr));

				if (ret < 0)

				{

					printf("ERROR on SEND UC errno = %s \n", strerror(errno));

					printf("errno value = %d\n", errno);

				}

			}

			else{

				printf("Internal error: Exchange received UC packet- not ORD or KA\n");

			}

			if (print)

			{

				delta_usec_quote = ((tv_quote_end.tv_sec - tv_quote_start.tv_sec) * 1000000) + (tv_quote_end.tv_usec - tv_quote_start.tv_usec);

				printf("@@@@@@@ QUOTE from port %u RTT in usec = %llu @@@@@@@\n", ntohs(servaddr.sin_port), (long long unsigned int)delta_usec_quote);

				print = 0;

			}

		}

	}

	close(fd_list[UC_SOCKET]);

	printf("closed UC socket\n");

	return 0;

}

int main(int argc, char *argv[])

{

	pthread_t uc_thread;

	int nThreadId = 1, i;

	for (i=1; i

	{

		if (strcmp(argv[i], "-l") == 0) {

			strcpy(if_address, argv[i+1]);

		} else if (strcmp(argv[i], "-n") == 0) {

			num_packets = atoi(argv[i+1]);

		} else if (strcmp(argv[i], "-m") == 0) {

			strcpy(mc_address, argv[i+1]);

		} else if (strcmp(argv[i], "-pm") == 0) {

			mc_dest_port = atoi(argv[i+1]);

		} else if (strcmp(argv[i], "-lp") == 0) {

			uc_local_port = atoi(argv[i+1]);

		} else if (strcmp(argv[i], "-sm") == 0) {

			mc_bufflen = atoi(argv[i+1]);

		} else if (strcmp(argv[i], "-su") == 0) {

			uc_bufflen = atoi(argv[i+1]);

			if (uc_bufflen < MIN_UC_BUFFLEN) {

				uc_bufflen = MIN_UC_BUFFLEN;

			}

		} else if (strcmp(argv[i], "-u") == 0) {

			sleep_time_usec = atoi(argv[i+1]);

		} else if ((strcmp(argv[i], "-h") == 0) || (strcmp(argv[i], "-help") == 0) || (strcmp(argv[i], "--help") == 0) || (strcmp(argv[i], "--h") == 0)) {

			usage();

			return 0;

		}

	}

	if ((argc == 1) || (strcmp(if_address, "NO IF ADDRESS!!!") == 0)) {

		usage();

		return 0;

	}

	pthread_create(&uc_thread, NULL, uc_func, (void*)nThreadId);

	send_mc_loop(0);

	printf("Going to close MC socket\n");

	close(fd_list[MC_SOCKET]);

	printf("Closed MC socket\n");

	return 0;

}