#define _GNU_SOURCE

#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <time.h>

#include <unistd.h>
#include <fcntl.h>

#include "common.h"
#include "ipc.h"
#include "pa2345.h"
#include "banking.h"

int childCount;
int procID;

FILE* pointer_events_log;
FILE* pointer_pipes_log;

pid_t parentPid;

typedef struct 
{
	int in;
	int out;
} Pipes;
Pipes* pointerPipes = NULL;

typedef struct
{
	int fromID;
	int toID;
} Data;

int* pointerBalance;
BalanceHistory* pointerBalanceHistory = NULL;

timestamp_t lamport_time = 0;

enum
{
	MAX_LEN_LOG_LINE = 5000
};

static const char * const log_pipe_read_fmt = "Pipe %d read data. Process - %d. Bytes - %lu. Message: %s\n";
static const char * const log_pipe_write_fmt = "Pipe %d write data. Process - %d. Bytes - %lu. Message: %s\n";
static const char * const log_pipe_close_fmt = "Pipe %d closed. Process - %d\n";
static const char * const log_pipe_create_fmt = "Pipe create (in %d out %d)\n";

void openLogsFile()
{
	pointer_events_log = fopen(events_log, "w+b");
	pointer_pipes_log = fopen(pipes_log, "w+b");
}

void pipesLogWrite(char* line)
{
	fprintf(pointer_pipes_log, "%s", line);
	fflush(pointer_pipes_log);
}

void eventsLogWrite(char* line)
{
	fprintf(pointer_events_log,"%s",line);
	fflush(pointer_events_log);
}

void incTime()
{
	lamport_time++;
}

timestamp_t getTime()
{
	return lamport_time;
}

void compareTime(timestamp_t t)
{
	if(t > lamport_time)
	{
		lamport_time = t;
	}
	incTime();
}

void createPipe(int id)
{
	//2 числа для канала. Для чтения и для записи.
	int fd[2];
	//Создаем канал
	pipe2(fd,O_NONBLOCK);
	//После создания канала, заполняем нашу структуру связями вход/выход
	pointerPipes[id].in = fd[0];
	pointerPipes[id].out = fd[1];
}

//pCount = количество потомков + родитель
void createPipes(int pCount)
{
	//Считаем количество каналов
	int count = childCount * (childCount + 1);
	//Выделяем память под массив каналов
	pointerPipes = (Pipes*)malloc(sizeof(Pipes)*count);

	//Проходим по всем каналам
	for (int i=0; i<count; i++)
	{
		createPipe(i);	
		
		//Запись в лог
		char line[MAX_LEN_LOG_LINE];
		sprintf(line,log_pipe_create_fmt,pointerPipes[i].in, pointerPipes[i].out);
		pipesLogWrite(line);
	}
}

void closePipe (int fd)
{
	char line[MAX_LEN_LOG_LINE];
	sprintf(line,log_pipe_close_fmt,fd,procID);
	pipesLogWrite(line);

	close(fd);
}

void closeExcessPipeIn(int id)
{
	closePipe(pointerPipes[id].in);
	pointerPipes[id].in = 0;	
}

void closeExcessPipeOut(int id)
{
	closePipe(pointerPipes[id].out);
	pointerPipes[id].out = 0;
}

void closeExcessPipes(int id)
{
	//Количество каналов
	int count = childCount * (childCount + 1);
	//Цикл, в котором закрываются ненужные каналы
	int fromID, toID;
	for (int i=0; i<count; i++)
	{
		fromID = i / childCount;
		toID = i % childCount;
		if (toID >= fromID) 
			toID = toID + 1;
		if (fromID == id)
		{
			closeExcessPipeIn(i);		
		}
		else
			if (toID == id)
			{
				closeExcessPipeOut(i);
			}
			else
			{
				closeExcessPipeIn(i);
				closeExcessPipeOut(i);
			}
	}
}

void closePipes()
{
	int count = childCount * (childCount+1);
	for (int i=0; i<count; i++)
	{
		closeExcessPipeIn(i);
		closeExcessPipeOut(i);
	}	
}

void freePipes()
{
	closePipes();
	free(pointerPipes);
	pointerPipes = NULL;
}

int calculatePipeNumber(int from, int to)
{
	if(to>from) 
		to = to-1;
	return from*childCount+to;
}

int pipeWrite(int fd, const Message * msg)
{
	if(fd==0) return -1;

	write(fd,&msg->s_header,sizeof(MessageHeader));
	write(fd,msg->s_payload,msg->s_header.s_payload_len);
	
	char line[MAX_LEN_LOG_LINE];
	sprintf(line,log_pipe_write_fmt,fd, procID, sizeof(MessageHeader)+msg->s_header.s_payload_len, msg->s_payload);
	pipesLogWrite(line);

	return 0;
}

int pipeRead(int fd, Message* msg, int recAny)
{
	if(fd==0) return -1;
	
	int flag;
	do
	{
		flag = read(fd,&msg->s_header,sizeof(MessageHeader)); 
		if(recAny == 0 && flag < 0)
		{
			return -1;
		}
	}
	while(flag < 0);

	flag = 0;
	do
	{
		flag = read(fd,msg->s_payload,msg->s_header.s_payload_len);
	}
	while(flag < 0);
	
	char line[MAX_LEN_LOG_LINE];
	sprintf(line,log_pipe_read_fmt,fd,procID, sizeof(MessageHeader)+msg->s_header.s_payload_len, msg->s_payload);
	pipesLogWrite(line);

	return 0;
}

int send(void * self, local_id dst, const Message * msg)
{
	Data* data = (Data*)self;
	int pipe = pointerPipes[calculatePipeNumber(data->fromID,dst)].out;

	return pipeWrite(pipe,msg);
}

int send_multicast(void * self, const Message * msg)
{
	Data* data = (Data*)self;
	for(int i=0; i<childCount+1; i++)
	{
		if(i!=data->fromID)
		{	
			int pipe = pointerPipes[calculatePipeNumber(data->fromID,i)].out;
			pipeWrite(pipe,msg);		
		}		
	}

	return 0;
}

int receive(void * self, local_id from, Message * msg)
{
	Data* data = (Data*)self;
	int pipe = pointerPipes[calculatePipeNumber(from,data->fromID)].in;
	int res = pipeRead(pipe,msg,1);
	if(res==0)
	{
		compareTime(msg->s_header.s_local_time);
	}
	return res;
}

int receive_any(void * self, Message * msg)
{
	Data* data = (Data*)self;
	for(int i=0; i<childCount+1;i++)
	{
		if(i!=data->fromID)
		{
			int pipe = pointerPipes[calculatePipeNumber(i,data->fromID)].in;
			if(pipeRead(pipe,msg,0)==0)
			{
				data->toID = i;
				compareTime(msg->s_header.s_local_time);
				return 0;
			}			
		}		
	}

	return -1;
}

Message getMessageTemplate()
{
	Message msg;
	msg.s_header.s_magic = MESSAGE_MAGIC;
	msg.s_header.s_local_time = getTime();

	return msg;
}

void waitAllMessage(MessageType type, int selfID)
{
	Message msg;
	Data data;
	data.fromID = selfID;
	for(int i=1; i<childCount+1; i++)
	{
		if(i!=selfID)
		{
			do
			{
				receive(&data,i,&msg);
			}		
			while (msg.s_header.s_type!=type);
		}
	}
}

void writeCurrentBalance(timestamp_t currentTime, balance_t amount, balance_t pending_in)
{
	int index = -2;
	//Попытка найти в истории запись, в которой s_time = currentTime
	for (int i=0; i<pointerBalanceHistory->s_history_len; i++)
	{
		if(pointerBalanceHistory->s_history[i].s_time == currentTime)
		{
			index = i;
			break;
		}
	}
	//Если нашли
	if(index > 0 && index < pointerBalanceHistory->s_history_len)
	{
		pointerBalanceHistory->s_history[index].s_time = currentTime;
		pointerBalanceHistory->s_history[index].s_balance_pending_in = pending_in;
		pointerBalanceHistory->s_history[index].s_balance = amount;
	}
	//Если не нашли, значит нужно добавить
	else
	{
		if(pointerBalanceHistory->s_history[pointerBalanceHistory->s_history_len-1].s_time < currentTime - 1)
		{
			for (timestamp_t t= pointerBalanceHistory->s_history[pointerBalanceHistory->s_history_len-1].s_time+1; t<currentTime; t++)
			{
				writeCurrentBalance(t,
					pointerBalanceHistory->s_history[pointerBalanceHistory->s_history_len-1].s_balance, 
					pointerBalanceHistory->s_history[pointerBalanceHistory->s_history_len-1].s_balance_pending_in);
			}
		}

		pointerBalanceHistory->s_history[pointerBalanceHistory->s_history_len].s_time = currentTime;
		pointerBalanceHistory->s_history[pointerBalanceHistory->s_history_len].s_balance_pending_in = pending_in;
		pointerBalanceHistory->s_history[pointerBalanceHistory->s_history_len].s_balance = amount;

		pointerBalanceHistory->s_history_len++;
	}
}

void Start(pid_t pid, int selfID)
{
	//??
	incTime();

	char line[MAX_LEN_LOG_LINE];
	sprintf(line,log_started_fmt,getTime(),selfID,pid,parentPid, pointerBalance[selfID]);
	eventsLogWrite(line);
	
	//Формирование сообщения
	Message msg = getMessageTemplate();
	sprintf(msg.s_payload,log_started_fmt,getTime(),selfID,pid,parentPid, pointerBalance[selfID]);
	msg.s_header.s_type = STARTED;
	msg.s_header.s_payload_len = strlen(msg.s_payload)+1;
	Data data;
	data.fromID = selfID;

	//Отправление сообщения о старте всем остальным процессам
	send_multicast(&data,&msg);

	//Получение сообщения STARTED всеми процессами, кроме текущего и родителя
	waitAllMessage(STARTED,selfID);

	//Выделяем память под структуру с историей баланса
	pointerBalanceHistory = (BalanceHistory*)malloc(sizeof(BalanceHistory));
	pointerBalanceHistory->s_id = selfID;
	pointerBalanceHistory->s_history_len = 0;

	timestamp_t zeroTime = 0;
	//Запись в историю значения баланса при старте t = 0
	writeCurrentBalance(zeroTime,pointerBalance[selfID],0);

	char line2[MAX_LEN_LOG_LINE];
	sprintf(line2,log_received_all_started_fmt,getTime(),selfID);
	eventsLogWrite(line);
}

void clear(void* object, size_t size)
{
	for(int i=0; i<size; i++)
		((char*)object)[i] = 0;
}

void getDataFromMsg(void* msgData, void* outData, size_t size)
{
	for(int i=0; i<size; i++)
		((char*)outData)[i] = ((char*)msgData)[i];
}

void Work(pid_t pid, int selfID)
{
	while(1)
	{
		//Запись текущего баланса
		//writeCurrentBalance(getTime(), pointerBalance[selfID]);

		Message msg;
		clear(&msg,sizeof(Message));
		Data data;
		clear(&data,sizeof(Data));
		data.fromID = selfID;

		//Попытка получения сообщения
		if(receive_any(&data,&msg) == 0)
		{
			//Если пришло сообщение TRANSFER
			if(msg.s_header.s_type == TRANSFER)
			{
				TransferOrder transferOrder;
				clear(&transferOrder,sizeof(transferOrder));
				getDataFromMsg(msg.s_payload, &transferOrder, msg.s_header.s_payload_len);

				//Если оно пришло не от родителя, то нужно увеличить баланс
				//и отправить родителю подтверждение
				if(data.toID != PARENT_ID)
				{
					//Запись в лог
					char line[MAX_LEN_LOG_LINE];
					sprintf(line,log_transfer_in_fmt,getTime(),selfID, transferOrder.s_amount, data.toID);
					eventsLogWrite(line);

					pointerBalance[selfID] = pointerBalance[selfID] + transferOrder.s_amount;
					writeCurrentBalance(getTime(), pointerBalance[selfID], 0);
					incTime();

					Message msg = getMessageTemplate();
					msg.s_header.s_type = ACK;
					msg.s_header.s_payload_len = 0;

					send(&data,PARENT_ID,&msg);
				}
				else
				//Если оно пришло от родителя, то нужно уменьшить баланс
				//и переслать сообщение дальше
				{
					char line[MAX_LEN_LOG_LINE];
					sprintf(line,log_transfer_out_fmt,getTime(),transferOrder.s_src, transferOrder.s_amount, transferOrder.s_dst);
					eventsLogWrite(line);

					pointerBalance[selfID] = pointerBalance[selfID] - transferOrder.s_amount;

					writeCurrentBalance(getTime(), pointerBalance[selfID], transferOrder.s_amount);
					writeCurrentBalance(getTime()+2,pointerBalance[selfID],0);
					incTime();

					msg.s_header.s_local_time = getTime();
					msg.s_header.s_magic = MESSAGE_MAGIC;

					send(&data,transferOrder.s_dst,&msg);
				}
			}
			else
			if(msg.s_header.s_type == STOP)
			{
				writeCurrentBalance(getTime(), pointerBalance[selfID], 0);
				break;
			}
		}
	}
}

void sendInfo(Data data, int selfID)
{
	Message msg = getMessageTemplate();
	msg.s_header.s_type = BALANCE_HISTORY;
	msg.s_header.s_payload_len = sizeof(BalanceState)*(pointerBalanceHistory->s_history_len) + sizeof(pointerBalanceHistory->s_history_len) 
		+ sizeof(pointerBalanceHistory->s_id);

	//Запись истории изменения баланса в сообщение
	for(int i =0; i<msg.s_header.s_payload_len; i++)
		((char*)msg.s_payload)[i] = ((char*)pointerBalanceHistory)[i];

	data.fromID = selfID;
	send(&data,PARENT_ID,&msg);
}

void Done(pid_t pid, int selfID)
{
	incTime();
	//Запись в лог
	char line[MAX_LEN_LOG_LINE];
	sprintf(line,log_done_fmt,getTime(),selfID,pointerBalance[selfID]);
	eventsLogWrite(line);

	//Формирование сообщения
	Message msg = getMessageTemplate();
	sprintf(msg.s_payload,log_done_fmt,getTime(),selfID,pointerBalance[selfID]);
	msg.s_header.s_type = DONE;
	msg.s_header.s_payload_len = strlen(msg.s_payload)+1;
	Data data;
	data.fromID = selfID;
	//Отправление сообщения об окончании работы всем остальным процессам
	send_multicast(&data,&msg);
	//Получение сообщения DONE всеми процессами, кроме текущего и родителя
	waitAllMessage(DONE,selfID);

	writeCurrentBalance(getTime(), pointerBalance[selfID], 0);

	incTime();
	sendInfo(data, selfID);
	free(pointerBalanceHistory);

	char line2[MAX_LEN_LOG_LINE];
	sprintf(line2,log_received_all_done_fmt,getTime(),selfID);
	eventsLogWrite(line2);
}

void preparationForStart()
{
	//Открытие файлов для логгирования
	openLogsFile();
	//Создание каналов
	createPipes(childCount + 1);
}

void childWork(int id)
{
	//Закрываем ненужные каналы. Нам нужны только на зпись и чтение из других потомков
	closeExcessPipes(id);
	//Цикл работы потомка
	Start(getpid(),id);
	Work(getpid(),id);
	Done(getpid(),id);
}

TransferOrder getOrder(local_id src, local_id dst, balance_t amount)
{
	TransferOrder transferOrder;
	transferOrder.s_src = src;
	transferOrder.s_dst = dst;
	transferOrder.s_amount = amount;

	return transferOrder;
}

void transfer(void * parent_data, local_id src, local_id dst, balance_t amount)
{
	incTime();

    Message msg = getMessageTemplate();
    TransferOrder transferOrder = getOrder(src,dst,amount);
    Data data;
    data.fromID = PARENT_ID;
    msg.s_header.s_type = TRANSFER;
    msg.s_header.s_payload_len = sizeof(TransferOrder);

    getDataFromMsg(&transferOrder, msg.s_payload,msg.s_header.s_payload_len);
    send(&data,src,&msg);
    //Получение ACK
    receive(&data,dst,&msg);
}

void getAndWriteAllHistory(int childCount)
{
	Data data;
	Message msg;

	data.fromID = PARENT_ID;

	AllHistory allHistory;
	allHistory.s_history_len = childCount;
	for (int i=1; i <=childCount; i++)
	{
		receive(&data,i,&msg);
		getDataFromMsg(msg.s_payload, &allHistory.s_history[i-1], msg.s_header.s_payload_len);
	}

	print_history(&allHistory);
}

/*
void bank_robbery(void * parent_data, local_id max_id)
{
	transfer(parent_data, 1, 2, 5);
	transfer(parent_data, 2, 1, 3);
}*/

void parentWork()
{
	procID = 0;
	parentPid = getpid();
	//Закрываем ненужные каналы у родителя
	closeExcessPipes(PARENT_ID);
	//Получаем все сообщения STARTED
	waitAllMessage(STARTED, PARENT_ID);
	bank_robbery(NULL,childCount);
	//Отправление сообщения STOP
	incTime();
	Data data;
	Message stopMsg = getMessageTemplate();
	stopMsg.s_header.s_type = STOP;
	stopMsg.s_header.s_payload_len = 0;
	data.fromID = PARENT_ID;
	send_multicast(&data,&stopMsg);
	//Получаем все сообщения DONE
	waitAllMessage(DONE,PARENT_ID);
	getAndWriteAllHistory(childCount);
	int st;
	do{}
	while (wait(&st)>0);
}

void startSystem()
{
	//pid_t - встроенный тип. Позволяет получать "ид процесса"
	pid_t forkPid;
	for (int i = 1; i < childCount + 1; i++)
	{
		//Создание потомка. Весь код после fork выполняется и в родителе, и в потомке
		forkPid = fork();
		//Если текущий процесс потомок
		if (forkPid == 0)
		{
			procID = i;
			childWork(i);
			break;
		}
	}
	if(forkPid != 0)
	{
		parentWork();
	}
}

void shutdown()
{
	//Освобождаем все каналы
	freePipes();

	//Закрываем логи
	fclose(pointer_events_log);
	fclose(pointer_pipes_log);
}

 int main(int argc, char *argv[]) 
{
	//Получение числа потомков
	if (argc <3 || strcmp(argv[1], "-p") !=0)
		return 1;
	childCount = atoi(argv[2]);

	pointerBalance = (int*)malloc(sizeof(int) * childCount + 1);
	pointerBalance[0] = 0;
	for (int i=1; i<=childCount; i++)
	{
		pointerBalance[i] = atoi(argv[i+2]);
	}

	preparationForStart();
	startSystem();
	shutdown();

	return 0;
}