/* ************************************************************************** */
/* **************** library initialization/deinitialization ***************** */
/* ************************************************************************** */

#include <malloc.h>
#include <string.h>
#include "MPSClient_defs.h"

#define DEFINE_VARIABLES
	#include "MPSClient_vars.h"
#undef DEFINE_VARIABLES

void abort_clients(int num_clients, u_long *clients, u_long my_index)
{
	int client_index;
	u_long mid;

	if ( my_index == 0 )
	{
/* only parent process */
		for ( client_index = 1; client_index < num_clients; client_index++ )
		{
			if ( mps_create("ABORT", 6, &mid) )
			{
				continue;
			}

			if ( mps_send(mid, 1, &(clients[client_index])) )
			{
				continue;
			}
		}
	}

	free(clients);
}

void disconnect_from_server()
{
	shutdown(conn_socket, SHUT_RDWR);
	closesocket(conn_socket);
	conn_socket = 0;
}

int connect_to_server(u_long rpid)
{
	u_char OpTp;
	u_long ConnectString;
	u_long ClientID;
	int Error, select_return;
	fd_set read_sockets, except_sockets;

	OpTp = OpTp_CONNECT;
	ConnectString = CONNECTION_STRING;

/* Create the socket. */
	conn_socket = socket(PF_INET, SOCK_STREAM, 0);
	if ( conn_socket == -1 )
	{
		conn_socket = 0;
		return -1;
	}

/* Give the socket a name. */
	conn_name.sin_family = AF_INET;
	conn_name.sin_port = htons(conn_port);

	hostinfo = gethostbyname("localhost");
	if ( hostinfo == NULL )
	{
		closesocket(conn_socket);
		conn_socket = 0;
		return -2;
	}
	conn_name.sin_addr = *(struct in_addr *) hostinfo->h_addr;

/* Connect to server. */
	if ( connect (conn_socket, (struct sockaddr *) (&conn_name), sizeof(conn_name)) )
	{
		closesocket(conn_socket);
		conn_socket = 0;
		return -3;
	}


/* Send connection sequence. */
	Error = 1;
	if ( td_write_u_char(conn_socket, &OpTp, &td_buffer) )
	{
		if ( td_write_u_long(conn_socket, &ConnectString, &td_buffer) )
		{
			if ( td_write_u_long(conn_socket, &rpid, &td_buffer) )
			{
				if ( td_flush(conn_socket, &td_buffer) )
				{
					Error = 0;
				}
			}
		}
	}

	if ( Error )
	{
		disconnect_from_server();
		return -5;
	}

/* Wait for connection response. */

	FD_ZERO(&read_sockets);
	FD_SET(conn_socket, &read_sockets);
	except_sockets = read_sockets;

	select_return = select(conn_socket + 1, &read_sockets, NULL, &except_sockets, NULL);

	if ( select_return == -1 ||
		FD_ISSET(conn_socket, &except_sockets)
		)
	{
		disconnect_from_server();
		return -6;
	}

/* Read connection response. */
	if ( !td_read_u_char(conn_socket, &OpTp) )
	{
		disconnect_from_server();
		return -7;
	}

	if ( OpTp != OpTp_ACCEPT_CONNECT )
	{
		disconnect_from_server();
		return -8;
	}

	if ( !td_read_u_long(conn_socket, &ClientID) )
	{
		disconnect_from_server();
		return -9;
	}

	if ( ClientID != rpid && rpid != 0 )
	{
		disconnect_from_server();
		return -10;
	}

	pid = ClientID;

	return 0;
}

int create_thread()
{
	pthread_attr_t thread_attr;
	pthread_t thread;

/* prepare thread attributes */
	if ( pthread_attr_init(&thread_attr) )
	{
		return -1;
	}

	pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED);

	pthread_mutex_lock(&thread_mutex);

/* create thread */
	if ( pthread_create(&thread, &thread_attr, &client_thread, NULL) )
	{
		pthread_attr_destroy(&thread_attr);
		pthread_mutex_unlock(&thread_mutex);

		return -2;
	}

	thread_created = 1;

	pthread_attr_destroy(&thread_attr);
	pthread_mutex_unlock(&thread_mutex);

	return 0;
}

void delete_thread()
{
	pthread_mutex_lock(&thread_mutex);

	if ( thread_created == 0 )
	{
		pthread_mutex_unlock(&thread_mutex);
		return;
	}

	exit_semaphore_waiting = 1;
	exit_client = 1;

	pthread_mutex_unlock(&thread_mutex);

	sem_wait(&exit_semaphore);

/* wait threads exit */
	pthread_mutex_lock(&thread_mutex);
	pthread_mutex_unlock(&thread_mutex);

	thread_created = 0;
}

void deinitialize_client()
{
	if (pid > 1 &&
		thread_created != 0 &&
		conn_socket != 0)
	{
		mps_end();
	}

/* delete thread */
	if ( thread_created != 0 )
	{
		delete_thread();
	}

/* disconnect from server */
	if ( conn_socket != 0 )
	{
		disconnect_from_server();
	}

#if defined(__MINGW32__)
	WSACleanup( );
#endif

	delete_thread_semaphores();
	sem_destroy(&exit_semaphore);
	pthread_mutex_destroy(&send_mutex);
	pthread_mutex_destroy(&thread_mutex);

	if ( td_buffer.buf != NULL )
	{
		free(td_buffer.buf);
	}
	td_buffer.pos = 0;
	td_buffer.size = 0;

	pid = 0;
}

u_long getmyid()
{
	return pid;
}

int initialize_client(u_long rpid)
{
/* initialize values */
	conn_port = 19792;
	conn_socket = 0;
	exit_client = 0;
	exit_semaphore_waiting = 0;
	pid = 0;
	thread_created = 0;
	thread_semaphores = NULL;

	td_buffer.pos = 0;
	td_buffer.size = 1460;
	td_buffer.buf = malloc(td_buffer.size);

	if ( td_buffer.buf == NULL )
	{
		return -1;
	}

	if ( pthread_mutex_init(&thread_mutex, NULL) )
	{
		free(td_buffer.buf);
		return -2;
	}

	if ( pthread_mutex_init(&send_mutex, NULL) )
	{
		free(td_buffer.buf);
		return -3;
	}

	if ( sem_init(&exit_semaphore, 0, 0) )
	{
		pthread_mutex_destroy(&thread_mutex);

		free(td_buffer.buf);
		return -4;
	}

	if ( create_thread_semaphore() == NULL )
	{
		pthread_mutex_destroy(&thread_mutex);

		free(td_buffer.buf);
		return -5;
	}

#if defined(__MINGW32__)
	wVersionRequested = MAKEWORD( 1, 0 );
	WSAStartup( wVersionRequested, &wsaData );
#endif

/* connect to server */
	if ( connect_to_server(rpid) )
	{
		deinitialize_client();

		return -6;
	}

/* create thread */
	if ( create_thread() )
	{
		deinitialize_client();

		return -7;
	}

	return 0;
}

int initialize_clients(int num_clients, u_long **clients_ptr, char *argv[])
{
	int ret, client_index, start;
	u_long *clients, mid, mid_init, sender_pid, my_index, msglen;
	char *msg;
	pid_t new_pid, *proc_pids;
#if !defined(__MINGW32__)
	struct timeval waittime;
#endif
#if !defined(USE_FORK)
	u_long myid;
	char myname[512];
	char *client_list, *next;

	client_list = NULL;
#endif

	proc_pids = NULL;
	*clients_ptr = NULL;

	if ( num_clients <= 0 ) return 0;

	ret = initialize_client(1);
#if !defined(USE_FORK)
	if ( ret && num_clients != 1 )
	{
		ret = initialize_client(0);
	}
#endif
	if ( ret )
	{
		return 0;
	}

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

	if ( clients == NULL )
	{
		deinitialize_client();
		return 0;
	}

	if ( num_clients == 1 )
	{
		clients[0] = getmyid();
		*clients_ptr = clients;
		return 1;
	}

#if defined(USE_FORK)
	clients[0] = getmyid();
#else
	clients[0] = 1;

	myid = getmyid();

	if ( myid == 1 )
#endif
	{
#if defined(USE_FORK)
		if ( mps_create("INIT-POSIX", 11, &mid_init) )
#else
		if ( mps_create("INIT-MINGW", 11, &mid_init) )
#endif
		{
			free(clients);
			deinitialize_client();
			return 0;
		}

		for ( client_index = 1; client_index < num_clients; client_index++ )
		{
			int aaa;
			if ( (aaa = mps_start(&msg, &msglen, &mid)) )
			{
				free(clients);
				deinitialize_client();
				return 0;
			}

			if ( msglen == 0 ||
				strncmp(msg, "NEW_ID", 6) != 0 )
			{
				mps_destroy(mid);
				free(msg);
				free(clients);
				deinitialize_client();
				return 0;
			}

			clients[client_index] = 0;
			sscanf(msg, "NEW_ID=%u", (unsigned int *) &(clients[client_index]));

			if ( mps_destroy(mid) )
			{
				free(msg);
				free(clients);
				deinitialize_client();
				return 0;
			}
			free(msg);

			if ( clients[client_index] == 0 )
			{
				free(clients);
				deinitialize_client();
				return 0;
			}
		}

		if ( mps_destroy(mid_init) )
		{
			free(clients);
			deinitialize_client();
			return 0;
		}

		proc_pids = (pid_t *) malloc(num_clients * sizeof(pid_t));

		if ( proc_pids == NULL )
		{
			free(clients);
			deinitialize_client();
			return 0;
		}
		memset(proc_pids, 0, num_clients * sizeof(pid_t));
		*clients_ptr = (u_long *) proc_pids;

		proc_pids[0] = getpid();

#if defined(USE_FORK)
		deinitialize_client();

		my_index = 0;
		for ( client_index = 1; client_index < num_clients; client_index++ )
		{
			new_pid = fork();

			if ( new_pid == -1 )
			{
				free(clients);
				return 0;
			}

			if ( new_pid == 0 )
			{
				my_index = client_index;
				break;
			}
			else
			{
				proc_pids[client_index] = new_pid;
			}
		}
#else
		my_index = 0;

		if ( argv[0][0] == '/' || argv[0][0] == '\\' || argv[0][1] == ':')
		{
			strncpy(myname, argv[0], 511);
			myname[511] = 0;
		}
		else
		{
			getcwd(myname, 512);
			strcat(myname, "/");
			strcat(myname, argv[0]);
		}

		for ( client_index = 1; client_index < num_clients; client_index++ )
		{
			new_pid = spawnv(_P_NOWAIT, myname, (const char**) argv);

			if ( new_pid == -1 )
			{
				free(clients);
				return 0;
			}
			else
			{
				proc_pids[client_index] = new_pid;
			}
		}

#endif

	}
#if !defined(USE_FORK)
	else
	{
		my_index = num_clients + 1;
	}
#endif

#if defined(USE_FORK)
	if ( initialize_client(clients[my_index]) )
	{
		free(clients);
		return 0;
	}
#else
	if ( myid == 1 )
	{
		client_list = (char *) malloc((num_clients - 1) * 12);
		if ( client_list == NULL )
		{
			abort_clients(num_clients, clients, my_index);
			deinitialize_client();
			return 0;
		}
	}
#endif

	if ( mps_create("SYNC-INIT", 10, &mid_init) )
	{
		abort_clients(num_clients, clients, my_index);
		deinitialize_client();
		return 0;
	}

	start = 0;
	if ( my_index == 0 )
	{
/* only parent process */
		if ( mps_create("START", 6, &mid) )
		{
			abort_clients(num_clients, clients, my_index);
			deinitialize_client();
			return 0;
		}

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

/* parent and child process */
	while ( start == 0 )
	{
		if ( mps_recv(&msg, &msglen, &mid, &sender_pid, ( (my_index == 0)?num_clients:1 ), &(clients[0]), NULL) )
		{
			abort_clients(num_clients, clients, my_index);
			deinitialize_client();
			return 0;
		}

		if ( msglen == 0)
		{
			mps_destroy(mid);
			free(msg);
			abort_clients(num_clients, clients, my_index);
			deinitialize_client();
			return 0;
		}
		else if ( strcmp(msg, "START") == 0 )
		{
			start = 1;
		}
		else if ( strncmp(msg, "SENDER_NOT_FOUND=", 17) != 0 )
		{
			mps_destroy(mid);
			free(msg);
			abort_clients(num_clients, clients, my_index);
			deinitialize_client();
			return 0;
		}

		if ( mps_destroy(mid) )
		{
			free(msg);
			abort_clients(num_clients, clients, my_index);
			deinitialize_client();
			return 0;
		}

		free(msg);

		if ( start == 0)
		{
#if defined(__MINGW32__)
			Sleep(100);
#else
			waittime.tv_sec = 0;
			waittime.tv_usec = 100000;
			select(0, NULL, NULL, NULL, &waittime);
#endif
		}
	}

	if ( my_index == 0 )
	{
/* only parent process */
		for ( client_index = 1; client_index < num_clients; client_index++ )
		{
			if ( mps_create("START", 6, &mid) )
			{
				abort_clients(num_clients, clients, my_index);
				deinitialize_client();
				return 0;
			}

			if ( mps_send(mid, 1, &(clients[client_index])) )
			{
				abort_clients(num_clients, clients, my_index);
				deinitialize_client();
				return 0;
			}
		}
	}

#if !defined(USE_FORK)
/* distribute ids */
	if ( myid == 1 )
	{
/* parent process */
		client_list[0] = 0;
		for ( client_index = 1; client_index < num_clients; client_index++ )
		{
			if ( client_index != 1 )
			{
				strcat(client_list, ",");
			}
			sprintf(myname, "%lu", clients[client_index]);
			strcat(client_list, myname);
		}

		for ( client_index = 1; client_index < num_clients; client_index++ )
		{
			if ( mps_create(client_list, strlen(client_list) + 1, &mid) )
			{
				free(client_list);
				abort_clients(num_clients, clients, my_index);
				deinitialize_client();
				return 0;
			}

			if ( mps_send_async(mid, 1, &(clients[client_index])) )
			{
				free(client_list);
				abort_clients(num_clients, clients, my_index);
				deinitialize_client();
				return 0;
			}
		}

		free(client_list);
	}
	else
	{
/* child process */
		if ( mps_recv(&msg, &msglen, &mid, &sender_pid, 1 , &(clients[0]), NULL) )
		{
			abort_clients(num_clients, clients, my_index);
			deinitialize_client();
			return 0;
		}

		if ( msglen == 0 ||
			strcmp(msg, "ABORT") == 0 ||
			strncmp(msg, "SENDER_NOT_FOUND=", 17) == 0
			)
		{
			mps_destroy(mid);
			free(msg);
			abort_clients(num_clients, clients, my_index);
			deinitialize_client();
			return 0;
		}

		client_index = 1;
		client_list = msg;
		next = msg;

		while ( client_index < num_clients && next!= NULL)
		{
			next = strchr(client_list, ',');

			if ( next != NULL )
			{
				*next = 0;
			}

			sscanf(client_list, "%lu", &(clients[client_index]));

			if ( clients[client_index] == myid )
			{
				my_index = client_index;
			}

			if ( next != NULL )
			{
				client_list = next + 1;
			}

			client_index++;
		}

		if ( mps_destroy(mid) )
		{
			free(msg);
			abort_clients(num_clients, clients, my_index);
			deinitialize_client();
			return 0;
		}

		free(msg);
	}
#endif

	if ( mps_destroy(mid_init) )
	{
		abort_clients(num_clients, clients, my_index);
		deinitialize_client();
		return 0;
	}

	*clients_ptr = clients;

	if (proc_pids != NULL)
	{
		free(proc_pids);
	}

	return my_index + 1;
}
