/* ------------------------------------------------------------------------
 *
 * commander.c
 *
 * Short description:
 *
 * ------------------------------------------------------------------------
 *
 * Copyright (c) 2017, Ericsson Canada Inc
 * 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/poll.h>
#include <sys/time.h>
#include <sys/prctl.h>
#include <common.h>

struct context { 
 	bool commander;
	bool all_tests;
 	bool unicast_test;
 	bool anycast_test;
 	bool anycast_loopback_test;
 	bool multicast_test;
 	bool multicast_loopback_test;
 	bool broadcast_test;
 	bool broadcast_loopback_test;
 	bool sequence_test;
 	bool multipoint_test;
 	int conf_workers;
	uint32_t node;
 	int msg_len;
 	int duration;
 	int stats_intv;
 	int sd;
 	int scaler_cnt;
 	int test;
 	int result;
 	int repeats;
};

static int eval_result(struct context *ctx, uint32_t member_id, uint32_t res)
{
	if (ctx->test != MULTICAST_TEST)
		return res;
	if (member_id != RECEIVER_ID2)
		return res;
	if (res != NOT_RECEIVED)
		return res;
	return 0;
}

static int wait_for_events(struct context *ctx, uint32_t id,
			   bool up, uint32_t exp, int timeout)
{
	struct tipc_addr listener = {COMMAND_GROUP, LISTENER_ID, ctx->node};
	struct pollfd pfd = {ctx->sd, POLLIN, 0};
	struct tipc_addr sockid, memberid;
	char sstr[60], mstr[60];
	char buf[100];
	int rc, err;
	int cnt = 0;

	if (!exp)
		return 0;

	if (!ctx->sd) {
		pfd.fd = tipc_socket(SOCK_RDM);
		if (pfd.fd < 0)
			die("Commander: failed to create listener socket\n");
		tipc_join(pfd.fd, &listener, true, false);
	}

	while (cnt < exp) {
		rc = poll(&pfd, 1, timeout);
		if (rc < 0)
			die("poll() failed\n");
		if (rc == 0)
			break;

		rc = tipc_recvfrom(pfd.fd, buf, 100, &sockid, &memberid, &err);
		if (rc != 0)
			die("rcvfrom() returned %u\n", rc);

		tipc_ntoa(&sockid, sstr, sizeof(sstr));
		tipc_ntoa(&memberid, mstr, sizeof(mstr));
		if (memberid.instance != id)
			continue;
		if (err)
			dbg2("Commander: Lost Member %s/%s\n", sstr, mstr);
		else
			dbg2("Commander: Discovered Member %s/%s\n", sstr, mstr);
		if ((!err && up) || (err && !up))
			cnt++;
	}

	if (!ctx->sd)
		tipc_close(pfd.fd);

	printf("Commander: Received %u %s Events for Member Id %u\n",
	       cnt, up ? "UP" : "DOWN", id);
	return cnt;
}

static int receive_reports(struct context *ctx, int expected, int *left, int wait)
{
	struct pollfd pfd = {ctx->sd, POLLIN, 0};
	struct tipc_addr sockid, memberid;
	char str[60], str2[60], str3[256];
	struct report_hdr hdr;
	int rc, err, res;
	int snt, rcvd, uc, ac, mc, bc, first, last;
	int reports = 0;
	int inst;

	do {
		rc = poll(&pfd, 1, wait);
		if (rc < 0)
			die("poll() failed\n");
		if (!rc)
			break;
		rc = tipc_recvfrom(ctx->sd, &hdr, sizeof(hdr),
				   &sockid, &memberid, &err);
		if (rc < 0)
			die("rcvfrom() failed\n");

		/* Member event */
		if (rc == 0) {
			if (!err)
				continue;
			(*left)++;
			if (!expected)
				break;
			continue;
		}

		/* Memberid report */
		inst = memberid.instance;
		tipc_ntoa(&sockid, str, sizeof(str));
		tipc_ntoa(&memberid, str2, sizeof(str2));
		sprintf(str3, "Report #%-2u from %s: ", reports, hdr.mbrstr);
		res = eval_result(ctx, memberid.instance, ntohl(hdr.error));
		ctx->result |= res;
		uc = ntohl(hdr.snt_uc);
		ac = ntohl(hdr.snt_ac);
		mc = ntohl(hdr.snt_mc);
		bc = ntohl(hdr.snt_bc);
		snt = uc + ac + mc + bc;
		if (inst == TRANSMITTER_ID) {
			printf("%sSent %u [%u,%u] (UC %u, AC %u, MC %u, BC %u) %s\n",
			       str3, snt, 0, snt ? snt - 1 : 0, uc, ac, mc, bc, errstr(res));
			memset(str3, 0x20, strlen(str3));
		}
		uc = ntohl(hdr.rcv_uc);
		ac = ntohl(hdr.rcv_ac);
		mc = ntohl(hdr.rcv_mc);
		bc = ntohl(hdr.rcv_bc);
		rcvd = uc + ac + mc + bc;
		first = ntohl(hdr.rcv_first);
		last = ntohl(hdr.rcv_last);
		reports++;
		if (inst == TRANSMITTER_ID && !rcvd)
			continue;
		printf("%sRecv %u [%u,%u] (UC %u, AC %u, MC %u, BC %u) %s\n",
		       str3, rcvd, first, last, uc, ac, mc, bc, errstr(res));
	} while (reports < expected || *left < expected);

	return reports;
}

static void build_command(struct command_hdr *hdr, struct context *ctx,
			  int cmd, int memberid, bool events, bool loopback)
{
	hdr->cmd = htonl(cmd);
	hdr->loopback = loopback;
	hdr->events = events;
	hdr->msg_len = htonl(ctx->msg_len);
	hdr->memberid = htonl(memberid);
	hdr->stats_intv = htonl(ctx->stats_intv);
}

static void scale_out(struct context *ctx,  int add, int cmd_id, int trf_id,
		      bool events, bool loopback)
{
	struct tipc_addr scaler = {COMMAND_GROUP, SCALER_ID, ctx->node};
	struct tipc_addr worker = {COMMAND_GROUP, cmd_id, ctx->node};
	struct command_hdr hdr;
	int disc, remains = add;

	if (!add)
		return;

	printf("Commander: Scaling out to %u Workers with Id %u/%u\n", add, cmd_id, trf_id);

	if (add == 1 && cmd_id != RECEIVER_ID1) {
		worker_main(cmd_id, true, ctx->node);
	} else {
		build_command(&hdr, ctx, CREATE_WORKER, cmd_id, false, false);
		while (remains--) {
			tipc_sendto(ctx->sd, &hdr, sizeof(hdr), &scaler);
		}
	}

	disc = wait_for_events(ctx, cmd_id, true, add, 20000);
	if (disc != add)
		die("Failed to discover %u workers of id %u/%u\n", add - disc, cmd_id, trf_id);
	build_command(&hdr, ctx, JOIN, trf_id, events, loopback);
	tipc_mcast(ctx->sd, &hdr, sizeof(hdr), &worker);

}

static void scale_in(struct context *ctx, int receivers, int receiver_id)
{
	struct tipc_addr dst = {COMMAND_GROUP, receiver_id, ctx->node};
	struct command_hdr hdr;

	printf("Commander: Scaling in to 0 Workers with Cmd Member Id %u\n", receiver_id);
	build_command(&hdr, ctx, LEAVE, receiver_id, false, false);
	tipc_mcast(ctx->sd, &hdr, sizeof(hdr), &dst);
}

static void test(struct context *ctx, int cmd, int receiver_id, 
		 int transmitters, int receivers1, int receivers2,
		 bool events, bool loopback)
{
	struct tipc_addr transmitter = {COMMAND_GROUP, TRANSMITTER_ID, ctx->node};
	int workers = receivers1 + receivers2 + transmitters;
	const char *test = cmdstr(cmd);
	struct command_hdr hdr;
	int reports, left = 0;

	printf(">> Starting %s %s\n", test, loopback ? "with Loopback" : "");
	ctx->test = cmd;
	ctx->result = OK;

	/* Let transmitter(s) join, potentially also as receiver(s) */
	scale_out(ctx, transmitters, TRANSMITTER_ID, RECEIVER_ID1, events, loopback);

	/* Let transmitter(s) start the test before scaleout of receivers */
	build_command(&hdr, ctx, cmd, receiver_id, events, false);
	tipc_mcast(ctx->sd, &hdr, sizeof(hdr), &transmitter);

	/* Start the receivers and let them join the traffic group */
	scale_out(ctx, receivers1, RECEIVER_ID1, RECEIVER_ID1, false, false);

	/* Additional broadcast receivers/multicast non-receivers */
	scale_out(ctx, receivers2, RECEIVER_ID2, RECEIVER_ID2, false, false);

	/* Wait the given duration */
	reports = receive_reports(ctx, 0, &left, ctx->duration);
	if (reports)
		die("%s Received unexpected failure report\n", test);
	if (left)
		die("%s %u Workers left unexpectedly\n", test, left);

	scale_in(ctx, receivers1, RECEIVER_ID1);
	scale_in(ctx, receivers2, RECEIVER_ID2);
	scale_in(ctx, transmitters, TRANSMITTER_ID);

	/* Receive the final reports */
	reports = receive_reports(ctx, workers, &left, 40000);
	if (reports != workers)
		die("%s: %u Workers didn't report\n", test, workers - reports);
	if (left != workers)
		die("%s: %u Workers didn't leave\n", test, workers - left);
	if (ctx->result)
		die("%s: FAILED: %s\n", test, errstr(ctx->result));

	printf(">> %s SUCCESSFUL \n\n", test);
}

static void commander_main(struct context *ctx)
{
	struct tipc_addr self = {COMMAND_GROUP, COMMANDER_ID, ctx->node};
	struct tipc_addr sockid;
	char str[60], str2[60];
	int wcnt = ctx->conf_workers;

	printf("*** TIPC Group Messaging Test Started ****\n\n");

	ctx->sd = tipc_socket(SOCK_RDM);
	if (ctx->sd < 0)
		die("Commander: failed to create command socket\n");
	tipc_join(ctx->sd, &self, true, false);
	tipc_ntoa(&self, str, sizeof(str));
	tipc_sockaddr(ctx->sd, &sockid);
	tipc_ntoa(&sockid, str2, sizeof(str2));
	dbg1("Commander: Joined Command Group as %s/%s\n", str, str2);

	printf("Commander: Waiting for Scalers\n");
	ctx->scaler_cnt = wait_for_events(ctx, SCALER_ID, true, 1, 20000);

	printf("Commander: Discovered %u Scalers\n", ctx->scaler_cnt);

	if (ctx->all_tests || ctx->unicast_test)
		test(ctx, UNICAST_TEST, RECEIVER_ID1, 1, 1, 0, true, false);

	if (ctx->all_tests || ctx->anycast_test)
		test(ctx, ANYCAST_TEST, RECEIVER_ID1, 1, wcnt, 0, true, false);

	if (ctx->all_tests || ctx->anycast_loopback_test)
		test(ctx, ANYCAST_TEST, RECEIVER_ID1, 1, wcnt, 0, true, true);

	if (ctx->all_tests || ctx->multicast_test)
		test(ctx, MULTICAST_TEST, RECEIVER_ID1, 1, wcnt / 2, wcnt / 2, true, false);

	if (ctx->all_tests || ctx->multicast_loopback_test)
		test(ctx, MULTICAST_TEST, RECEIVER_ID1, 1, wcnt / 2, wcnt / 2, true, true);

	if (ctx->all_tests || ctx->broadcast_test)
		test(ctx, BROADCAST_TEST, RECEIVER_ID1, 1, wcnt / 2, wcnt / 2, true, false);

	if (ctx->all_tests || ctx->broadcast_loopback_test)
		test(ctx, BROADCAST_TEST, RECEIVER_ID1, 1, wcnt / 2, wcnt / 2, true, true);

	if (ctx->all_tests || ctx->sequence_test) {
		test(ctx, UC_AFT_BC_SEQ_TEST, RECEIVER_ID1, 1, wcnt, 0, true, false);
		test(ctx, BC_AFT_UC_SEQ_TEST, RECEIVER_ID1, 1, wcnt, 0, true, false);
	}

	if (ctx->all_tests || ctx->multipoint_test)
		test(ctx, MULTIPOINT_TEST, RECEIVER_ID2, wcnt, 0, 1, false, false);

	tipc_close(ctx->sd);

	printf("*** TIPC Group Messaging Test Finished ****\n\n");
}

static void syntax(char *argv[])
{
	fprintf(stderr, "Usage: ");
	fprintf(stderr, "bus_test ");
	fprintf(stderr, "[-c [<test>]] [-w <workers>] [-l <msg length>] [-d <duration>]\n"
		"        [-i <interval>] [-v [-v]\n"
		"          -c make this instance the commander (default off)\n"
		"          -u run unicast test only (default off)\n"
		"          -a run anycast test only (default off)\n"
		"          -A run anycast test with loopback only (default off)\n"
		"          -m run multicast test only (default off)\n"
		"          -M run multicast test with loopback only (default off)\n"
		"          -b run broadcast test only (default off)\n"
		"          -B run broadcast test with loopback only (default off)\n"
		"          -s run sequence number test only (default off)\n"
		"          -f run flow control multipoint-to-point test only (default off)\n"
		"          -w set size of traffic group to use (default 32)\n"
		"          -n node local groups (default cluster global)\n"
		"          -l set length of messages to be exchanged (default 66,000)\n"
		"          -d set duration of each test (default 20 sec)\n"
		"          -r set number of times to repeat test (default 1)\n"
		"          -i set statistics report interval (default 4 sec)\n"
		"          -v verbosity level ([0,3] default 0)\n");
}

static int parse_args(int argc, char *argv[], struct context *args)
{
	int option;

	memset(args, 0, sizeof(*args));
	args->msg_len = 66000;
	args->duration = 20000;
	args->conf_workers = 32;
	args->stats_intv = 4;
	args->all_tests = true;
	args->repeats = 1;

	while ((option = getopt(argc, argv, "cw:l:d:r:i:uaAmMbBsfnhv")) != -1) {

		switch (option) {
		case 'c':
			args->commander = true;
			break;
		case 'u':
			args->all_tests = false;
			args->unicast_test = true;
			break;
		case 'a':
			args->all_tests = false;
			args->anycast_test = true;
			break;
		case 'A':
			args->all_tests = false;
			args->anycast_loopback_test = true;
			break;
		case 'm':
			args->all_tests = false;
			args->multicast_test = true;
			break;
		case 'M':
			args->all_tests = false;
			args->multicast_loopback_test = true;
			break;
		case 'b':
			args->all_tests = false;
			args->broadcast_test = true;
			break;
		case 'B':
			args->all_tests = false;
			args->broadcast_loopback_test = true;
			break;
		case 's':
			args->all_tests = false;
			args->sequence_test = true;
			break;
		case 'f':
			args->all_tests = false;
			args->multipoint_test = true;
			break;
		case 'w':
			if (optarg)
			       args->conf_workers = atoi(optarg);
			break;
		case 'l':
			if (optarg)
			       args->msg_len = atoi(optarg);
			break;
		case 'd':
			if (optarg)
			       args->duration = atoi(optarg) * 1000;
			break;
		case 'i':
			if (optarg)
			       args->stats_intv = atoi(optarg);
			break;
		case 'r':
			if (optarg)
			       args->repeats = atoi(optarg);
			break;
		case 'n':
			args->node = tipc_own_node();
			break;
		case 'v':
			vlevel++;
			break;
		default:
			syntax(argv);
			return -1;
		}
		if (!args->all_tests)
			args->commander = true;
	}
	return 0;
}

int main(int argc, char *argv[])
{
	struct context args;
	int i = 0;

	if (parse_args(argc, argv, &args))
		exit(EXIT_FAILURE);

	if (args.commander && wait_for_events(&args, COMMANDER_ID, true, 1, 200)) {
		printf("Commander exists, ignoring arguments\n");
		args.commander = false;
	}

	if (!args.commander) {
		scaler_main(args.node);
		exit(EXIT_SUCCESS);
	}
	while (args.repeats-- && !args.result) {
		if (args.repeats)
			printf("ITERATION # %u:\n\n", ++i);
		commander_main(&args);
	}
	exit(args.result);
}
