/***************************************************************/
/* Simulacao de uma Versao do Algoritmo de Consenso Paxos      */
/*  							       */	
/* Disciplina: CI721 - Sistemas Distribuídos		       */
/* Autor: Flavio Rogerio Uber				       */
/* Professor: Elias P. Duarte Jr.			       */
/*							       */				
/***************************************************************/
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include "smpl.h"

/* eventos */
#define send_msg 1
#define recv_msg 2
#define recv_ack 3
#define recv_nack 4
#define fault 5
#define decision 6


typedef struct {
	int id;
	int *tempo;
	int instante;
	int *aceite;
	int valor;
} tmsg;

/* descritor do nodo */
typedef struct {
	int id;
        int *STATE;
	int id_atual;
	int id_aceito;
	int vl_consenso;
}tnodo;

tmsg *msgs;


	

tnodo *nodo;

main(int argc, char *argv[]){
	static int N,
                   token,
		   P,
                   event, r, i, j, k, rnd, qt_ack, valor_final, qt, id_temp, falho;


	static char fa_name[5];
	if (argc != 2){
		puts("Uso correto: trabalho <num_nodos>"); 
		exit(1);						    
	}
	N = atoi(argv[1]);
	smpl(0,"Um exemplo de simulacao");
	reset();
	stream(1);


	
	/* inicializacao */
	nodo = (tnodo *)malloc(sizeof(tnodo)*N);
	for (i=0;i<N;i++) {
		memset(fa_name,'\0',5);
		printf(fa_name,"%d",i);
		nodo[i].id = facility (fa_name,1);
		nodo[i].id_atual=i;
		nodo[i].id_aceito = -1;
		nodo[i].vl_consenso = -1;
	} //end for

	/* inicializando o vetor STATE */
	for(i=0; i<N; i++){
		nodo[i].STATE = (int *)malloc(sizeof(int)*N);
		for(j=0; j<N; j++) 
			nodo[i].STATE[j] = -1;
	}

	msgs = (tmsg *)malloc(sizeof(tmsg)*N);
	for (i=0;i<N;i++)
	{
		msgs[i].id=-1;
		msgs[i].tempo = (int *)malloc(sizeof(int)*N);
		msgs[i].aceite = (int *)malloc(sizeof(int)*N);
		for(j=0;j<N;j++)
			msgs[i].tempo[j] = 0;
		for(j=0;j<N;j++)
			msgs[i].aceite[j] = 0;
		msgs[i].valor = -1;
	}



	/* schedule do primeiro nodo */
	

	schedule(send_msg, 10.0, 0);   //escalonamento inicial para envio da prmeira mensagem
	schedule(send_msg, 12.0, 2); //escalonamento de um segundo processo enviando mensagem
	//schedule(fault,0,0);         //escalonamento de um processo falho desde o início da simulacao


	printf("**************************************************************\n");
	printf("*** Simulação de uma Versão do Algoritmo de Consenso Paxos ***\n");
	printf("**************************************************************\n");



	srand((unsigned)time(NULL));	//Para a escolha do número randômico é preciso mudar a semente a cada execucao

	while (time1() < 300) {
		cause(&event, &token);
                i = (token+1)%N;       //estabelece o nodo i que sera testado
		nodo[token].STATE[token] = 0; //o vetor STATE no token em questão, para sua propria posicao e setado como 0

		switch(event) {
			case send_msg:
				if (status(nodo[token].id) != 0) break; // se o processo estiver falho nao havera envio da mensagem
				
				//seta id da mensagem com o contador atual de mensagens do token (o valor do consenso será o mesmo)

				msgs[token].id = nodo[token].id_atual;	
				msgs[token].valor = nodo[token].id_atual;
				msgs[token].instante = time1();
				
				printf("\n\nProcesso %d envia mensagem  com ID %d aos demais.\n",token, msgs[token].id);


				//o nodo emissor sera setado como tendo aceito a mensagem, para que isso seja usado na validacao
				nodo[token].id_aceito = nodo[token].id_atual;
				
				//atualiza o valor do id da proxima mensagem que o token gerara	
				nodo[token].id_atual = nodo[token].id_atual + N;

				rnd = rand() % 101; //escolhe um numero randomico entre 0 e 100
				rnd = time1() + rnd;

							
				while (i != token) { 
					if (status(nodo[i].id) == 1) { //caso o processo esteja falho isso sera informado no vetor STATE de todos os processos
						k = (i+1)%N;
						while (k != i) {
							nodo[k].STATE[i] = 1;
							k = (k+1)%N;
						}
						printf("\nO processo %d esta falho.",i);
					}
					else { //caso contrario o STATE sera definido como 0, será gerdo um tempo aleatorio para o recebimento da mensagem
						nodo[token].STATE[i] = 0;
						msgs[token].tempo[i] = rnd;
						schedule(recv_msg, rnd, i);
						printf("\nO processo %d recebera a mensagem no instante de tempo %d.", i, rnd);
 					}
					i=(i+1)%N; 
					rnd = rand() % 101;
					rnd = rnd + time1();
				}



			break;


			case recv_msg:
				k=0;
				printf("\n\n");
				while (k<N) {
					if (msgs[k].tempo[token]+msgs[k].instante == time1()){ //procura dentre as mensagens, qual delas foi enviada para o processo em questao (token) -- esta verificacao é feita atraves do instante de tempo setado aleatoriamente no momento do envio da mensagem
						
						printf("\nO processo %d recebeu a mensagem do processo %d no tempo %5.1f.", token, k, time1()-msgs[k].instante);
						if (nodo[token].id_aceito < msgs[k].id) { //verifica se ja aceitou uma mensagem com ID menor
							msgs[k].aceite[token] = 1;   //caso tenha recebido com ID menor, e feito o aceite
							nodo[token].id_aceito = msgs[k].id;
							j = (token+1)%N;
							while (j != token) {
								schedule(fault, 0, 3); //escalonamento de um processo falho durante a simulacao
								if (status(nodo[j].id) == 1) { //verifica se ha falha em algum processo
									k = (j+1)%N; //se processo falho, avisa os demais processos e não envia para o processo falho
									while (k != j) {
										nodo[k].STATE[j] = 1;
										k = (k+1)%N;
									}
									printf("\nO processo %d nao envia ACK para o processo %d, pois ele esta falho.",token,j);
								}
								else { //se processo nao-falho escalona para o recebimento do ack imediatamente
									schedule(recv_ack, 0, j);
									printf("\nO processo %d envia ACK para o processo %d.",token,j);
								}							
								j = (j+1)%N;
							}
						}
						else { //se tiver aceito com ID maior e feita a rejeicao da mensagem
							msgs[k].aceite[token] = -1;
							j = (token+1)%N;
							while (j != token) {
								if (status(nodo[j].id) == 1) { //para envio do NACK tambem e feita a verificacao de falhas
									k = (j+1)%N;
									while (k != j) {
										nodo[k].STATE[j] = 1;
										k = (k+1)%N;
									}
									printf("\nO processo %d nao envia NACK para o processo %d, pois ele esta falho.", token, j);
								}
								else {
									schedule(recv_nack, 0, j); //se nao-falho, envia o NACK para o processo
									printf("\nO processo %d envia NACK para o processo %d, pois já aceitou mensagem enviada pelo processo %d.", token, j, nodo[token].id_aceito%N);
								}
								j = (j+1)%N;
							}
						}
					}
				k++;
				}
	
				break;


			case recv_ack:
				k=0;	
				qt_ack = 1;	
				falho = 0;
				// printf("\n\n");		
				while (((msgs[k].id)%N != token)&&(k<N)) k++; //o recebimento do ACK sera avaliado apenas para o processo que enviou a mensagem -- essa verificacao é feita atraves do ID da mensagem
				if (k<N) {
					for(i=0;i<N;i++){ //é feita a contagem de ACK recebida para aquela mensagem
						if (msgs[k].aceite[i] == 1) {
							qt_ack ++; 
						}
					}
					if (qt_ack > 1) 
					printf("\n\nO processo %d recebeu %d ACK.",token, qt_ack-1); // informa no log a quantidade de ACK recebida
				
					for(i=0;i<N;i++) { //verifica a quantidade de processos falhos conhecidos
						if (nodo[token].STATE[i] == 1)
							falho++;
					}
					if (qt_ack > (N-falho)/2) { //a maioria é calculada diminuindo a quantidade de processos falhos
						printf("\nAgora o processo %d recebeu ACK da maioria dos processos não-falhos e esta pronto para o consenso.",token);
						valor_final = msgs[k].valor; //o valor do consenso será o valor informado na mensagem
						nodo[token].vl_consenso = valor_final;				
						k=(token+1)%N;		
						while (k != token){
							if (nodo[token].STATE[k] != 1){ //prepara o consenso apenas para os processo nao-falhos
								schedule(decision,0,k);
								printf("\nA decisao de consenso foi enviada para o processo %d.",k);
							}
							else
								printf("\nA decisao de consenso nao foi enviada ao processo %d, pois ele esta falho.",k);
							k=(k+1)%N;
						}
					}
					else {
						printf("\nAinda não há ACK suficiente para o consenso."); //se nao houver maioria, isso e informado no log
					}
				}
				
				break;

			case recv_nack:
				k = 0;
				while (((msgs[k].id)%N != token)&&(k<N)) k++; //mesma validacao, para que o NACK seja tratado pelo emissor da mensagem
				if (k<N) {
					for(i=0;i<N;i++){
						if (msgs[k].aceite[i] == -1){ 
							id_temp = nodo[i].id_aceito; //a partir do ID da mensagem aceita
							if (token > id_temp%N)	     // e calculado o novo ID para a nova mensagem a ser enviada	
								id_temp = id_temp + (token - id_temp%N);
							else
								id_temp = id_temp + (N-((token - id_temp%N)*-1)); 
							nodo[token].id_atual = id_temp;
							printf("\n\nO processo %d recebeu NACK e fara a geracao de um novo ID para reenviar a mensagem.", token);
								msgs[token].id=-1; //reinicializa-se a mensagem do processo
								for(j=0;j<N;j++)
									msgs[token].tempo[j] = 0;
								for(j=0;j<N;j++)
									msgs[token].aceite[j] = 0;
								msgs[token].valor = -1;
							
							schedule(send_msg,0,token); //escalona o envio de nova mensagem
						}
							
					
					}

				}

				break;

			case fault:
				r = request(nodo[3].id, 3, 0); //o processo 1 (ou 3) sera setado como falho
				nodo[3].vl_consenso = 0; //O processo 1 (ou 3) nao mais participara do consenso 
				break;

			case decision:	
				nodo[token].vl_consenso = valor_final; //o valor do consenso sera atribuído ao processo em questao
				qt = 0;
				for(i=0;i<N;i++) {
					if (nodo[i].vl_consenso == -1) //conta-se quantos processo ja receberam o valor de consenso
						qt++;
				}
				if (qt==0) { //se todos tiverem recebido o consenso esta finalizado
					printf("\n** Consenso Atingido **\n\n");
					exit(1);
				}

		}
	}

}