#define __USE_GNU

#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
extern "C" {
	#include <sys/socket.h>
	#include <netinet/in.h>
}

#include <kernel.hfa>
#include <stats.hfa>
#include <time.hfa>
#include <thread.hfa>

#include "filecache.hfa"
#include "options.hfa"
#include "worker.hfa"

extern void register_fixed_files( cluster &, int *, unsigned count );

Duration default_preemption() {
	return 0;
}

//=============================================================================================
// Globals
//=============================================================================================
struct ServerProc {
	processor self;
};

void ?{}( ServerProc & this ) {
	/* paranoid */ assert( options.clopts.instance != 0p );
	(this.self){ "Benchmark Processor", *options.clopts.instance };

	#if !defined(__CFA_NO_STATISTICS__)
		if( options.clopts.procstats ) {
			print_stats_at_exit( this.self, options.clopts.instance->print_stats );
		}
		if( options.clopts.viewhalts ) {
			print_halts( this.self );
		}
	#endif
}

extern void init_protocol(void);
extern void deinit_protocol(void);

//=============================================================================================
// Main
//============================================================================================='
int main( int argc, char * argv[] ) {
	//===================
	// Parse args
	const char * path = parse_options(argc, argv);

	//===================
	// Open Files
	printf("Filling cache from %s\n", path);
	fill_cache( path );

	//===================
	// Open Socket
	printf("Listening on port %d\n", options.socket.port);
	int server_fd = socket(AF_INET, SOCK_STREAM, 0);
	if(server_fd < 0) {
		abort( "socket error: (%d) %s\n", (int)errno, strerror(errno) );
	}

	int ret = 0;
	struct sockaddr_in address;
	int addrlen = sizeof(address);
	memset( (char *)&address, '\0' );
	address.sin_family = AF_INET;
	address.sin_addr.s_addr = htonl(INADDR_ANY);
	address.sin_port = htons( options.socket.port );

	int waited = 0;
	for() {
		ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
		if(ret < 0) {
			if(errno == EADDRINUSE) {
				if(waited == 0) {
					printf("Waiting for port\n");
				} else {
					printf("\r%d", waited);
					fflush(stdout);
				}
				waited ++;
				sleep( 1`s );
				continue;
			}
			abort( "bind error: (%d) %s\n", (int)errno, strerror(errno) );
		}
		break;
	}

	ret = listen( server_fd, options.socket.backlog );
	if(ret < 0) {
		abort( "listen error: (%d) %s\n", (int)errno, strerror(errno) );
	}

	//===================
	// Run Server Cluster
	{
		cluster cl = { "Server Cluster", options.clopts.params };
		#if !defined(__CFA_NO_STATISTICS__)
			print_stats_at_exit( cl, CFA_STATS_READY_Q | CFA_STATS_IO );
		#endif
		options.clopts.instance = &cl;


		int pipe_cnt = options.clopts.nworkers * 2;
		int pipe_off;
		int * fds;
		[fds, pipe_off] = filefds( pipe_cnt );
		for(i; 0 ~ pipe_cnt ~ 2) {
			int ret = pipe(&fds[pipe_off + i]);
			if( ret < 0 ) { abort( "pipe error: (%d) %s\n", (int)errno, strerror(errno) ); }
		}

		if(options.file_cache.fixed_fds) {
			register_fixed_files(cl, fds, pipe_off);
		}

		{
			ServerProc procs[options.clopts.nprocs];

			init_protocol();
			{
				Worker workers[options.clopts.nworkers];
				for(i; options.clopts.nworkers) {
					// if( options.file_cache.fixed_fds ) {
					// 	workers[i].pipe[0] = pipe_off + (i * 2) + 0;
					// 	workers[i].pipe[1] = pipe_off + (i * 2) + 1;
					// }
					// else
					{
						workers[i].pipe[0] = fds[pipe_off + (i * 2) + 0];
						workers[i].pipe[1] = fds[pipe_off + (i * 2) + 1];
						workers[i].sockfd  = server_fd;
						workers[i].addr    = (struct sockaddr *)&address;
						workers[i].addrlen = (socklen_t*)&addrlen;
						workers[i].flags   = 0;
					}
					unpark( workers[i] );
				}
				printf("%d workers started on %d processors\n", options.clopts.nworkers, options.clopts.nprocs);
				{
					char buffer[128];
					while(!feof(stdin)) {
						fgets(buffer, 128, stdin);
					}

					printf("Shutting Down\n");
				}

				for(i; options.clopts.nworkers) {
					printf("Cancelling %p\n", (void*)workers[i].cancel.target);
					workers[i].done = true;
					cancel(workers[i].cancel);
				}

				printf("Shutting down socket\n");
				int ret = shutdown( server_fd, SHUT_RD );
				if( ret < 0 ) { abort( "shutdown error: (%d) %s\n", (int)errno, strerror(errno) ); }

				//===================
				// Close Socket
				printf("Closing Socket\n");
				ret = close( server_fd );
				if(ret < 0) {
					abort( "close socket error: (%d) %s\n", (int)errno, strerror(errno) );
				}
			}
			printf("Workers Closed\n");

			deinit_protocol();
		}

		for(i; pipe_cnt) {
			ret = close( fds[pipe_off + i] );
			if(ret < 0) {
				abort( "close pipe error: (%d) %s\n", (int)errno, strerror(errno) );
			}
		}
		free(fds);

	}

	//===================
	// Close Files
	printf("Closing Files\n");
	close_cache();
}