/* ************************************************************************** */
/* ************************ communication threads *************************** */
/* ************************************************************************** */

#include <stdlib.h>
#include <unistd.h>
#include <pthread.h>
#include <string.h>
#include "MPSServer_defs.h"
#include "MPSServer_vars.h"
#include "tpl2.h"

#define MAX_SEARCH_DATA 50

struct _search_data_ {
	u_long pid;
	int num, max;
	operation_data oids[MAX_SEARCH_DATA];
};

typedef struct _search_data_ search_data;


int ExecuteRecvFail(operation_data *recv_op);
int ExecuteSendFail(operation_data *send_op, message_data *send_message);
int WriteCreateRet(connection_data *connection, operation_data *operation);
int WriteDestroyRet(connection_data *connection, operation_data *operation);
int WriteEndRet(connection_data *connection, operation_data *operation);
int WriteRecvRet(connection_data *connection, operation_data *operation);
int WriteSendRet(connection_data *connection, operation_data *operation);
int WriteStartRet(connection_data *connection, operation_data *operation);

int CheckOperation(operation_data *op, search_data *search)
{
	int found;
	u_long process_index;

	found = 0;

	for ( process_index = 0; process_index < op->num_process_ids; process_index++ )
	{
		if ( search->pid == op->process_ids[process_index] )
		{
			op->process_id2 = search->pid;

			found = 1;
			break;
		}
	}

	if ( found == 0 ) return 0;

	search->oids[search->num] = *op;
	search->num++;

	if ( search->num == search->max )
	{
		return 1;
	}
	else
	{
		return 0;
	}
}

int CheckRecvSendOp(operation_data *recv_op, operation_data *send_op)
{
	int found;
	u_long process_index;

	tpl_message tplmsg;

	found = 0;

	for ( process_index = 0; process_index < recv_op->num_process_ids; process_index++ )
	{
		if ( send_op->process_id == recv_op->process_ids[process_index] )
		{
			found = 1;
			break;
		}
	}

	if ( found == 0 ) return 0;

	found = 0;

	for ( process_index = 0; process_index < send_op->num_process_ids; process_index++ )
	{
		if ( recv_op->process_id == send_op->process_ids[process_index] )
		{
			found = 1;
			break;
		}
	}

	if ( found == 0 ) return 0;

	if ( recv_op->filter != NULL )
	{
		switch (recv_op->ftype)
		{
			case 0:
				if ( recv_op->filter(send_op->mtext, send_op->mlength, send_op->process_id) == 0 ) return 0;
				break;
			case 1:
				tplmsg.addr = send_op->mtext;
				tplmsg.length = send_op->mlength;
				tplmsg.pos = (tplmsg.length >= 4)?4:0;

				if ( send_op->fsender == -1 )
				{
					send_op->ftag = (tplmsg.length >= 4)?ntohl(*((u_long *) (tplmsg.addr))):0;;
					send_op->fsender = process_list_FindIndex(&process_list, send_op->process_id);
				}

				if ( ((MATCHING_FUNC *)recv_op->filter)(send_op->fsender, send_op->ftag, &tplmsg) == 0 ) return 0;
				break;
		}
	}

	return 1;
}

int CheckSendRecvOp(operation_data *send_op, operation_data *recv_op)
{
	int found;
	u_long process_index;

	tpl_message tplmsg;

	found = 0;

	for ( process_index = 0; process_index < recv_op->num_process_ids; process_index++ )
	{
		if ( send_op->process_id == recv_op->process_ids[process_index] )
		{
			found = 1;
			break;
		}
	}

	if ( found == 0 ) return 0;

	found = 0;

	for ( process_index = 0; process_index < send_op->num_process_ids; process_index++ )
	{
		if ( recv_op->process_id == send_op->process_ids[process_index] )
		{
			found = 1;
			break;
		}
	}

	if ( found == 0 ) return 0;

	if ( recv_op->filter != NULL )
	{
		switch (recv_op->ftype)
		{
			case 0:
				if ( recv_op->filter(send_op->mtext, send_op->mlength, send_op->process_id) == 0 ) return 0;
				break;
			case 1:
				tplmsg.addr = send_op->mtext;
				tplmsg.length = send_op->mlength;
				tplmsg.pos = (tplmsg.length >= 4)?4:0;

				if ( send_op->fsender == -1 )
				{
					send_op->ftag = (tplmsg.length >= 4)?ntohl(*((u_long *) (tplmsg.addr))):0;;
					send_op->fsender = process_list_FindIndex(&process_list, send_op->process_id);
				}

				if ( ((MATCHING_FUNC *)recv_op->filter)(send_op->fsender, send_op->ftag, &tplmsg) == 0 ) return 0;
				break;
		}
	}

	return 1;
}

void CloseConnection(connection_data *connection, operation_data *operation, int thread_mutex_locked)
{
	u_long pid;
	process_data *process;
	int search_more_send, search_more_recv, index_send, index_recv, exec_send_op, ret_val;
	search_data search_send, search_recv;

	if (!thread_mutex_locked)
	{
		pthread_mutex_lock(&thread_mutex);
	}

	if ( operation != NULL )
	{
		oid_sequencer++;
		operation->operation_id = oid_sequencer;

		WriteEndRet(connection, operation);
	}

	shutdown(connection->conn_socket, SHUT_RDWR);

	closesocket(connection->conn_socket);

	connection->conn_socket = 0;
	pid = connection->process_id;
	connection->process_id = 0;
	process = connection->process;
	connection->process = NULL;

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

	process->tdata->num_conn_sockets--;

	if ( operation != NULL )
	{
		process_list_Delete(&process_list, pid);
	}
	else
	{
		process->cdata = NULL;
		process->tdata = NULL;
	}

	num_clients--;


	search_more_recv = 1;
	search_more_send = 1;
/* execute operations */
	do {
		search_recv.pid = pid;
		search_recv.num = 0;
		search_recv.max = MAX_SEARCH_DATA;

		if ( search_more_recv )
		{
			operation_list_ForEach(&operation_recv_list, (int (*)(operation_data *, void *)) (&CheckOperation), &search_recv);

			search_more_recv = ( search_recv.num == search_recv.max )?1:0;
		}

		search_send.pid = pid;
		search_send.num = 0;
		search_send.max = MAX_SEARCH_DATA;

		if ( search_more_send )
		{
			operation_list_ForEach(&operation_send_list, (int (*)(operation_data *, void *)) (&CheckOperation), &search_send);

			search_more_send = ( search_send.num == search_send.max )?1:0;
		}

		index_recv = 0;
		index_send = 0;

		while ( index_recv < search_recv.num || index_send < search_send.num )
		{
			if ( index_recv >= search_recv.num )
			{
				exec_send_op = 1;
			}
			else if ( index_send >= search_send.num )
			{
				exec_send_op = 0;
			}
			else if ( search_recv.oids[index_recv].operation_id > search_send.oids[index_send].operation_id )
			{
				exec_send_op = 1;
			}
			else
			{
				exec_send_op = 0;
			}

			if ( exec_send_op == 0 )
			{
/* execute recv operation */
				ret_val = ExecuteRecvFail(&(search_recv.oids[index_recv]));
				if ( ret_val == 0 )
				{
					write_log_sendrecv_exec(4, search_recv.oids[index_recv].process_id, pid, search_recv.oids[index_recv].message_id, -1, search_recv.oids[index_recv].mlength, 0, NULL);
				}
				else
				{
					write_log_sendrecv_exec(0, search_recv.oids[index_recv].process_id, pid, search_recv.oids[index_recv].message_id, -1, 0, ret_val, "error executing operation");
					operation_list_Delete(&operation_recv_list, search_recv.oids[index_recv].operation_id);
				}

				index_recv++;
			}
			else
			{
/* execute send operation */
				ret_val = ExecuteSendFail(&(search_send.oids[index_send]), NULL);
				if ( ret_val == 0 )
				{
					write_log_sendrecv_exec(2, pid, search_send.oids[index_send].process_id, -1, search_send.oids[index_send].message_id, 0, 0, NULL);
				}
				else
				{
					write_log_sendrecv_exec(0, pid, search_send.oids[index_send].process_id, -1, search_send.oids[index_send].message_id, 0, ret_val, "error executing operation");
					operation_list_Delete(&operation_send_list, search_send.oids[index_send].operation_id);
				}

				index_send++;
			}

		}

	} while ( search_more_recv || search_more_send );


	if (!thread_mutex_locked)
	{
		pthread_mutex_unlock(&thread_mutex);
	}
}

void DestroyOperation(operation_data *operation)
{
	switch (operation->operation_type)
	{
		case OpTp_START:
		case OpTp_CREATE:
			if ( operation->mtext != NULL )
			{
				free(operation->mtext);
				operation->mtext = NULL;
			}
			break;
		case OpTp_SEND:
		case OpTp_RECV:
			if ( operation->process_ids != NULL )
			{
				free(operation->process_ids);
				operation->process_ids = NULL;
			}
			break;
	}

	operation->operation_type = OpTp_NONE;
}

int ExecuteCreate(connection_data *connection, operation_data *operation)
{
	int ret;
	process_data *process;
	message_data *message;

	ret = 0;
	pthread_mutex_lock(&thread_mutex);

	oid_sequencer++;
	operation->operation_id = oid_sequencer;

	process = connection->process;

	mid_sequencer++;
	operation->message_id = mid_sequencer;

	message = message_list_Insert(&(process->message_list), operation->message_id, 0);

	if ( message == NULL )
	{
		ret = -2;
	}
	else
	{
		message->mlength = operation->mlength;
		message->mtext = operation->mtext;
		if ( WriteCreateRet(connection, operation) )
		{
			operation->mtext = NULL;
			ret = -3;
		}
	}

	if (ret == 0)
	{
		write_log_operation_create(1, connection->process_id, (operation->SignalPtr == 0)?1:2, operation->mlength, operation->mtext, operation->message_id, 0, NULL);
	}

	pthread_mutex_unlock(&thread_mutex);

	return ret;
}

int ExecuteDestroy(connection_data *connection, operation_data *operation)
{
	int ret;
	process_data *process;

	ret = 0;
	pthread_mutex_lock(&thread_mutex);

	oid_sequencer++;
	operation->operation_id = oid_sequencer;

	process = connection->process;

	if ( message_list_Delete(&(process->message_list), operation->message_id) )
	{
		ret = -2;
	}

	if ( WriteDestroyRet(connection, operation) )
	{
		ret = -3;
	}

	if (ret == 0)
	{
		write_log_operation_destroy(1, connection->process_id, (operation->SignalPtr == 0)?1:2, operation->message_id, 0, NULL);
	}

	pthread_mutex_unlock(&thread_mutex);

	return ret;
}

int ExecuteEnd(connection_data *connection, operation_data *operation)
{
	write_log_operation_end(1, connection->process_id, (operation->SignalPtr == 0)?1:2, 0, NULL);
	write_log_client_disconnect(connection->process_id, "end operation disconnect");
	CloseConnection(connection, operation, 0);

	return 0;
}

int ExecuteRecv(connection_data *recv_conn, operation_data *recv_op)
{
	int ret, ids_valid;
	u_long process_index, send_message_id;
	connection_data *send_conn;
	operation_data *send_op, *new_op;
	process_data *send_process, *recv_process;
	process_search_data *process_search;
	message_data *send_message, *recv_message;

	ret = 0;
	pthread_mutex_lock(&thread_mutex);

	oid_sequencer++;
	recv_op->operation_id = oid_sequencer;


	recv_process = recv_conn->process;
	recv_op->process = recv_process;

	/* reserve message id for received message */
	mid_sequencer++;
	recv_op->message_id = mid_sequencer;


/* check if all process id's are valid */
	ids_valid = 1;
	for ( process_index = 0; process_index < recv_op->num_process_ids; process_index++ )
	{
		process_search = process_list_FindEntryEqual(&process_list, recv_op->process_ids[process_index]);
		if ( process_search == NULL )
		{
			recv_op->process_id2 = recv_op->process_ids[process_index];
			ids_valid = 0;
			break;
		}
		else if ( process_search->process->cdata == NULL )
		{
			recv_op->process_id2 = recv_op->process_ids[process_index];
			ids_valid = 0;
			break;
		}
	}

/* if at least one is invalid */
	if ( ids_valid == 0 )
	{
		ret = ExecuteRecvFail(recv_op);

		if ( ret == 0 )
		{
			write_log_operation_recv(4, recv_conn->process_id, (recv_op->SignalPtr == 0)?1:2, recv_op->num_process_ids, recv_op->process_ids, recv_op->fname2, recv_op->process_id2, recv_op->message_id, -1, recv_op->mlength, 0, NULL);
		}

		pthread_mutex_unlock(&thread_mutex);
		return ret;
	}


/* find matching operation */
	send_op = operation_list_ForEach(&operation_send_list, (int (*)(operation_data *, void *)) (&CheckSendRecvOp), recv_op);

/* if matching operation not found */
	if ( send_op == NULL )
	{
		new_op = operation_list_Insert(&operation_recv_list, recv_op->operation_id, 0);

		if ( new_op == NULL )
		{
			ret = -3;
		}
		else
		{
			*new_op = *recv_op;

			write_log_operation_recv(5, recv_conn->process_id, (recv_op->SignalPtr == 0)?1:2, recv_op->num_process_ids, recv_op->process_ids, recv_op->fname2, -1, recv_op->message_id, -1, recv_op->mlength, 0, NULL);
		}
	}
/* if matching operation found */
	else
	{
		send_process = send_op->process;

		send_conn = send_process->cdata;

		/* message id already generated */
/*		mid_sequencer++;
		recv_op->message_id = mid_sequencer;*/
		recv_op->process_id2 = send_op->process_id;
		send_op->process_id2 = recv_op->process_id;

		recv_message = message_list_Insert(&(recv_process->message_list), recv_op->message_id, 0);

		if ( recv_message == NULL )
		{
			ret = -5;
		}
		else
		{
			recv_message->mlength = send_op->mlength;
			recv_message->mtext = send_op->mtext;

			recv_op->mlength = recv_message->mlength;
			recv_op->mtext = recv_message->mtext;

		if ( WriteRecvRet(recv_conn, recv_op) )
			{
				ret = -6;
			}
			else
			{
				if ( WriteSendRet(send_conn, send_op) )
				{
					ret = -7;
				}
				else
				{
					send_message_id = send_op->message_id;
					if ( operation_list_Delete(&operation_send_list, send_op->operation_id) )
					{
						ret = -8;
					}
					else
					{
						send_message = message_list_FindEntryEqual(&(send_process->message_list), send_message_id);

						if ( send_message == NULL )
						{
							ret = -9;
						}
						else
						{
							send_message->mlength = 0;
							send_message->mtext = NULL;

							if ( message_list_Delete(&(send_process->message_list), send_message->message_id) )
							{
								ret = -10;
							}
							else
							{
								write_log_operation_recv(1, recv_conn->process_id, (recv_op->SignalPtr == 0)?1:2, recv_op->num_process_ids, recv_op->process_ids, recv_op->fname2, recv_op->process_id2, recv_op->message_id, send_message_id, recv_op->mlength, 0, NULL);
							}
						}
					}
				}
			}
		}
	}

	pthread_mutex_unlock(&thread_mutex);

	return ret;
}

int ExecuteRecvFail(operation_data *recv_op)
{
	process_data *recv_process;
	int ret;
	message_data *recv_message;
	char buffer[32];

	ret = 0;
	recv_process = recv_op->process;

	/* message id already generated */
/*	mid_sequencer++;
	recv_op->message_id = mid_sequencer;*/

	recv_message = message_list_Insert(&(recv_process->message_list), recv_op->message_id, 0);

	if ( recv_message == NULL )
	{
		ret = -5;
	}
	else
	{
		sprintf(buffer, "SENDER_NOT_FOUND=%li", recv_op->process_id2);

		recv_op->mtext = strdup(buffer);
		recv_op->mlength = strlen(recv_op->mtext) + 1;

		recv_message->mlength = recv_op->mlength;
		recv_message->mtext = recv_op->mtext;

		if ( WriteRecvRet(recv_process->cdata, recv_op) )
		{
			ret = -6;
		}
	}

	return ret;
}

int ExecuteSend(connection_data *send_conn, operation_data *send_op)
{
	int ret, ids_valid;
	u_long process_index;
	connection_data *recv_conn;
	operation_data *recv_op, *new_op;
	process_data *send_process, *recv_process;
	process_search_data *process_search;
	message_data *send_message, *recv_message;
	u_long send_mlength, recv_mid;
	char *send_mtext;

	ret = 0;
	pthread_mutex_lock(&thread_mutex);

	oid_sequencer++;
	send_op->operation_id = oid_sequencer;


	send_process = send_conn->process;
	send_op->process = send_process;

	send_message = message_list_FindEntryEqual(&(send_process->message_list), send_op->message_id);

	if ( send_message == NULL )
	{
		pthread_mutex_unlock(&thread_mutex);

		return -2;
	}
	else
	{
		send_op->mlength = send_message->mlength;
		send_op->mtext = send_message->mtext;
	}

/* check if all process id's are valid */
	ids_valid = 1;
	for ( process_index = 0; process_index < send_op->num_process_ids; process_index++ )
	{
		process_search = process_list_FindEntryEqual(&process_list, send_op->process_ids[process_index]);
		if ( process_search == NULL )
		{
			send_op->process_id2 = send_op->process_ids[process_index];
			ids_valid = 0;
			break;
		}
		else if ( process_search->process->cdata == NULL )
		{
			send_op->process_id2 = send_op->process_ids[process_index];
			ids_valid = 0;
			break;
		}
	}

/* if at least is invalid */
	if ( ids_valid == 0 )
	{
		ret = ExecuteSendFail(send_op, send_message);

		if ( ret == 0)
		{
			write_log_operation_send(2, send_conn->process_id, (send_op->SignalPtr == 0)?1:2, send_op->message_id, send_op->num_process_ids, send_op->process_ids, send_op->process_ids[process_index], -1, 0, NULL);
		}

		pthread_mutex_unlock(&thread_mutex);
		return ret;
	}

/* find matching operation */
	recv_op = operation_list_ForEach(&operation_recv_list, (int (*)(operation_data *, void *)) (&CheckRecvSendOp), send_op);

/* if matching operation not found */
	if ( recv_op == NULL )
	{
		new_op = operation_list_Insert(&operation_send_list, send_op->operation_id, 0);

		if ( new_op == NULL )
		{
			ret = -3;
		}
		else
		{
			*new_op = *send_op;
			write_log_operation_send(3, send_conn->process_id, (send_op->SignalPtr == 0)?1:2, send_op->message_id, send_op->num_process_ids, send_op->process_ids, -1, -1, 0, NULL);
		}
	}
/* if matching operation found */
	else
	{
		recv_process = recv_op->process;

		recv_conn = recv_process->cdata;

		/* message id already generated */
		/*mid_sequencer++;
		recv_op->message_id = mid_sequencer;*/
		recv_op->process_id2 = send_op->process_id;
		send_op->process_id2 = recv_op->process_id;

		send_mlength = send_message->mlength;
		send_mtext = send_message->mtext;

		send_message->mlength = 0;
		send_message->mtext = NULL;

		recv_mid = recv_op->message_id;

		recv_message = message_list_Insert(&(recv_process->message_list), recv_op->message_id, 0);
		/* from here send_message is invalid */

		if ( recv_message == NULL )
		{
			ret = -5;
		}
		else
		{
			recv_message->mlength = send_mlength;
			recv_message->mtext = send_mtext;

			recv_op->mlength = recv_message->mlength;
			recv_op->mtext = recv_message->mtext;

			if ( WriteRecvRet(recv_conn, recv_op) )
			{
				ret = -6;
			}
			else
			{
				if ( WriteSendRet(send_conn, send_op) )
				{
					ret = -7;
				}
				else
				{
					if ( operation_list_Delete(&operation_recv_list, recv_op->operation_id) )
					{
						ret = -8;
					}
					else
					{
						if ( message_list_Delete(&(send_process->message_list), send_op->message_id) )
						{
							ret = -9;
						}
						else
						{
							write_log_operation_send(1, send_conn->process_id, (send_op->SignalPtr == 0)?1:2, send_op->message_id, send_op->num_process_ids, send_op->process_ids, send_op->process_id2, recv_mid, 0, NULL);
						}
					}
				}
			}
		}
	}

	pthread_mutex_unlock(&thread_mutex);

	return ret;
}

int ExecuteSendFail(operation_data *send_op, message_data *send_message)
{
	process_data *send_process;
	int ret;

	ret = 0;
	send_process = send_op->process;

	if ( send_message == NULL )
	{
		send_message = message_list_FindEntryEqual(&(send_process->message_list), send_op->message_id);

		if ( send_message == NULL )
		{
			return -2;
		}
		else
		{
			send_op->mlength = send_message->mlength;
			send_op->mtext = send_message->mtext;
		}
	}


	if ( send_process->cdata != NULL )
	{
		if ( WriteSendRet(send_process->cdata, send_op) )
		{
			ret = -7;
		}
	}

	if ( message_list_Delete(&(send_process->message_list), send_message->message_id) )
	{
		ret = -9;
	}

	return ret;
}

int ExecuteStart(connection_data *connection, operation_data *operation)
{
	u_long new_pid;
	int ret;
	process_data *process;
	process_search_data *process_search;
	message_data *message;
	char buffer[24];

	ret = 0;
	new_pid = 0;
	pthread_mutex_lock(&thread_mutex);

	oid_sequencer++;
	operation->operation_id = oid_sequencer;

	while ( new_pid == 0 )
	{
		pid_sequencer++;
		new_pid = pid_sequencer;

		process_search = process_list_FindEntryEqual(&process_list, new_pid);

		if ( process_search != NULL )
		{
			new_pid = 0;
		}
	}

	process_search = process_list_Insert(&process_list, new_pid, 0);

	if ( process_search == NULL )
	{
		pthread_mutex_unlock(&thread_mutex);

		return -4;
	}

	if ( process_search->process == NULL )
	{
		process_list_Delete(&process_list, new_pid);
		pthread_mutex_unlock(&thread_mutex);

		return -5;
	}

	sprintf(buffer, "NEW_ID=%li", new_pid);

	operation->mtext = strdup(buffer);
	operation->mlength = strlen(buffer) + 1;


	process = connection->process;

	mid_sequencer++;
	operation->message_id = mid_sequencer;

	message = message_list_Insert(&(process->message_list), operation->message_id, 0);

	if ( message == NULL )
	{
		ret = -2;
	}
	else
	{
		message->mlength = operation->mlength;
		message->mtext = operation->mtext;
		if ( WriteStartRet(connection, operation) )
		{
			operation->mtext = NULL;
			ret = -3;
		}
	}

	if (ret == 0)
	{
		write_log_operation_start(1, connection->process_id, (operation->SignalPtr == 0)?1:2, new_pid, operation->message_id, operation->mlength, 0, NULL);
	}

	pthread_mutex_unlock(&thread_mutex);

	return ret;
}

filter_function find_filter_function(char *fname, int *ftype, char **fname2)
{
	function_data *function_ptr;

	function_ptr = &(functions[0]);

	while ( function_ptr->fname != NULL )
	{
		if ( strcmp(fname, function_ptr->fname) == 0 )
		{
			if ( ftype != NULL ) *ftype = function_ptr->ftype;
			if ( fname2 != NULL ) *fname2 = function_ptr->fname;
			return function_ptr->fptr;
		}

		function_ptr++;
	}

	return NULL;
}

int ReadCreate(connection_data *connection, operation_data *operation)
{
	if ( !td_read_u_long(connection->conn_socket, &(operation->SignalPtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->MIDPtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->mlength)) ) return -3;

	if ( operation->mlength != 0 )
	{
		operation->mtext = (char *) malloc(operation->mlength);

		if ( operation->mtext == NULL ) return -4;

		if ( !td_read_char_array(connection->conn_socket, operation->mtext, operation->mlength) )
		{
			free(operation->mtext);
			return -3;
		}
	}
	else
	{
		operation->mtext = NULL;
	}

	operation->operation_type = OpTp_CREATE;
	return 0;
}

int ReadDestroy(connection_data *connection, operation_data *operation)
{
	if ( !td_read_u_long(connection->conn_socket, &(operation->SignalPtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->message_id)) ) return -3;

	operation->operation_type = OpTp_DESTROY;
	return 0;
}

int ReadEnd(connection_data *connection, operation_data *operation)
{
	if ( !td_read_u_long(connection->conn_socket, &(operation->SignalPtr)) ) return -3;

	operation->operation_type = OpTp_END;
	return 0;
}

int ReadRecv(connection_data *connection, operation_data *operation)
{
	u_long process_index;
	u_long flength;
	char *fname;

	if ( !td_read_u_long(connection->conn_socket, &(operation->SignalPtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->MessagePtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->MLengthPtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->MIDPtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->PIDPtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->num_process_ids)) ) return -3;

	if ( operation->num_process_ids != 0 )
	{
		operation->process_ids = (u_long *) malloc(operation->num_process_ids * sizeof(u_long));

		if ( operation->process_ids == NULL ) return -4;

		for ( process_index = 0; process_index < operation->num_process_ids; process_index++ )
		{
			if ( !td_read_u_long(connection->conn_socket, &(operation->process_ids[process_index])) )
			{
				free(operation->process_ids);
				return -3;
			}
		}
	}
	else
	{
		operation->process_ids = NULL;
	}

	if ( !td_read_u_long(connection->conn_socket, &flength) )
	{
		free(operation->process_ids);
		return -3;
	}

	if ( flength != 0 )
	{
		fname = (char *) malloc(flength + 1);

		if ( fname == NULL )
		{
			free(operation->process_ids);
			return -4;
		}

		if ( !td_read_string(connection->conn_socket, fname, flength) )
		{
			free(fname);
			free(operation->process_ids);
			return -3;
		}

		operation->filter = find_filter_function(fname, &operation->ftype, &operation->fname2);

		if ( operation->filter == NULL )
		{
			free(fname);
			free(operation->process_ids);
			return -5;
		}

	}
	else
	{
		operation->fname2 = NULL;
		operation->filter = NULL;
	}

	operation->operation_type = OpTp_RECV;
	return 0;
}

int ReadSend(connection_data *connection, operation_data *operation)
{
	u_long process_index;

	if ( !td_read_u_long(connection->conn_socket, &(operation->SignalPtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->message_id)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->num_process_ids)) ) return -3;

	if ( operation->num_process_ids != 0 )
	{
		operation->process_ids = (u_long *) malloc(operation->num_process_ids * sizeof(u_long));

		if ( operation->process_ids == NULL ) return -4;

		for ( process_index = 0; process_index < operation->num_process_ids; process_index++ )
		{
			if ( !td_read_u_long(connection->conn_socket, &(operation->process_ids[process_index])) )
			{
				free(operation->process_ids);
				return -3;
			}
		}
	}
	else
	{
		operation->process_ids = NULL;
	}

	operation->fsender = -1;

	operation->operation_type = OpTp_SEND;
	return 0;
}

int ReadStart(connection_data *connection, operation_data *operation)
{
	operation->mlength = 0;
	operation->mtext = NULL;

	if ( !td_read_u_long(connection->conn_socket, &(operation->SignalPtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->MessagePtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->MLengthPtr)) ) return -3;

	if ( !td_read_u_long(connection->conn_socket, &(operation->MIDPtr)) ) return -3;

	operation->operation_type = OpTp_START;
	return 0;
}

#if defined(WAIT_FOR_WRITE)
	void WaitForWrite(int socket)
	{
		fd_set write_sockets;

		FD_ZERO(&write_sockets);
		FD_SET(socket, &write_sockets);

		select(socket + 1, NULL, &write_sockets, NULL, NULL);
	}
#else
	#define WaitForWrite(x)
#endif

int WriteCreateRet(connection_data *connection, operation_data *operation)
{
	u_char OpTp;

	if (connection == NULL) return -10;

	OpTp = OpTp_CREATE_RET;

	WaitForWrite(connection->conn_socket);

	if ( !td_write_u_char(connection->conn_socket, &OpTp, &(connection->td_buffer)) ) return -1;

	if ( !td_write_u_long(connection->conn_socket, &(operation->SignalPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->MIDPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->message_id), &(connection->td_buffer)) ) return -2;

	if ( !td_flush(connection->conn_socket, &(connection->td_buffer)) ) return -4;

	return 0;
}

int WriteDestroyRet(connection_data *connection, operation_data *operation)
{
	u_char OpTp;

	if (connection == NULL) return -10;

	OpTp = OpTp_DESTROY_RET;

	WaitForWrite(connection->conn_socket);

	if ( !td_write_u_char(connection->conn_socket, &OpTp, &(connection->td_buffer)) ) return -1;

	if ( !td_write_u_long(connection->conn_socket, &(operation->SignalPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_flush(connection->conn_socket, &(connection->td_buffer)) ) return -4;

	return 0;
}

int WriteEndRet(connection_data *connection, operation_data *operation)
{
	u_char OpTp;

	if (connection == NULL) return -10;

	OpTp = OpTp_END_RET;

	WaitForWrite(connection->conn_socket);

	if ( !td_write_u_char(connection->conn_socket, &OpTp, &(connection->td_buffer)) ) return -1;

	if ( !td_write_u_long(connection->conn_socket, &(operation->SignalPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_flush(connection->conn_socket, &(connection->td_buffer)) ) return -4;

	return 0;
}

int WriteRecvRet(connection_data *connection, operation_data *operation)
{
	u_char OpTp;

	if (connection == NULL) return -10;

	OpTp = OpTp_RECV_RET;

	WaitForWrite(connection->conn_socket);

	if ( !td_write_u_char(connection->conn_socket, &OpTp, &(connection->td_buffer)) ) return -1;

	if ( !td_write_u_long(connection->conn_socket, &(operation->SignalPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->MessagePtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->MLengthPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->MIDPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->PIDPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->message_id), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->process_id2), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->mlength), &(connection->td_buffer)) ) return -2;

	if ( operation->mlength != 0 ) {
		if ( !td_write_char_array(connection->conn_socket, operation->mtext, operation->mlength, &(connection->td_buffer)) ) return -3;
	}

	if ( !td_flush(connection->conn_socket, &(connection->td_buffer)) ) return -4;

	return 0;
}

int WriteSendRet(connection_data *connection, operation_data *operation)
{
	u_char OpTp;

	if (connection == NULL) return -10;

	OpTp = OpTp_SEND_RET;

	WaitForWrite(connection->conn_socket);

	if ( !td_write_u_char(connection->conn_socket, &OpTp, &(connection->td_buffer)) ) return -1;

	if ( !td_write_u_long(connection->conn_socket, &(operation->SignalPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_flush(connection->conn_socket, &(connection->td_buffer)) ) return -4;

	return 0;
}

int WriteStartRet(connection_data *connection, operation_data *operation)
{
	u_char OpTp;

	if (connection == NULL) return -10;

	OpTp = OpTp_START_RET;

	WaitForWrite(connection->conn_socket);

	if ( !td_write_u_char(connection->conn_socket, &OpTp, &(connection->td_buffer)) ) return -1;

	if ( !td_write_u_long(connection->conn_socket, &(operation->SignalPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->MessagePtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->MLengthPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->MIDPtr), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->message_id), &(connection->td_buffer)) ) return -2;

	if ( !td_write_u_long(connection->conn_socket, &(operation->mlength), &(connection->td_buffer)) ) return -2;

	if ( operation->mlength != 0 )
	{
		if ( !td_write_char_array(connection->conn_socket, operation->mtext, operation->mlength, &(connection->td_buffer)) ) return -3;
	}

	if ( !td_flush(connection->conn_socket, &(connection->td_buffer)) ) return -4;

	return 0;
}

void *conn_thread(thread_data *data)
{
	int max_socket, select_return, return_value;
	struct timeval select_timeout;
	fd_set read_sockets, except_sockets;
	int isockets, socket_error, mutex_locked;
	u_char OpTp;
	operation_data operation;

	mutex_locked = 0;

/* wait for initialization of data */
	pthread_mutex_lock(&thread_mutex);
	pthread_mutex_unlock(&thread_mutex);

	while ( !data->exit_thread )
	{
		select_timeout.tv_sec = 0;
		select_timeout.tv_usec = 100000;
		FD_ZERO(&read_sockets);

		max_socket = 0;

		for (isockets = 0; isockets < MAX_CLIENTS_PER_THREAD; isockets++)
		{
			if ( data->conn_sockets[isockets].conn_socket != 0 )
			{
				if ( data->conn_sockets[isockets].conn_socket > max_socket ) max_socket = data->conn_sockets[isockets].conn_socket;

				FD_SET(data->conn_sockets[isockets].conn_socket, &read_sockets);
			}
		}

/* if no connections then exit thread */
		if ( max_socket == 0 )
		{
/* thread locked, count connections and then exit thread */
			pthread_mutex_lock(&thread_mutex);

			for (isockets = 0; isockets < MAX_CLIENTS_PER_THREAD; isockets++)
			{
				if ( data->conn_sockets[isockets].conn_socket != 0 )
				{
					if ( data->conn_sockets[isockets].conn_socket > max_socket ) max_socket = data->conn_sockets[isockets].conn_socket;

					FD_SET(data->conn_sockets[isockets].conn_socket, &read_sockets);
				}
			}

			if ( max_socket == 0 )
			{
				mutex_locked = 1;
				break;
			}
			else
			{
				pthread_mutex_unlock(&thread_mutex);
			}
		}

		except_sockets = read_sockets;

		select_return = select(max_socket + 1, &read_sockets, NULL, &except_sockets, &select_timeout);
		if ( select_return == -1)
		{
			exit_server = 1;
			break;
		}
		if ( select_return == 0) continue;

/* check sockets */
		for (isockets = 0; isockets < MAX_CLIENTS_PER_THREAD; isockets++)
		{
			if ( data->conn_sockets[isockets].conn_socket == 0 ) continue;

			if (!FD_ISSET(data->conn_sockets[isockets].conn_socket, &read_sockets) &&
				!FD_ISSET(data->conn_sockets[isockets].conn_socket, &except_sockets)
				)
			{
				continue;
			}


			socket_error = 1;
/* if there is data to read, then read and execute operation*/
			if ( !FD_ISSET(data->conn_sockets[isockets].conn_socket, &except_sockets) )
			{
				if ( td_read_u_char(data->conn_sockets[isockets].conn_socket, &OpTp) )
				{
					operation.process_id = data->conn_sockets[isockets].process_id;
					operation.operation_type = OpTp_NONE;
					get_time(&operation.operation_time);

					switch (OpTp)
					{
						case OpTp_END:
							return_value = ReadEnd(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_end(0, data->conn_sockets[isockets].process_id, (operation.SignalPtr == 0)?1:2, return_value, "error reading operation");
								break;
							}

							return_value = ExecuteEnd(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_end(0, data->conn_sockets[isockets].process_id, (operation.SignalPtr == 0)?1:2, return_value, "error executing operation");
								DestroyOperation(&operation);
								break;
							}

							socket_error = 0;
							break;
						case OpTp_START:
							return_value = ReadStart(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_start(0, data->conn_sockets[isockets].process_id, 0, -1, -1, 0, return_value, "error reading operation");
								break;
							}

							return_value = ExecuteStart(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_start(0, data->conn_sockets[isockets].process_id, (operation.SignalPtr == 0)?1:2, -1, -1, 0, return_value, "error executing operation");
								DestroyOperation(&operation);
								break;
							}

							socket_error = 0;
							break;
						case OpTp_CREATE:
							return_value = ReadCreate(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_create(0, data->conn_sockets[isockets].process_id, 0, 0, NULL, -1, return_value, "error reading operation");
								break;
							}

							return_value = ExecuteCreate(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_create(0, data->conn_sockets[isockets].process_id, (operation.SignalPtr == 0)?1:2, operation.mlength, operation.mtext, -1, return_value, "error executing operation");
								DestroyOperation(&operation);
								break;
							}

							socket_error = 0;
							break;
						case OpTp_DESTROY:
							return_value = ReadDestroy(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_destroy(0, data->conn_sockets[isockets].process_id, 0, -1, return_value, "error reading operation");
								break;
							}

							return_value = ExecuteDestroy(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_destroy(0, data->conn_sockets[isockets].process_id, (operation.SignalPtr == 0)?1:2, operation.message_id, return_value, "error executing operation");
								DestroyOperation(&operation);
								break;
							}

							socket_error = 0;
							break;
						case OpTp_SEND:
							return_value = ReadSend(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_send(0, data->conn_sockets[isockets].process_id, (operation.SignalPtr == 0)?1:2, -1, 0, NULL, -1, -1, return_value, "error reading operation");
								break;
							}

							return_value = ExecuteSend(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_send(0, data->conn_sockets[isockets].process_id, (operation.SignalPtr == 0)?1:2, operation.message_id, operation.num_process_ids, operation.process_ids, -1, -1, return_value, "error executing operation");
								DestroyOperation(&operation);
								break;
							}

							socket_error = 0;
							break;
						case OpTp_RECV:
							return_value = ReadRecv(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_recv(0, data->conn_sockets[isockets].process_id, (operation.SignalPtr == 0)?1:2, 0, NULL, NULL, -1, -1, -1, 0, return_value, "error reading operation");
								break;
							}

							return_value = ExecuteRecv(&(data->conn_sockets[isockets]), &operation);
							if ( return_value )
							{
								write_log_operation_recv(0, data->conn_sockets[isockets].process_id, (operation.SignalPtr == 0)?1:2, operation.num_process_ids, operation.process_ids, operation.fname2, -1, -1, -1, 0, return_value, "error executing operation");
								DestroyOperation(&operation);
								break;
							}

							socket_error = 0;
							break;
						default:
							write_log_client_disconnect(OpTp, "xxx");

							break;
					}

					if ( socket_error != 0 )
					{
						write_log_client_disconnect(data->conn_sockets[isockets].process_id, "server side disconnect");
					}
				}
				else
				{
					write_log_client_disconnect(data->conn_sockets[isockets].process_id, "client side disconnect");
				}
			}
			else
			{
				write_log_client_disconnect(data->conn_sockets[isockets].process_id, "communication error");
			}

/* if error then disconnect client */
			if ( socket_error != 0 )
			{
				CloseConnection(&(data->conn_sockets[isockets]), NULL, 0);
			}

		}




	}


/* exit thread */
	if ( mutex_locked == 0)
	{
		pthread_mutex_lock(&thread_mutex);
	}

	for (isockets = 0; isockets < MAX_CLIENTS_PER_THREAD; isockets++)
	{
		if ( data->conn_sockets[isockets].conn_socket != 0 )
		{
			CloseConnection(&(data->conn_sockets[isockets]), NULL, 1);
		}
	}

	if ( num_threads == 1 )
	{
		first_thread_data = NULL;
	}
	else
	{
		data->prev->next = data->next;
		data->next->prev = data->prev;

		if ( first_thread_data == data )
		{
			first_thread_data = data->next;
		}
	}
	free(data);

	if ( exit_semaphore_waiting && num_threads == 1 )
	{
		sem_post(&exit_semaphore);
	}
	num_threads--;

	pthread_mutex_unlock(&thread_mutex);

	pthread_exit(NULL);

/* to avoid warning */
	return NULL;
}
