#include <malloc.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <signal.h>
#include <unistd.h>

#if defined(__MINGW32__)
	#include <winsock2.h>
#else
	#include <netinet/in.h>
#endif

#include "tpl.h"
#include "MPSClient_lib.h"

struct _tpl_message_ {
	void *addr;
	u_long length, pos;
} ;

typedef struct _tpl_message_ tpl_message;

struct _tpl_function_ {
	MATCHING_FUNC *addr;
	char *name;
	struct _tpl_function_ *next;
} ;

typedef struct _tpl_function_ tpl_function;


u_long num_clients, *clients = NULL;
tpl_function *functions = NULL;

void barrier_central()
{
	char *msg;
	int *waiting_status;
	int pos, pos2;
	u_long *connected_clients;
	u_long num_not_waiting, mid, msglen, sender_pid;

	waiting_status = (int *) malloc(num_clients * sizeof(int));

	if (waiting_status == NULL) return;

	connected_clients = (u_long *) malloc(num_clients * sizeof(u_long));

	if (connected_clients == NULL)
	{
		free(waiting_status);
		return;
	}

	for ( pos = 0; pos < num_clients; pos++ )
	{
		waiting_status[pos] = 0; /* 0 = not waiting */
	}

	waiting_status[0] = 1; /* 1 = waiting */
	num_not_waiting = num_clients - 1;

	/* wait for other clients to reach the barrier */
	while ( num_not_waiting != 0 )
	{
		pos2 = 0;
		for ( pos = 1; pos < num_clients; pos++ )
		{
			if (waiting_status[pos] == 0)
			{
				connected_clients[pos2] = clients[pos];
				pos2++;
			}
		}


		if ( mps_recv(&msg, &msglen, &mid, &sender_pid, pos2, &(connected_clients[0]), NULL) )
		{
			free(connected_clients);
			free(waiting_status);
			return;
		}

		if ( msglen == 0)
		{
		}
		else if ( strcmp(msg, "REQUEST_FINISH") == 0 )
		{
			for ( pos = 1; pos < num_clients; pos++ )
			{
				if ( clients[pos] == sender_pid )
				{
					if ( waiting_status[pos] == 0 ) num_not_waiting--;
					waiting_status[pos] = 1; /* 1 = not waiting */
					break;
				}
			}
		}
		else if ( strncmp(msg, "SENDER_NOT_FOUND=", 17) == 0 )
		{
			for ( pos = 1; pos < num_clients; pos++ )
			{
				if ( clients[pos] == sender_pid )
				{
					if ( waiting_status[pos] == 0 ) num_not_waiting--;
					waiting_status[pos] = 2; /* 2 = disconnected */
					break;
				}
			}
		}

		if ( mps_destroy(mid) )
		{
			if ( msg != NULL ) free(msg);
			free(connected_clients);
			free(waiting_status);
			return;
		}

		if ( msg != NULL ) free(msg);

	}

	/* signalize other clients to continue past the barrier */
	for ( pos = 1; pos < num_clients; pos++ )
	{
		if (waiting_status[pos] == 1)
		{
			if ( mps_create("FINISH", 7, &mid) )
			{
				free(connected_clients);
				free(waiting_status);
				return;
			}

			if ( mps_send(mid, 1, &(clients[pos])) )
			{
				free(connected_clients);
				free(waiting_status);
				return;
			}
		}
	}

	free(connected_clients);
	free(waiting_status);
}

void barrier_other()
{
	char *msg;
	u_long mid, msglen, sender_pid;
	int finish;

	finish = 0;

	if ( mps_create("REQUEST_FINISH", 15, &mid) )
	{
		return;
	}

	if ( mps_send_async(mid, 1, &(clients[0])) )
	{
		return;
	}

	while ( finish == 0 )
	{
		if ( mps_recv(&msg, &msglen, &mid, &sender_pid, 1 , &(clients[0]), NULL) )
		{
			return;
		}

		if ( msglen == 0)
		{
		}
		else if ( strcmp(msg, "FINISH") == 0 )
		{
			finish = 1;
		}
		else if ( strncmp(msg, "SENDER_NOT_FOUND=", 17) == 0 )
		{
			mps_destroy(mid);
			free(msg);
			return;
		}

		if ( mps_destroy(mid) )
		{
			if ( msg != NULL ) free(msg);
			return;
		}

		if ( msg != NULL ) free(msg);
	}

}

void delete_functions(tpl_function *f)
{
	tpl_function *ft;

	while ( f != NULL )
	{
		ft = f;
		f = f->next;
		free(ft);
	}
}

void tpl_initialize(int *nr_tasks, int *rank, int *argc, char ***argv)
{
	pid_t mypid, *proc_pids;
	int ret, pos;
#if defined(__MINGW32__)
	HANDLE proc_handle;
#else
	struct timeval waittime;
#endif

	num_clients = *nr_tasks;
	ret = initialize_clients(num_clients, &clients, *argv);

	*rank = ret - 1;
	if ( ret == 0 )
	{
		if ( clients != NULL )
		{
			mypid = getpid();
			proc_pids = (pid_t *) clients;

			for ( pos = 0; pos < num_clients; pos++ )
			{
				if (proc_pids[pos] != 0 &&
					proc_pids[pos] != mypid)
				{
#if defined(__MINGW32__)
					proc_handle = OpenProcess(PROCESS_TERMINATE, FALSE, proc_pids[pos]);

					if ( proc_handle != NULL )
					{
						TerminateProcess(proc_handle, 1);

						CloseHandle(proc_handle);
					}
#else
					kill(proc_pids[pos], SIGTERM);
#endif
				}
			}

#if !defined(__MINGW32__)
			waittime.tv_sec = 0;
			waittime.tv_usec = 1000000;
			select(0, NULL, NULL, NULL, &waittime);

			for ( pos = 0; pos < num_clients; pos++ )
			{
				if (proc_pids[pos] != 0 &&
					proc_pids[pos] != mypid)
				{
					kill(proc_pids[pos], SIGKILL);
				}
			}
#endif
			free(clients);
		}

		exit(1);
	}
}

void tpl_deinitialize(void)
{
	u_long mid_sync;

	if ( functions != NULL )
	{
		delete_functions(functions);
		functions = NULL;
	}

	if ( clients != NULL )
	{
		if ( mps_create("SYNC-END", 9, &mid_sync) )
		{
			mid_sync = 0;
		}

		/* barrier synchronization */
		if ( getmyid() == clients[0] )
		{
			/* central process */
			barrier_central();
		}
		else
		{
			/* other processes */
			barrier_other();
		}

		if ( mid_sync != 0 )
		{
			mps_destroy(mid_sync);
		}

		free(clients);
		clients = NULL;
	}

	deinitialize_client();
}

void tpl_register_function(MATCHING_FUNC *function, char *function_name)
{
	tpl_function *f;

	f = (tpl_function *) malloc(sizeof(tpl_function));

	if ( f == NULL ) return;

	f->addr = function;
	f->name = strdup(function_name);
	f->next = functions;
	functions = f;
}

void tpl_send(int *recipients, int nr_recipients, int tag, void *message)
{
	int rec;
	u_long mid, client_pid;

	if ( message == NULL ) return;

	if ( ((tpl_message *) message)->length >= 4 &&
		((tpl_message *) message)->addr != NULL
		)
	{
		*((u_long *) (((tpl_message *) message)->addr)) = htonl(tag);
	}


	for ( rec = 0; rec < nr_recipients; rec++ )
	{
		if ( mps_create(((tpl_message *) message)->addr, ((tpl_message *) message)->length, &mid) )
		{
			continue;
		}

		if (recipients[rec] < 0)
		{
			client_pid = (u_long) ( recipients[rec] );
		}
		else if (recipients[rec] >= num_clients )
		{
			client_pid = (u_long) ( recipients[rec] + clients[num_clients - 1] - (num_clients - 1) );
		}
		else
		{
			client_pid = clients[recipients[rec]];
		}

		if ( mps_send_async(mid, 1, &(client_pid)) )
		{
			continue;
		}
	}

	tpl_destroy(message);
}

char *find_function_name(MATCHING_FUNC *function)
{
	tpl_function *f;

	if ( function == NULL ) return NULL;

	f = functions;

	while ( f != NULL )
	{
		if ( function == f->addr ) return f->name;
		f = f->next;
	}

	return NULL;
}

static int comp_u_long(const void *m1, const void *m2)
{
	return ( *((u_long *) m1) == *((u_long *) m2) ) ? 0 : ( ( *((u_long *) m1) < *((u_long *) m2) )?-1:1  );
}

void tpl_recv(MATCHING_FUNC *match, int *sender, int *tag, void **message)
{
	tpl_message *tplmsg;
	u_long mid, msglen, pid, *pid_ptr;
	char *msg;

	if ( message == NULL ) return;

	if ( mps_recv(&msg, &msglen, &mid, &pid, num_clients, clients, find_function_name(match)) )
	{
		*sender = -1;
		*tag = -1;
		*message = NULL;
		return;
	}

	mps_destroy(mid);

	if ( msglen == 0 ||
		strncmp(msg, "SENDER_NOT_FOUND=", 17) == 0 )
	{
		free(msg);
		*sender = -1;
		*tag = -1;
		*message = NULL;
		return;
	}

	tplmsg = (tpl_message *) malloc(sizeof(tpl_message));
	if ( tplmsg == NULL )
	{
		free(msg);
		*sender = -1;
		*tag = -1;
		*message = NULL;
		return;
	}

	pid_ptr = (u_long *) bsearch(&pid, clients, num_clients, sizeof(u_long), comp_u_long);

	*sender = (pid_ptr == NULL)? -1:( pid_ptr - clients ) ;
	*tag = (msglen >= 4)?ntohl(*((u_long *) (msg))):0;
	*message = tplmsg;
	tplmsg->addr = msg;
	tplmsg->length = msglen;
	tplmsg->pos = (msglen >= 4)?4:0;
}

void tpl_create(void **message)
{
	tpl_message *msg;

	if ( message == NULL ) return;

	msg = (tpl_message *) malloc(sizeof(tpl_message));
	if ( msg == NULL )
	{
		*message = NULL;
		return;
	}

	msg->addr = malloc(4);
	if ( msg->addr == NULL )
	{
		free(msg);
		*message = NULL;
		return;
	}

	msg->length = 4;
	msg->pos = 4;
	*((u_long *) (msg->addr)) = 0;

	*message = msg;
}

void tpl_destroy(void *message)
{
	if ( message == NULL ) return;

	if ( ((tpl_message *) message)->addr != NULL )
	{
		free(((tpl_message *) message)->addr);
		((tpl_message *) message)->addr = NULL;
	}
	((tpl_message *) message)->length = 0;
	((tpl_message *) message)->pos = 0;

	free(message);
}



void tpl_pkbyte(void *message, void *p, int nitems)
{
	void *newaddr;

	if ( message == NULL ) return;

	if ( ((tpl_message *)message)->pos + nitems > ((tpl_message *)message)->length )
	{
		newaddr = realloc(((tpl_message *)message)->addr, ((tpl_message *)message)->pos + nitems);

		if ( newaddr == NULL )
		{
			return;
		}

		((tpl_message *)message)->length = ((tpl_message *)message)->pos + nitems;
		((tpl_message *)message)->addr = newaddr;
	}

	if ( p == NULL )
	{
		memset( (void *) ((u_long) (((tpl_message *)message)->addr) + ((tpl_message *)message)->pos) , 0, nitems);
	}
	else
	{
		memcpy( (void *) ((u_long) (((tpl_message *)message)->addr) + ((tpl_message *)message)->pos) , p, nitems);
	}
	((tpl_message *)message)->pos += nitems;
}

void tpl_upkbyte(void *message, void *p, int nitems)
{
	int ni1, ni2;

	if ( p == NULL ) return;

	if ( message == NULL )
	{
		ni1 = 0;
		ni2 = nitems;
	}
	else
	{
		if ( ((tpl_message *)message)->pos + nitems <= ((tpl_message *)message)->length )
		{
			ni1 = nitems;
			ni2 = 0;
		}
		else
		{
			ni1 = ((tpl_message *)message)->length - ((tpl_message *)message)->pos;
			ni2 = nitems - ni1;
		}
	}

	if ( ni1 > 0 )
	{
		memcpy( p, (void *) ((u_long) (((tpl_message *)message)->addr) + ((tpl_message *)message)->pos) , ni1);
	}

	if ( ni2 > 0 )
	{
		memset( (void *) ((u_long) (p) + ni1), 0, ni2);
	}
}

void tpl_pkchar(void *message, char *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upkchar(void *message, char *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

void tpl_pkdouble(void *message, double *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upkdouble(void *message, double *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

void tpl_pkfloat(void *message, float *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upkfloat(void *message, float *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

void tpl_pkint(void *message, int *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upkint(void *message, int *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

void tpl_pklong(void *message, long *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upklong(void *message, long *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

void tpl_pklongdouble(void *message, long double *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upklongdouble(void *message, long double *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

void tpl_pkshort(void *message, short *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upkshort(void *message, short *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

void tpl_pkuchar(void *message, unsigned char *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upkuchar(void *message, unsigned char *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

void tpl_pkuint(void *message, unsigned int *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upkuint(void *message, unsigned int *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

void tpl_pkulong(void *message, unsigned long *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upkulong(void *message, unsigned long *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

void tpl_pkushort(void *message, unsigned short *p, int nitems)
{
	tpl_pkbyte(message, p, nitems * sizeof(*p));
}

void tpl_upkushort(void *message, unsigned short *p, int nitems)
{
	tpl_upkbyte(message, p, nitems * sizeof(*p));
}

