#include <sys/types.h>
#include <sys/socket.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <sys/param.h>
#include <sys/poll.h>
#include <sys/wait.h>
#include <signal.h>
#include <errno.h>
#include <unistd.h>
#include <linux/tipc.h>
#include <stdint.h>
#include <getopt.h>
#include <pthread.h>
#include "common.h"

struct req_param {
	uint32_t msgsize;
	uint32_t imp;
	uint32_t sleep;
	uint32_t nmsgs;
};

static int flags = 0;
struct sockaddr_tipc sa = { 0 };

/*
 * Create a msghdr with a single IOV with random data
 */
static struct msghdr *create_msg(uint32_t msgsize) {

	struct msghdr *m;
	struct iovec *iov;
	uint8_t *data;
	uint8_t *iovp;

	iov = malloc(sizeof(struct iovec));
	if (!iov)
		diep("malloc()");
	data = malloc(msgsize);
	iovp = data + sizeof(struct pkthdr);
	for (;iovp < data+msgsize; iovp++)
		*iovp = rand()%255;

	iov[0].iov_base = data;
	iov[0].iov_len = msgsize;

	m = malloc(sizeof(struct msghdr));
	if (!m)
		diep("malloc()");
	memset(m, 0, sizeof(*m));
	m->msg_iov = iov;
	m->msg_iovlen = 1;
	return m;
}

static void pollout_or_die(uint32_t fd)
{
	struct pollfd pfds[1];

	pfds->fd = fd;
	pfds->events = POLLOUT;

	if (poll(pfds, 1, 3500) != 1)
		diep("poll() timed out");
	if (pfds->revents^POLLOUT)
		diep("poll() Did not receive POLLOUT");
}


/*
 * Send TIPC RDM traffic over multicast, if nmsgs is zero
 * this will run forever
 */
static void *rdm_send(void *data)
{
	struct req_param *req = data;
	uint32_t sd = socket(AF_TIPC, SOCK_RDM, 0);
	struct msghdr *m;
	struct pkthdr *header;

	if (req->msgsize > TIPC_MAX_USER_MSG_SIZE) {
		req->msgsize = TIPC_MAX_USER_MSG_SIZE;
		printf("RDM: Message size clamped to %d\n", TIPC_MAX_USER_MSG_SIZE);
	}
	if (setsockopt(sd, SOL_TIPC, TIPC_IMPORTANCE, &req->imp, sizeof(uint32_t)) < 0)
		diep("TIPC_IMPORTANCE");

	m = create_msg(req->msgsize);
	m->msg_name = &sa;
	m->msg_namelen = sizeof(sa);
	header = m->msg_iov[0].iov_base;
	header->seq = 0;
	while (1) {
		if (req->nmsgs && (header->seq + 1) == req->nmsgs)
			header->extra = CMD_DONE;
		if ((sendmsg(sd, m, flags) <= 0) && (errno == EAGAIN)) {
			pollout_or_die(sd);
			continue;
		}
		if (req->sleep > 0) {
			usleep(req->sleep);
		}
		if (header->extra == CMD_DONE) {
			printf("Done sending %d messages\n", req->nmsgs);
			return 0;
		}
		header->seq++;
	}
	close(sd);
}

int32_t main(int32_t argc, char* argv[], char* envp[])
{
	char c;
	struct req_param req = {0};
	pthread_t *sender_thread;
	pthread_t *tmp_sender;
	uint32_t threads = 1;
	uint32_t i;
	int err;

	req.msgsize = TIPC_MAX_USER_MSG_SIZE;
	req.imp = TIPC_LOW_IMPORTANCE;
	/*Default to continuous blocking multicast send of 66k messages*/
	sa.family = AF_TIPC,
	sa.addrtype = TIPC_ADDR_MCAST,
	sa.addr.nameseq.type = TEST_TYPE,
	sa.addr.nameseq.lower = TEST_INSTANCE,
	sa.addr.nameseq.upper = TEST_INSTANCE,
	sa.addr.name.domain = 0;

	while ((c = getopt(argc, argv, "n:m:s:i:t:buh")) != -1)
	{
		switch(c) {
		case 'm':
			req.msgsize=atoi(optarg);
		break;
		case 'n':
			req.nmsgs = atoi(optarg);
		break;
		case 's':
			req.sleep = atoi(optarg);
			printf("Will sleep %u ns between each send\n", req.sleep);
		break;
		case 't':
			threads = atoi(optarg);
		break;
		case 'i':
			if (strcmp(optarg, "low") == 0) {
				req.imp = TIPC_LOW_IMPORTANCE;
				printf("Set importance to low\n");
			}
			else if (strcmp(optarg, "medium") == 0) {
				req.imp = TIPC_MEDIUM_IMPORTANCE;
				printf("Set importance to medium\n");
			}
			else if (strcmp(optarg, "high") == 0) {
				req.imp = TIPC_HIGH_IMPORTANCE;
				printf("Set importance to high\n");
			}
			else if (strcmp(optarg, "critical") == 0) {
				req.imp = TIPC_CRITICAL_IMPORTANCE;
				printf("Set importance to critical\n");
			}
			else
				exit(-1);
		break;
		case 'b':
			flags = MSG_DONTWAIT;
			printf("Will do nonblocking send\n");
			break;
		case 'u':
			sa.addrtype = TIPC_ADDR_NAME;
			sa.addr.name.name.type = TEST_TYPE;
			sa.addr.name.name.instance = TEST_INSTANCE;
			sa.addr.name.domain = 0;
			printf("Will send unicast\n");
			break;
		case 'h':
			printf("rdm_sender - Transmit SOCK_RDM data over multicast"\
				" or unicast to {%u,%u}\n\n",
				TEST_TYPE, TEST_INSTANCE);
			printf("Parameters:\n[-m <bytes>] (message size)\n"\
				"[-n <number>] (number of messages to send)\n"\
				"[-s <usec>] (time to sleep between sends)]\n"\
				"[-i <low|medium|high|critical>] (message importance)\n"\
				"[-b] (enable MSG_DONTWAIT)\n"\
				"[-u] (send unicast instead)]\n");
		/*Fall through*/
		default:
			exit(0);
		break;
		}

	}
	printf("TIPC RDM sender started. Will send %u %u byte messages\n", req.nmsgs, req.msgsize);

	sender_thread = malloc(threads * sizeof(pthread_t));
	tmp_sender = sender_thread;
	for (i=0; i < threads; i++, tmp_sender++) {
		if (pthread_create(tmp_sender, NULL, rdm_send, &req))
			return -1;
	};
	tmp_sender = sender_thread;
	for (i=0; i < threads; i++, tmp_sender++) {
		err = pthread_join(*tmp_sender, NULL);
		if (err && err != ESRCH) {
			printf("err:%d != ESRCH:%d\n", err, ESRCH);
			diep("pthread_join");
		}
	}
	exit(0);
}
