sysdis-project/central-manager/central-manager.c

#include "central-manager.h"

/*
*
* @argv : {0:program name}
*
* @history
*   [0] Initialisation des variables globales
*   [1] Lancement des THREADS d'écoute
*       1. On démarre le SERVEUR TCP d'écoute globale
*       2. On démarre le SERVEUR UDP d'écoute globale
*   [2] On attends la fin de tous les THREADS
*   [3] On libère les variables globale
*
*/
int main(int argc, char* argv[]){

	printf("**** Execution tree structure\n");
	printf("** [procedureName]\n");
	printf("** {threadName}\n");
	printf("** [parent]{child}[subchild] Description\n\n\n");
	printf("**** Execution tree\n");

	/* [0] Initialisation des variables globales
	=========================================================*/
	sgca_data         = (struct plane*) malloc( sizeof(struct plane) );
	sgca_update       = (struct plane*) malloc( sizeof(struct plane) );
	sgca_index.data   = 0;
	sgca_index.update = 0;

	/* Variables locales */
	struct listn_thrd_arg tcp_listn_arg = { SERV_HOST, TCP_LIST };
	struct listn_thrd_arg udp_vterm_arg = { MCST_VTER, UDP_VTER };
	struct listn_thrd_arg udp_cterm_arg = { MCST_CTER, UDP_CTER };


	/* [1] Lancement des THREADS d'écoute
	=========================================================*/
	/* (1) Ecoute TCP */
	pthread_create(&listenManagers[0], NULL, LISTEN_TCP, (void*) &tcp_listn_arg);
	if( DEBUGMOD&THR ) printf("{tcp_listn}              démarré\n");

	/* (2) Ecoute UDP multicast */
	// pthread_create(&listenManagers[1], NULL, LISTEN_UDP, (void*)(intptr_t) UDP_MCST);
	// if( DEBUGMOD&THR ) printf("{udp_mcast} émarré\n\n");

	/* (3) Ecoute UDP viewTerm */
	pthread_create(&listenManagers[2], NULL, LISTEN_UDP, (void*) &udp_vterm_arg);
	if( DEBUGMOD&THR ) printf("{udp_vterm}              démarré\n");

	/* (4) Ecoute UDP ctrlTerm */
	pthread_create(&listenManagers[3], NULL, LISTEN_UDP, (void*) &udp_cterm_arg);
	if( DEBUGMOD&THR ) printf("{udp_cterm}              démarré\n");


	/* [2] On attends la fin de tous les THREADS
	==========================================================*/
	for( char i = 0 ; i < 4 ; i++ )
		pthread_join(listenManagers[(int)i], NULL);


	/* [3] On libère les variables globales
	==========================================================*/
	free(sgca_data);
	free(sgca_update);
}






/* Attente de connection TCP
*
* @history
*   [0] Initialisation des variables
*   [1] On démarre le SERVEUR TCP d'écoute globale
*   [2] Attente d'une demande de connection TCP -> création d'un THREAD
*   [3] On attends la fin de tous les THREADS
*   [4] On ferme la SOCKET d'écoute TCP globale
*
*/
void* LISTEN_TCP(void* THREADABLE_ARG){
	/* [0] Initialisation des variables
	==========================================================*/
	int                    CLIENT_SOCKET;                         // contiendra la socket TCP à envoyer sur un THREAD
	struct sockaddr_in     clientInfo;                            // contiendra les infos client
	socklen_t              len;                                   // taille de la socket
	int                    index, i;                              // compteurs
	struct listn_thrd_arg* arg = THREADABLE_ARG;                  // Addr + Port serveur


	// retour de @DROP_TCP_SERVER
	int LISTENSOCK;   // contiendra la socket d'écoute TCP


	/* [1] On démarre le SERVEUR TCP d'écoute globale
	==========================================================*/
	if( DROP_TCP_SERVER(arg->port, &LISTENSOCK) < 0 ){

		if( DEBUGMOD&SCK ) printf("{tcp_listn}              Erreur création socket d'écoute\n");

		// On ferme la SOCKET d'écoute globale
		printf("{tcp_listn}              FERMETURE SOCKET D'ECOUTE TCP!\n");
		close(LISTENSOCK);

		return NULL;
	}



	printf("{tcp_listn}              listen on %s:%d\n", arg->addr, arg->port);



	/* [2] Attente d'une demande de connection, pour création d'un THREAD
	============================================================================*/
	while( 1 ){

		/* 1. On initialise les SOCKET en attendant la connexion et le rang du "manager" inactif */
		CLIENT_SOCKET = -1;
		index         = -1;

		/* 2. On attends une connection TCP */
		CLIENT_SOCKET = accept(LISTENSOCK, (struct sockaddr*) &clientInfo, &len);

		/* 3. Si erreur, on attend une nouvelle connection */
		if( CLIENT_SOCKET < 0 ){
			if( DEBUGMOD&SCK ) printf("{tcp_listn}              accept: Erreur connection\n");
			break;
		}

		/* 4. On cherche un "manager" libre (inactif) */
		for( i = 0 ; i < MAX_TCP_THR ; i++ )
			if( activeTCPManagers[i] == 0 ){ index = i; break; }

		// si on a trouvé un "manager" libre
		if( index != -1 ){

			/* 5. On lance un thread pour le traitement de ce client */
			pthread_create(&TCPManagers[index], NULL, managePlane, (void*)(intptr_t) CLIENT_SOCKET);

			if( DEBUGMOD&THR ) printf("{tcp_listn}{com}(%d)      démarré\n", index);

			/* 6. On signale que ce "manager" est maintenant actif */
			activeTCPManagers[index] = 1;

		}else
			if( DEBUGMOD&THR ) printf("{tcp_listn}              Aucun thread libre\n");

	}


	/* [3] On attends la fin de tous les THREADS
	==========================================================*/
	for( i = 0 ; i < MAX_TCP_THR ; i++ )
		pthread_join(TCPManagers[i], NULL);


	/* [4] On ferme la SOCKET d'écoute globale
	==========================================================*/
	printf("{tcp_listn}              FERMETURE SOCKET D'ECOUTE TCP!\n");
	close(LISTENSOCK);

	return NULL;
}







/* Attente de connection UDP
*
* @history
*   [0] Initialisation des variables
*   [1] On démarre le SERVEUR UDP d'écoute globale
*   [2] On attends un client
*   [3] On gère la requête
*       1. On parse la requête
*       2. Si demande de socket de communication
*       	1. Création socket port random
*       	2. On récupère le port en question
*       3. On envoie la réponse
*   [4] On envoie la réponse
*   [5] On démarre un thread de gestion
*   [N] On ferme la SOCKET d'écoute globale
*
*/
void* LISTEN_UDP(void* THREADABLE_ARG){
	/* [0] Initialisation des variables
	==========================================================*/
	int                    CLIENT_SOCKET;                            // contiendra la socket UDP à envoyer sur un THREAD
	struct sockaddr_in     listenInfo;                               // contiendra les infos de la socket LISTEN
	struct sockaddr_in     comInfo;                                  // contiendra les infos de la socket COM
	struct sockaddr_in     clientInfo;                               // contiendra les infos client
	socklen_t              len;                                      // taille de la socket
	int                    read;                                     // compteurs
	char                   buffer[MAX_BUF_LEN];                      // buffer requête
	struct bind_header     request;                                  // requête parsée
	int                    i, index;                                 // compteurs
	struct listn_thrd_arg* arg = THREADABLE_ARG;                     // Addr + Port serveur
	char                   entity[9+1];

	if( strcmp(arg->addr, MCST_VTER) == 0 && arg->port == UDP_VTER ) strcpy(entity, "udp_vterm");
	else                                                             strcpy(entity, "udp_cterm");

	// retour de @DROP_UDP_SERVER
	int SOCKET;


	/* [1] On démarre le SERVEUR UDP d'écoute globale
	==========================================================*/
	if( DROP_UDP_SERVER(arg->addr, arg->port, &SOCKET, &listenInfo, strcmp(arg->addr, SERV_HOST) != 0) < 0 ){

		if( DEBUGMOD&SCK ) printf("{%s}              Erreur de création socket d'écoute\n", entity);

		// On ferme la SOCKET d'écoute globale
		printf("{%s}              FERMETURE SOCKET D'ECOUTE UDP!\n", entity);
		close(SOCKET);

		return NULL;
	}

	printf("{%s}              listen on %s:%d\n", entity, arg->addr, arg->port);



	/* [2] Attente de connection
	============================================================================*/
	while( 1 ){

		/* 0. On initialise les SOCKET en attendant la connexion d'un client */
		CLIENT_SOCKET = -1;
		index         = -1;

		/* 1. On attends une connection UDP */
		len = sizeof(struct sockaddr_in);
		read = recvfrom(SOCKET, buffer, MAX_BUF_LEN, 0, (struct sockaddr*) &clientInfo, &len);

		/* 2. Si erreur reception ou taille incorrecte -> retour à l'écoute */
		if( read < BINDHDR_LEN ){
			if( DEBUGMOD&BUF ) printf("{%s}              read('%s') = %d bytes (expected: %d)\n", entity, buffer, read, (int) (BINDHDR_LEN) );
			continue;
		}

		/* 3. On récupère l'adresse IP du client */
		if( DEBUGMOD&SCK ) printf("{%s}              '%s' connecté\n", entity, inet_ntoa(clientInfo.sin_addr));



		/* [3] Gestion de la requête
		=========================================================*/
		/* 1. On parse la requête */
		memcpy(&request.flags, buffer,                 sizeof(char));
		memcpy(&request.addr,  buffer+sizeof(char),    sizeof(char)*15);
		memcpy(&request.port,  buffer+sizeof(char)*16, sizeof(unsigned short));
		printf("{%s}              received: bind_header{flag = %d; addr = '%s'; port = %d}\n", entity, (int) request.flags, request.addr, request.port);

		/* 2. Si on veut un port de communicatin */
		if( request.flags&BINDHEAD_SCK ){

			/* 2.1 On bind une socket sur un port random */
			if( DROP_UDP_SERVER(SERV_HOST, 0, &CLIENT_SOCKET, &comInfo, 0) < 0 ){

				if( DEBUGMOD&SCK ) printf("{%s}              Erreur de création de la socket COM\n", entity);

				// On ferme la SOCKET CLIENT
				close(CLIENT_SOCKET);

				// On retire le flags PORT pour dire qu'on a pas pu ouvrir une socket de comm.
				request.flags -= BINDHEAD_SCK;
			}

		}


		/* 2.2 Si on veut on port de communication */
		if( request.flags&BINDHEAD_SCK ){

			/* On récupère le port de la socket de communication */
			len = sizeof(struct sockaddr_in);
			if( getsockname(CLIENT_SOCKET, (struct sockaddr*) &comInfo, &len) < 0 ){

				if( DEBUGMOD&SCK ) printf("{%s}              Erreur de recherche du port COM ouvert\n", entity);

				close(CLIENT_SOCKET);

				// On retire le flags PORT pour dire qu'on a pas pu ouvrir une socket de comm.
				request.flags -= BINDHEAD_SCK;

			// Si on a le port -> on le met dans la reponse
			}else{
				strcpy(request.addr, SERV_HOST);
				request.port = htons(comInfo.sin_port);
				if( DEBUGMOD&SCK ) printf("{%s}{udp_com}     socket opened on %s:%d\n", entity, request.addr, request.port);
			}

		}


		/* [4] Envoi de la réponse
		=========================================================*/
		bzero(buffer, MAX_BUF_LEN);
		memcpy(buffer,                 &request.flags, sizeof(char));
		memcpy(buffer+sizeof(char),    &request.addr,  sizeof(char)*15);
		memcpy(buffer+sizeof(char)*16, &request.port,  sizeof(unsigned short));

		len = sizeof(struct sockaddr_in);
		if( sendto(SOCKET, buffer, BINDHDR_LEN/sizeof(char) + 1, 0, (struct sockaddr*) &clientInfo, len) < 0 ){
			printf("{%s}              Impossible de répondre au client!\n", entity);
			continue;
		}

		printf("{%s}              sent: bind_header{flag = %d; addr = '%s'; port = %d}\n", entity, (int) request.flags, request.addr, request.port);



		/* [5] On démarre la tache sur un thread dédié
		=========================================================*/
		/* 1. On cherche un "manager" libre (inactif) */
		for( i = 0 ; i < MAX_UDP_THR ; i++ )
			if( activeUDPManagers[i] == 0 ){ index = i; break; }

		/* 2. si on a trouvé un "manager" libre */
		if( index != -1 ){

			/* 2.1. On lance un thread pour le traitement de ce client */
			pthread_create(&UDPManagers[index], NULL, manageTerminal, (void*)(intptr_t) CLIENT_SOCKET);

			if( DEBUGMOD&THR ) printf("{%s}{udp_com}(%d)  démarré\n", entity, index);

			/* 2.2. On signale que ce "manager" est maintenant actif */
			activeUDPManagers[index] = 1;

		}else
			if( DEBUGMOD&THR ) printf("{%s} Aucun thread UDP libre!\n", entity);

	}



	/* [n] On ferme la SOCKET d'écoute globale
	==========================================================*/
	printf("{%s}              FERMETURE SOCKET D'ECOUTE UDP!\n", entity);
	close(SOCKET);

	return NULL;
}








/* Gestion d'une connexion PLANE
*
* @THREADABLE_SOCKET<void*>			SOCKET de la connexion client
*
* @history
* [1] Initialisation des variables
* @loop
*   (2) Attente de requête
*   (3) Gestion de la requête
*   (4) Envoi de la réponse
* [5] On libère la mémoire
* [6] Fermeture de la connection (SOCKET)
* [n] Arrêt du THREAD
* 	1. On met à jour "activeManagers"
* 	2. On arrête le THREAD
*
*/
void* managePlane(void* THREADABLE_SOCKET){

	/* [1] Initialisation des variables
	=========================================================*/
	/* 1. Variables utiles */
	int    read, i, index, pindex; // compteurs
	char   buffer[MAX_BUF_LEN];    // buffer
	struct plane       data;       // données de l'avion

	/* 2. On récupère la socket */
	int TCP_SOCKET = (intptr_t) THREADABLE_SOCKET; // Socket client

	/* 3. On récupère le rang du thread parmi les "managers" */
	index = -1;
	for( i = 0 ; i < MAX_TCP_THR ; i++ )
		if( TCPManagers[i] == pthread_self() ){ index = i; break; }

	// Erreur de thread
	if( index == -1 ){
		if( DEBUGMOD&THR ) printf("{tcp_com}(%d)             Unknown thread index. Aborting\n", index);
		pthread_exit(NULL);
	}


	while( 1 ){

		/* (2) Récupération de la requête
		---------------------------------------------------------*/
		/* 1. On lit sur la socket */
		read = recv(TCP_SOCKET, buffer, MAX_BUF_LEN, 0);

		/* 2.1. Si erreur reception (-1:erreur, 0:fermeture client propre) */
		if( read <= 0 ){

			if( DEBUGMOD&BUF ) printf("{tcp_com}(%d)             read: %d -> must exit thread\n", index, read);
			break;

		}


		/* 2.2. Si message trop court */
		if( read < PLANE_LEN ){
			if( DEBUGMOD&BUF ) printf("{tcp_com}(%d)             read: %d (expected: %d)\n", index, read, (int) PLANE_LEN);
			continue;
		}


		/* 3. On parse la requête*/
		memcpy(&data.code, buffer+sizeof(char)*0+sizeof(int)*0, sizeof(char)*6);
		memcpy(&data.x,    buffer+sizeof(char)*6+sizeof(int)*0, sizeof(int));
		memcpy(&data.y,    buffer+sizeof(char)*6+sizeof(int)*1, sizeof(int));
		memcpy(&data.z,    buffer+sizeof(char)*6+sizeof(int)*2, sizeof(int));
		memcpy(&data.spd,  buffer+sizeof(char)*6+sizeof(int)*4, sizeof(int));
		memcpy(&data.cap,  buffer+sizeof(char)*6+sizeof(int)*3, sizeof(int));
		printf("{tcp_com}(%d)             received: plane_req{code = '%s'; x = %d; y = %d; z = %d; cap = %d; spd = %d}\n", index, data.code, data.x, data.y, data.z, data.cap, data.spd);


		/* (3) Gestion de la requête -> enregistrement
		---------------------------------------------------------*/
		pindex = -1;

		/* 1. On regarde si l'avion existe */
		for( i = 0 ; i < sgca_index.data ; i++ ){

			pindex = sgca_index.data;

			// On ajoute une entrée à data
			sgca_data = (struct plane*) realloc(sgca_data, sizeof(struct plane)*sgca_index.data + 1);
			sgca_index.data++;

		}

		/* 3. On met à jour les données*/
		memcpy(&sgca_data[pindex], &data, sizeof(struct plane));




		/* (4) Envoi de la réponse
		---------------------------------------------------------*/
		printf("send: %d\n", (int) send(TCP_SOCKET, "coucou\n", 8, 0));



	}


	/* [5] On libère la mémoire
	=========================================================*/



	/* [6] Fermeture de la connection (SOCKET)
	=========================================================*/
	printf("{tcp_com}(%d)             Fermeture de la socket de communication!\n", index);
	close(TCP_SOCKET);



	/* [n] Arrêt du THREAD
	============================================================================*/
	/* 1. On met à jour "activeManagers" */
	if( index != -1 )
		activeTCPManagers[index] = 0;

	/* 2. On arrête le THREAD */
	if( DEBUGMOD&THR ) printf("{tcp_com}(%d)             libéré\n", index);
	pthread_exit(NULL);
}






/* Gestion d'une connexion TERMINAL
*
* @THREADABLE_SOCKET<void*>			SOCKET de la connexion client
*
* @history
* [1] Initialisation des variables
* [2] Récupération de la requête
* [3] Traitement de la requête
* [4] Création de la réponse
* [5] Envoi de la réponse
* [6] On vide les buffers
* [7] Fermeture de la connection (SOCKET)
* [n] Arrêt du THREAD
* 	1. On récupère le rang dans les "managers"
* 	2. On met à jour "activeManagers"
* 	3. On arrête le THREAD
*
*/
void* manageTerminal(void* THREADABLE_SOCKET){

	/* [1] Initialisation des variables
	=========================================================*/
	int read; // compteur
	struct sockaddr_in clientInfo;
	socklen_t len;
	int UDP_SOCKET = (intptr_t) THREADABLE_SOCKET; // Socket client
	char request[MAX_BUF_LEN]; // Requête
	// char response[MAX_BUF_LEN]; // Réponse

	do{

		printf("{udp_x-term}{udp_com} waiting for terminal request\n");
		/* [2] Récupération de la requête
		=========================================================*/
		/* 1. On lit sur la socket */
		len = sizeof(struct sockaddr_in);
		read = recvfrom(UDP_SOCKET, request, MAX_BUF_LEN, 0, (struct sockaddr*) &clientInfo, &len);

		/* 2. Si erreur reception */
		if( DEBUGMOD&BUF ) printf("{udp_x-term}{udp_com} READ = %d\n", read);

		if( read < 0 )
			continue;

		/* 3. On désérialise la requête*/
		printf("{udp_x-term}{udp_com} TERMINAL Request(%d bytes) : '%s'\n", read, request);

		/* [3] Gestion de la requête
		=========================================================*/


		/* [4] Envoi reponse
		=========================================================*/
		strcpy(request+strlen(request), "-bla\0");
		send(UDP_SOCKET, request, strlen(request), 0);


	}while( 0 );

	/* [n] Arrêt du THREAD
	============================================================================*/
	/* 1. On récupère le rang dans les "managers" */
	int i, index = -1;
	for( i = 0 ; i < MAX_UDP_THR ; i++ )
		if( UDPManagers[i] == pthread_self() ){ index = i; break; }

	/* 2. On met à jour "activeManagers" */
	if( index != -1 )
		activeUDPManagers[index] = 0;

	/* 3. On arrête le THREAD */
	if( DEBUGMOD&THR ) printf("{udp_x-term}{udp_com}(%d) libéré\n", index);
	pthread_exit(NULL);
}