/* ------------------------------------------------------------------------
 *
 * mcast_tipc.c
 *
 * Short description: TIPC multicast demo (client side)
 *
 * ------------------------------------------------------------------------
 *
 * Copyright (c) 2015, Ericsson Canada
 * All rights reserved.
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * Redistributions of source code must retain the above copyright notice, this
 * list of conditions and the following disclaimer.
 * Redistributions in binary form must reproduce the above copyright notice,
 * this list of conditions and the following disclaimer in the documentation
 * and/or other materials provided with the distribution.
 * Neither the name of the copyright holders nor the names of its
 * contributors may be used to endorse or promote products derived from this
 * software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 *
 * ------------------------------------------------------------------------
 */

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/param.h>
#include <errno.h>
#include <unistd.h>
#include <netinet/in.h>
#include <tipcc.h>
#include <sys/epoll.h>
#include <sys/time.h>
#include <sys/prctl.h>

#define INTV_SZ 10000
#define MAX_PEERS 4096

#define die(fmt, arg...) do			\
	{					\
		printf("groupcast:" fmt, ## arg);\
		perror(NULL);			\
		exit(1);			\
	} while(0)

static struct tipc_addr self = {
	.type = 4711,
	.instance = 0,
	.node = 0
};

static struct tipc_addr sockid;

static struct tipc_addr dst = {
	.type = 4711,
	.instance = 0,
	.node = 0
};

#define BUF_LEN 66000
static char buf[BUF_LEN];
static char self_str[30];
static char sockid_str[30];
static char memberid_str[30];

unsigned int member_cnt = 0;
unsigned int dst_member_cnt = 0;
unsigned int snt_bc = 0;
unsigned int snt_mc = 0;
unsigned int snt_ac = 0;
unsigned int snt_uc = 0;
unsigned int rcv_bc = 0;
unsigned int rcv_mc = 0;
unsigned int rcv_ac = 0;
unsigned int rcv_uc = 0;

enum {
	NONE =  0,
	BCAST = 1,
	MCAST = 2,
	UCAST = 3,
	ACAST = 4
};

static struct hdr {
	int type : 8;
	bool reply;
} *hdr = (void*)buf;

static unsigned long long elapsedmsec(struct timeval *from, struct timeval *to)
{
	long long from_us, to_us;

	from_us = from->tv_sec * 1000000 + from->tv_usec;
	to_us = to->tv_sec * 1000000 + to->tv_usec;
	return (to_us - from_us) / 1000;
}

void pr_stats(int rcv_cnt)
{
	if (rcv_cnt % INTV_SZ)
		return;
	printf("Recv %u broadcasts, %u multicasts, %u anycasts, %u unicasts; ",
	       rcv_bc, rcv_mc, rcv_ac, rcv_uc);
	printf("Sent %u unicasts\n", snt_uc);

}

void send_msgs(int sfd, int mtyp, bool reply, int len)
{
	int rc;

	hdr->type = mtyp;
	hdr->reply = reply;

	do {
		if (mtyp == BCAST) {
			rc = tipc_send(sfd, buf, len);
			if (rc == len)
				snt_bc++;
		} else 	if (mtyp == ACAST) {
			rc = tipc_sendto(sfd, buf, len, &dst);
			if (rc == len)
				snt_ac++;
		} else 	if (mtyp == MCAST) {
			rc = tipc_mcast(sfd, buf, len, &dst);
			if (rc == len)
				snt_mc++;
		} else {
			die("Unknown msg type\n");
		}
	} while (rc == len);

	if (errno != EAGAIN)
		die("Failed to send\n");
}

void send_reply(int efd, int sfd, int len, struct tipc_addr *to)
{
	struct epoll_event revents;
	int rc;

	hdr->type = UCAST;
	hdr->reply = false;

	while (1) {
		rc = tipc_sendto(sfd, buf, len, to);
		if (rc == len)
			return;
		if (errno != EAGAIN)
			break;
		if (0 > epoll_wait(efd, &revents, 100, -1))
			die("epoll() failed\n");
	}
}

void member_event(struct tipc_addr *sockid, struct tipc_addr *memberid, int err)
{
	tipc_ntoa(memberid, memberid_str, sizeof(memberid_str));
	tipc_ntoa(sockid, sockid_str, sizeof(sockid_str));
	if (!err) {
		printf("Member %s discovered at %s\n",
		       memberid_str, sockid_str);
		member_cnt++;
		if (memberid->instance == dst.instance)
			dst_member_cnt++;
	} else {
		printf("Member %s at %s lost\n", memberid_str, sockid_str);
		member_cnt--;
		if (memberid->instance == dst.instance)
			dst_member_cnt--;
	}
}

void read_msgs(int efd, int sfd, int len)
{
	struct tipc_addr sockid, memberid;
	int rc, err;

	while (1) {
		rc = tipc_recvfrom(sfd, buf, BUF_LEN, &sockid, &memberid, &err);
		if (rc == 0) {
			member_event(&sockid, &memberid, err);
			continue;
		}
		if (rc != len)
			break;

		if (hdr->type == BCAST)
			pr_stats(rcv_bc++);
		else if (hdr->type == MCAST)
			pr_stats(rcv_mc++);
		else if(hdr->type == ACAST)
			pr_stats(rcv_ac++);
		else if(hdr->type == UCAST)
			pr_stats(rcv_uc++);

		if (hdr->reply) {
			send_reply(efd, sfd, len, &sockid);
			snt_uc++;
		};
	}
	if (errno != EAGAIN)
		die("Received unexpected msg on member socket\n");
}

void group_transceive(int efd, int sfd, int mtyp, bool reply, int len)
{
	struct epoll_event revents;

	if (0 > epoll_wait(efd, &revents, 100, -1))
		die("epoll() failed\n");

	if (revents.events & EPOLLERR)
		die("epoll() EPOLLERR\n");

	if (revents.events & EPOLLIN)
		read_msgs(efd, sfd, len);

	if (mtyp && dst_member_cnt && (revents.events & EPOLLOUT))
		send_msgs(sfd, mtyp, reply, len);
}

int syntax(char *argv[])
{
	fprintf(stderr, "Usage:\n");
	fprintf(stderr," %s ", argv[0]);
	fprintf(stderr, "[-b] [-m] [-a] [-l <msglen>]\n"
		"       [-b send group broadcast (default off)\n"
		"       [-m send group multicast (default off)\n"
		"       [-a send group anycast (default off)\n"
		"       [-r request response on sent Xcast (default off)\n"
		"       [-d destination member instance (default 0)\n"
		"       [-i member instance (default 0)\n"
		"       [-l <msglen>] message size (default 66000)\n"
		"       [-L loopback broadcast/multicast/anycast (default off)\n"
		"       -a, -b , -m are mutually exclusive\n");
	return 1;
}

int main(int argc, char *argv[])
{
	int option;
	int len = 66000;
	int mtyp = NONE;
	bool ucast_reply = false;
	bool loopback = false;
	unsigned int prev =0, snt = 0;
	struct epoll_event event;
	int conf_inst = 0;
	int efd, sfd;
	unsigned long long elapsed_ms_tot;
	unsigned long long elapsed_ms_intv;
	struct timeval start_time;
	struct timeval start_intv;
	struct timeval now;
	unsigned long msg_per_sec_tot, msg_per_sec_intv;
	unsigned int thruput_tot, thruput_intv;

	while ((option = getopt(argc, argv, ":abd:i:l:Lmr")) != -1) {
		switch (option) {
		case 'l':
			if (optarg)
				len = atoi(optarg);
			continue;
		case 'i':
			if (optarg)
				conf_inst = atoi(optarg);
			continue;
		case 'b':
			if (mtyp)
				return syntax(argv);
			mtyp = BCAST;
			continue;
		case 'm':
			if (mtyp)
				return syntax(argv);
			mtyp = MCAST;
			continue;
		case 'a':
			if (mtyp)
				return syntax(argv);
			mtyp = ACAST;
			continue;
		case 'r':
			ucast_reply = true;
			continue;
		case 'L':
			loopback = true;
			continue;
		case 'd':
			if (optarg)
				dst.instance = atoi(optarg);
			continue;
		default:
			return syntax(argv);
		}
	}

	/* Prepared epolled socket */
	efd = epoll_create(1);
	if (efd == -1)
		die("Failed to create epoll object\n");

	sfd = tipc_socket(SOCK_RDM);
	if (sfd < 0)
		die("Failed to create member socket\n");
	tipc_sock_non_block(sfd);

	event.data.fd = sfd;
	event.events = EPOLLIN | EPOLLOUT | EPOLLERR | EPOLLET;

	if (epoll_ctl(efd, EPOLL_CTL_ADD, sfd, &event) == -1)
		die("epoll_ctl failure");

	/* Join group */
	self.instance = conf_inst ? conf_inst : 0;
	if (tipc_join(sfd, &self, true, loopback))
		die("Failed to join group\n");
	tipc_ntoa(&self, self_str, sizeof(self_str));
	tipc_sockaddr(sfd, &sockid);
	tipc_ntoa(&sockid, sockid_str, sizeof(sockid_str));
	printf("Joined as member %s at socket %s\n", self_str, sockid_str);

	/* Start receiving/transmitting */
	memset(buf, 0xff, sizeof(buf));
	gettimeofday(&start_time, 0);
	gettimeofday(&start_intv, 0);

	while (1) {
		group_transceive(efd, sfd, mtyp , ucast_reply, len);
		snt = snt_bc + snt_mc + snt_ac + snt_uc;
		if (mtyp == NONE)
			continue;
		if ((snt - prev) < INTV_SZ)
			continue;
		prev = snt;

		/* Print out send statistics at regular intervals */
		gettimeofday(&now, 0);
		elapsed_ms_tot = elapsedmsec(&start_time, &now);
		msg_per_sec_tot = (snt * 1000) / elapsed_ms_tot;
		thruput_tot = (msg_per_sec_tot * BUF_LEN * 8) / 1000000;
		elapsed_ms_intv = elapsedmsec(&start_intv, &now);
		msg_per_sec_intv = (INTV_SZ * 1000) / elapsed_ms_intv;
		thruput_intv = (msg_per_sec_intv * BUF_LEN * 8) / 1000000;
		printf("Sent %u broadcast, %u multicast, %u anycast, "
		       "throughput %u MB/s, last intv %u MB/s\n",
		       snt_bc, snt_mc, snt_ac, thruput_tot, thruput_intv);
		start_intv = now;
	}
	exit(0);
}
