/*
 * 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/prctl.h>

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

static struct tipc_addr srv = {
	.type = 76453,
	.instance = 17,
	.node = 0
};

struct hdr {
	unsigned int seqno;
	unsigned int remains;
};

struct peer {
	struct peer *next;
	unsigned int rcv_nxt;
	unsigned int cnt;
	unsigned int total_cnt;
	struct tipc_addr addr;
};

struct peer *peers = NULL;
#define BUF_LEN 66000
static char buf[BUF_LEN];
static char srvbuf[20];
static int verbose;

struct peer* find_peer(struct tipc_addr *addr)
{
	struct peer **peer;

	for (peer = &peers; *peer; peer = &(*peer)->next) {
		if ((*peer)->addr.instance != addr->instance)
			continue;
		if ((*peer)->addr.node != addr->node)
			continue;
		return *peer;
	}
	*peer = calloc(1, sizeof(struct peer));
	(*peer)->addr.instance = addr->instance,
	(*peer)->addr.node = addr->node;
	return *peer;
}

void mcast_transmitter(int len, unsigned int count)
{
	struct hdr *hdr = (struct hdr *)buf;
	unsigned int remains = count;
	int sd;

	printf("Waiting for first peer\n");
	tipc_srv_wait(&srv, -1);
	sd = tipc_socket(SOCK_DGRAM);
	if (sd < 0)
		die("Failed to create client socket\n");

	printf("Multicasting %u DGRAM messages of size %u bytes\n", count, len);
	while(remains) {
		hdr->seqno = htonl(ntohl(hdr->seqno) + 1);
		hdr->remains = htonl(count);
		if (0 >= tipc_mcast(sd, buf, len, &srv))
			die("Failed to send multicast\n");
		remains--;
	}
	printf("Transmitting finished: sent %u messages \n", count);
}

void mcast_receiver(void)
{
	int sd;
	struct tipc_addr src;
	struct peer *peer;
	unsigned int _seqno, rcv_nxt, gap;
	struct hdr *hdr = (struct hdr *)buf;

	prctl(PR_SET_PDEATHSIG, SIGHUP);

	sd = tipc_socket(SOCK_RDM);
	if (sd < 0)
		die("Failed to create receiver socket\n");

	if (tipc_bind(sd, srv.type, srv.instance, srv.instance, srv.node))
		die("Failed to bind receiver socket");

	printf("Receiver socket created and bound to %s\n", srvbuf);

	while (1) {
		if (0 >= tipc_recvfrom(sd, buf, BUF_LEN, &src, NULL, 0))
			die("unexpected message on multicast receiver socket\n");
		_seqno = ntohl(hdr->seqno);
		peer = find_peer(&src);
		rcv_nxt = peer->rcv_nxt;
		gap = _seqno - rcv_nxt;
		if (gap && peer->cnt)
			printf("Missing %u msgs from %x:%u exp %u, rcv %u\n",
			       gap, src.node, src.instance, rcv_nxt, _seqno);
		peer->rcv_nxt = _seqno + 1;
		if (!peer->cnt)
			peer->total_cnt = ntohl(hdr->remains);
		if (!(peer->cnt++ % 10000) && verbose)
			printf("Now at %u msgs from %x:%u\n", 
			       peer->cnt, src.node, src.instance);

		if (ntohl(hdr->remains) != 1)
			continue;
		printf("Reception finished: received %u messages from %x:%u, missing %u\n",
		       peer->cnt, src.node, src.instance, peer->total_cnt - peer->cnt);
		memset(peer, 0, sizeof(struct peer));
	}
}

void mcast_receiver_subscriber(void)
{
	struct tipc_addr rcvr;
	bool up;
	int sd;

	prctl(PR_SET_PDEATHSIG, SIGHUP);

	sd = tipc_topsrv_conn(0);
	if (sd < 0)
		die("Failed to create subscriber socket\n");

	if (tipc_srv_subscr(sd, srv.type, srv.instance, srv.instance, true, -1))
		die("subscription for server failed\n");

	while (1) {
		if (tipc_srv_evt(sd, &rcvr, 0, &up, 0))
			die("reception of service event failed\n");
		if (!verbose)
			continue;
		if (up)
			printf("New receiver discovered at node %x:%u\n",
			       rcvr.node, rcvr.instance);
		else
			printf("Receiver %x:%u lost\n", rcvr.node, rcvr.instance);
	}
}

int main(int argc, char *argv[])
{
	int c;
	int len = 66000;
	unsigned int count = ~0;
	bool xmit_only = false, rcvr_only = false;

	while ((c = getopt(argc, argv, ":csvl:n:")) != -1) {
		switch (c) {
		case 'l':
			if (optarg)
				len = atoi(optarg);
			continue;
		case 'c':
			xmit_only = true;
			continue;
		case 'v':
			verbose++;
			continue;
		case 's':
			rcvr_only = true;
			continue;
		case 'n':
			if (optarg)
				count = atoi(optarg);
			continue;
		default:
			fprintf(stderr, "Usage:\n");
			fprintf(stderr," %s ", argv[0]);
			fprintf(stderr, "[-c] [-s] [-l <msglen>]\n"
                         "       [-c client (xmitit) mode only (default off)\n"
                         "       [-s server (receive) mode only (default off)\n"
		         "       [-l <msglen>] message size (default 66000)\n"
                         "       [-v increase verbosity (default off)\n"
		         "       [-n <num_of_msgs>] number of messages to send (default 0xfffffff)\n");
			return 1;
		}
	}

	memset(buf, 0xff, sizeof(buf));
	tipc_ntoa(&srv, srvbuf, sizeof(srvbuf));

	if (!xmit_only && !fork())
		mcast_receiver();

	if (rcvr_only)
		while(1);

	if (!fork())
		mcast_receiver_subscriber();

	mcast_transmitter(len, count);
	exit(0);
}
