/*
 * channel_test.c	Tests for channels
 *
 * Version:	$Id: d322ff20be747104b209c6e9b6c68872e2f19fb7 $
 *
 *   This program is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the Free Software
 *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
 *
 * @copyright 2016 Alan DeKok (aland@freeradius.org)
 */

RCSID("$Id: d322ff20be747104b209c6e9b6c68872e2f19fb7 $")

#include <freeradius-devel/io/channel.h>
#include <freeradius-devel/io/control.h>
#include <freeradius-devel/util/debug.h>
#include <freeradius-devel/util/syserror.h>
#include <freeradius-devel/util/talloc.h>

#ifdef HAVE_GETOPT_H
#  include <getopt.h>
#endif

#include <pthread.h>
#include <sys/event.h>

#define MAX_MESSAGES		(2048)
#define MAX_CONTROL_PLANE	(1024)
#define MAX_KEVENTS		(10)

#define MPRINT1 if (debug_lvl) printf
#define MPRINT2 if (debug_lvl > 1) printf

static int			debug_lvl = 0;
static int			kq_master, kq_worker;
static fr_atomic_queue_t	*aq_master, *aq_worker;
static fr_control_t		*control_master, *control_worker;
static int			max_messages = 10;
static int			max_control_plane = 0;
static int			max_outstanding = 1;
static bool			touch_memory = false;

/**********************************************************************/
typedef struct request_s request_t;

request_t *request_alloc(UNUSED TALLOC_CTX *ctx, UNUSED request_init_args_t const *args)
{
	return NULL;
}

void request_verify(UNUSED char const *file, UNUSED int line, UNUSED request_t const *request)
{
}

/**********************************************************************/

static NEVER_RETURNS void usage(void)
{
	fprintf(stderr, "usage: channel_test [OPTS]\n");
	fprintf(stderr, "  -c <control-plane>     Size of the control plane queue.\n");
	fprintf(stderr, "  -m <messages>	  Send number of messages.\n");
	fprintf(stderr, "  -o <outstanding>       Keep number of messages outstanding.\n");
	fprintf(stderr, "  -t                     Touch memory for fake packets.\n");
	fprintf(stderr, "  -x                     Debugging mode.\n");

	fr_exit_now(EXIT_FAILURE);
}

static void *channel_master(void *arg)
{
	bool			running, signaled_close;
	int			rcode, i, num_events;
	int			num_outstanding, num_messages;
	int			num_replies;
	fr_message_set_t	*ms;
	TALLOC_CTX		*ctx;
	fr_channel_t		*channel = arg;
	fr_channel_t		*new_channel;
	fr_channel_event_t	ce;
	struct kevent		events[MAX_KEVENTS];

	MEM(ctx = talloc_init_const("channel_master"));

	ms = fr_message_set_create(ctx, MAX_MESSAGES, sizeof(fr_channel_data_t), MAX_MESSAGES * 1024);
	if (!ms) {
		fprintf(stderr, "Failed creating message set\n");
		fr_exit_now(EXIT_FAILURE);
	}

	MPRINT1("Master started.\n");

	/*
	 *	Signal the worker that the channel is open
	 */
	rcode = fr_channel_signal_open(channel);
	if (rcode < 0) {
		fprintf(stderr, "Failed signaling open: %s\n", fr_syserror(errno));
		fr_exit_now(EXIT_FAILURE);
	}

	/*
	 *	Bootstrap the queue with messages.
	 */
	num_replies = num_outstanding = num_messages = 0;

	running = true;
	signaled_close = false;

	while (running) {
		fr_time_t now;
		int num_to_send;
		fr_channel_data_t *cd, *reply;

#if 0
		/*
		 *	Drain the input queues before sleeping.
		 */
		while ((reply = fr_channel_recv_reply(channel)) != NULL) {
			num_replies++;
			num_outstanding--;
			MPRINT1("Master got reply %d, outstanding=%d, %d/%d sent.\n",
				num_replies, num_outstanding, num_messages, max_messages);
			fr_message_done(&reply->m);
		}
#endif

		/*
		 *	Ensure we have outstanding messages.
		 */
		if (num_messages >= max_messages) {
			MPRINT1("Master DONE sending\n");
			goto check_close;
		}

		num_to_send = max_outstanding - num_outstanding;
		if ((num_messages + num_to_send) > max_messages) {
			num_to_send = max_messages - num_messages;
		}
		MPRINT1("Master sending %d messages\n", num_to_send);

		for (i = 0; i < num_to_send; i++) {
			cd = (fr_channel_data_t *) fr_message_alloc(ms, NULL, 100);
			fr_assert(cd != NULL);

			num_outstanding++;
			num_messages++;

			cd->m.when = fr_time();

			if (touch_memory) {
				size_t j, k;

				for (j = k = 0; j < cd->m.data_size; j++) {
					k += cd->m.data[j];
				}

				cd->m.data[4] = k;
			}

			memcpy(cd->m.data, &num_messages, sizeof(num_messages));

			MPRINT1("Master sent message %d\n", num_messages);
			rcode = fr_channel_send_request(channel, cd, &reply);
			if (rcode < 0) {
				fprintf(stderr, "Failed sending request: %s\n", fr_syserror(errno));
			}
			fr_assert(rcode == 0);
			if (reply) {
				num_replies++;
				num_outstanding--;
				MPRINT1("Master got reply %d, outstanding=%d, %d/%d sent.\n",
					num_replies, num_outstanding, num_messages, max_messages);
				fr_message_done(&reply->m);
			}
		}

		/*
		 *	Signal close only when done.
		 */
check_close:

		if (!signaled_close && (num_messages >= max_messages) && (num_outstanding == 0)) {
			MPRINT1("Master signaling worker to exit.\n");
			rcode = fr_channel_signal_responder_close(channel);
			if (rcode < 0) {
				fprintf(stderr, "Failed signaling close: %s\n", fr_syserror(errno));
				fr_exit_now(EXIT_FAILURE);
			}

			signaled_close = true;
		}

		MPRINT1("Master waiting on events.\n");
		fr_assert(num_messages <= max_messages);

		num_events = kevent(kq_master, NULL, 0, events, MAX_KEVENTS, NULL);
		MPRINT1("Master kevent returned %d\n", num_events);

		if (num_events < 0) {
			if (num_events == EINTR) continue;

			fprintf(stderr, "Failed waiting for kevent: %s\n", fr_syserror(errno));
			fr_exit_now(EXIT_FAILURE);
		}

		if (num_events == 0) continue;

		/*
		 *	Service the events.
		 */
		for (i = 0; i < num_events; i++) {
			(void) fr_channel_service_kevent(channel, control_master, &events[i]);
		}

		now = fr_time();

		MPRINT1("Master servicing control-plane aq %p\n", aq_master);

		while (true) {
			uint32_t id;
			size_t data_size;
			char data[256];

			data_size = fr_control_message_pop(aq_master, &id, data, sizeof(data));
			if (!data_size) break;

			fr_assert(id == FR_CONTROL_ID_CHANNEL);

			ce = fr_channel_service_message(now, &new_channel, data, data_size);
			MPRINT1("Master got channel event %d\n", ce);

			switch (ce) {
			case FR_CHANNEL_DATA_READY_REQUESTOR:
				MPRINT1("Master got data ready signal\n");
				fr_assert(new_channel == channel);

				reply = fr_channel_recv_reply(channel);
				if (!reply) {
					MPRINT1("Master SIGNAL WITH NO DATA!\n");
					continue;
				}

				do {
					num_replies++;
					num_outstanding--;
					MPRINT1("Master got reply %d, outstanding=%d, %d/%d sent.\n",
						num_replies, num_outstanding, num_messages, max_messages);
					fr_message_done(&reply->m);
				} while ((reply = fr_channel_recv_reply(channel)) != NULL);
				break;

			case FR_CHANNEL_CLOSE:
				MPRINT1("Master received close signal\n");
				fr_assert(new_channel == channel);
				fr_assert(signaled_close == true);
				running = false;
				break;

			case FR_CHANNEL_NOOP:
				MPRINT1("Master got NOOP\n");
				break;

			default:
				fprintf(stderr, "Master got unexpected CE %d\n", ce);

				/*
				 *	Not written yet!
				 */
				fr_assert(0 == 1);
				break;
			} /* switch over signal returned */
		} /* drain the control plane */
	} /* loop until told to exit */

	MPRINT1("Master exiting.\n");

	/*
	 *	Force all messages to be garbage collected
	 */
	MPRINT2("GC\n");
	fr_message_set_gc(ms);

	if (debug_lvl > 1) fr_message_set_debug(ms, stdout);

	/*
	 *	After the garbage collection, all messages marked "done" MUST also be marked "free".
	 */
	rcode = fr_message_set_messages_used(ms);
	MPRINT2("Master messages used = %d\n", rcode);
	fr_assert(rcode == 0);

	talloc_free(ctx);

	return NULL;
}

static void *channel_worker(void *arg)
{
	bool running = true;
	int rcode, num_events;
	int worker_messages = 0;
	fr_message_set_t *ms;
	TALLOC_CTX *ctx;
	fr_channel_t *channel = arg;
	fr_channel_event_t ce;
	struct kevent events[MAX_KEVENTS];

	MEM(ctx = talloc_init_const("channel_worker"));

	ms = fr_message_set_create(ctx, MAX_MESSAGES, sizeof(fr_channel_data_t), MAX_MESSAGES * 1024);
	if (!ms) {
		fprintf(stderr, "Failed creating message set\n");
		fr_exit_now(EXIT_FAILURE);
	}

	MPRINT1("\tWorker started.\n");

	while (running) {
		int i;
		fr_time_t now;
		fr_channel_t *new_channel;

		MPRINT1("\tWorker waiting on events.\n");

		num_events = kevent(kq_worker, NULL, 0, events, MAX_KEVENTS, NULL);
		MPRINT1("\tWorker kevent returned %d events\n", num_events);

		if (num_events < 0) {
			if (errno == EINTR) continue;

			fprintf(stderr, "Failed waiting for kevent: %s\n", fr_syserror(errno));
			fr_exit_now(EXIT_FAILURE);
		}

		if (num_events == 0) continue;

		for (i = 0; i < num_events; i++) {
			(void) fr_channel_service_kevent(channel, control_worker, &events[i]);
		}

		MPRINT1("\tWorker servicing control-plane aq %p\n", aq_worker);

		now = fr_time();

		while (true) {
			uint32_t id;
			size_t data_size;
			char data[256];
			fr_channel_data_t *cd, *reply;

			data_size = fr_control_message_pop(aq_worker, &id, data, sizeof(data));
			if (!data_size) break;

			fr_assert(id == FR_CONTROL_ID_CHANNEL);

			ce = fr_channel_service_message(now, &new_channel, data, data_size);
			MPRINT1("\tWorker got channel event %d\n", ce);

			switch (ce) {

			case FR_CHANNEL_OPEN:
				MPRINT1("\tWorker received a new channel\n");
				fr_assert(new_channel == channel);
				break;

			case FR_CHANNEL_CLOSE:
				MPRINT1("\tWorker requested to close the channel.\n");
				fr_assert(new_channel == channel);
				running = false;

				/*
				 *	Drain the input before we ACK the exit.
				 */
				while ((cd = fr_channel_recv_request(channel)) != NULL) {
					worker_messages++;
					MPRINT1("\tWorker got message %d\n", worker_messages);
					fr_message_done(&cd->m);
				}

				(void) fr_channel_responder_ack_close(channel);
				break;

			case FR_CHANNEL_DATA_READY_RESPONDER:
				MPRINT1("\tWorker got data ready signal\n");
				fr_assert(new_channel == channel);

				cd = fr_channel_recv_request(channel);
				if (!cd) {
					MPRINT1("\tWorker SIGNAL WITH NO DATA!\n");
					break;
				}

				while (cd) {
					int message_id;

					worker_messages++;

					fr_assert(cd->m.data != NULL);
					memcpy(&message_id, cd->m.data, sizeof(message_id));
					MPRINT1("\tWorker got message %d (says %d)\n", worker_messages, message_id);

					reply = (fr_channel_data_t *) fr_message_alloc(ms, NULL, 100);
					fr_assert(reply != NULL);

					reply->m.when = fr_time();
					fr_message_done(&cd->m);

					if (touch_memory) {
						size_t j, k;

						for (j = k = 0; j < reply->m.data_size; j++) {
							k += reply->m.data[j];
						}

						reply->m.data[4] = k;
					}


					MPRINT1("\tWorker sending reply to messages %d\n", worker_messages);
					rcode = fr_channel_send_reply(channel, reply, &cd);
					if (rcode < 0) {
						fprintf(stderr, "Failed sending reply: %s\n", fr_syserror(errno));
					}
					fr_assert(rcode == 0);
				}
				break;

			case FR_CHANNEL_NOOP:
				MPRINT1("\tWorker got NOOP\n");
				fr_assert(new_channel == channel);
				break;

			default:
				fprintf(stderr, "\tWorker got unexpected CE %d\n", ce);

				/*
				 *	Not written yet!
				 */
				fr_assert(0 == 1);
				break;
			} /* switch over signals */

			/*
			 *	Get a new idea of "now".
			 */
			now = fr_time();
		} /* drain the control plane */
	}

	MPRINT1("\tWorker exiting.\n");

	/*
	 *	Force all messages to be garbage collected
	 */
	MPRINT2("Worker GC\n");
	fr_message_set_gc(ms);

	if (debug_lvl > 1) fr_message_set_debug(ms, stdout);

	/*
	 *	After the garbage collection, all messages marked "done" MUST also be marked "free".
	 */
	rcode = fr_message_set_messages_used(ms);
	fr_cond_assert(rcode == 0);

	talloc_free(ctx);

	return NULL;
}



int main(int argc, char *argv[])
{
	int			c;
	fr_channel_t		*channel;
	TALLOC_CTX		*autofree = talloc_autofree_context();
	pthread_attr_t		attr;
	pthread_t		master_id, worker_id;

	fr_time_start();

	while ((c = getopt(argc, argv, "c:hm:o:tx")) != -1) switch (c) {
		case 'x':
			debug_lvl++;
			break;

		case 'c':
			max_control_plane = atoi(optarg);
			break;

		case 'm':
			max_messages = atoi(optarg);
			break;

		case 'o':
			max_outstanding = atoi(optarg);
			break;

		case 't':
			touch_memory = true;
			break;

		case 'h':
		default:
			usage();
	}

	if (max_outstanding > max_messages) max_outstanding = max_messages;

	if (!max_control_plane) {
		max_control_plane = MAX_CONTROL_PLANE;
		if (max_outstanding > max_control_plane) max_control_plane = max_outstanding;
	}

#if 0
	argc -= (optind - 1);
	argv += (optind - 1);
#endif

	kq_master = kqueue();
	fr_assert(kq_master >= 0);

	kq_worker = kqueue();
	fr_assert(kq_worker >= 0);

	aq_master = fr_atomic_queue_alloc(autofree, max_control_plane);
	fr_assert(aq_master != NULL);

	aq_worker = fr_atomic_queue_alloc(autofree, max_control_plane);
	fr_assert(aq_worker != NULL);

	control_master = fr_control_create(autofree, kq_master, aq_master, 1024);
	fr_assert(control_master != NULL);

	control_worker = fr_control_create(autofree, kq_worker, aq_worker, 1025);
	fr_assert(control_worker != NULL);

	channel = fr_channel_create(autofree, control_master, control_worker, false);
	if (!channel) {
		fprintf(stderr, "channel_test: Failed to create channel\n");
		fr_exit_now(EXIT_FAILURE);
	}

	/*
	 *	Start the two threads, with the channel.
	 */
	(void) pthread_attr_init(&attr);
	(void) pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

	(void) pthread_create(&master_id, &attr, channel_master, channel);
	(void) pthread_create(&worker_id, &attr, channel_worker, channel);

	(void) pthread_join(master_id, NULL);
	(void) pthread_join(worker_id, NULL);

	close(kq_master);
	close(kq_worker);

	fr_channel_debug(channel, stdout);

	fr_exit_now(EXIT_SUCCESS);
}
