heis/elev-threaded/netmod.c

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <errno.h>
#include <pthread.h>
#include <time.h>

#include "localsys.h"
#include "netmod.h"
#include "local_queue.h"
#include "globals.h"
#include "mastermod.h"


int ping_tries = 0;

int tcp_write(int sock, char s[]) {
	return send(sock, s, BUFLEN_TCP, 0);
}

void* tcp_read(void* s) {
	int sock = (int) s;
	char rec_buf[BUFLEN_TCP];
	char sen_buf[BUFLEN_TCP];
	//char order_buf[BUFLEN_TCP];
	int flag = -1;
	int cand_isset = 0;
	int err;
	
	while (1) {
		err = recv(sock, rec_buf, BUFLEN_TCP, 0);
		if (err < 0) {
			printf("Connection error recv from socket id %d\n", sock);	
			if (get_role() != SLAVE) set_alive(sock, 0);
			return NULL;	
		}
		else if (err == 0){
			printf("Connection with socket id %d was shut down\n", sock);	
			if (get_role() != SLAVE) set_alive(sock, 0);
			return NULL;	
		}
		flag = (int) rec_buf[0];
		
		switch (flag) {
			case PING:
				if (!cand_isset) {
					if (sock == get_master_cand_sock()) {
						sen_buf[0] = (char) CANDIDATE;
						sen_buf[1] = (char) 1;
						cand_isset = 1;
						if (write(sock, sen_buf, BUFLEN_TCP) == -1) printf("Couldnt set candidate!\n");
					}
					else {
						sen_buf[0] = (char) CANDIDATE;
						sen_buf[1] = (char) 0;
						if (write(sock, sen_buf, BUFLEN_TCP) == -1) printf("Couldnt set candidate!\n");
						cand_isset = 1;
					}
				}
				sen_buf[0] = (char) PONG;
				write(sock, sen_buf, BUFLEN_TCP);
				break;
			
			case PONG:
				ping_tries = 0;
				break;
			
			case STATUS:
				set_elev_state(sock, rec_buf);
				printf("Status update: state %d, floor %d, end destination %d\n", rec_buf[1], rec_buf[2], rec_buf[3]);
				
				break;
			
			case ORDER:
				printf("New order. Call %d, floor %d\n", rec_buf[1], rec_buf[2]);
				/*order_buf[0] = (char) ORDER;
				order_buf[1] = rec_buf[1];
				order_buf[2] = rec_buf[2];*/
				sen_buf[0] = (char) REQUEST;
				sen_buf[1] = (char) rec_buf[1];
				sen_buf[2] = (char) rec_buf[2];
				
				int elev = find_opt_elev((int) rec_buf[1], (int) rec_buf[2]);
				printf("Optimal elevator has socketid %d\n", elev);	
				
				if (elev != -1) {
					if (write(elev, sen_buf, BUFLEN_TCP) == -1) printf("Couldnt deliver order!\n");
				}
				else add_to_queue((int) rec_buf[1], (int) rec_buf[2]);
				break;
			
			case REQUEST:
				elev_set_button_lamp((int) rec_buf[1], (int) rec_buf[2], 1);
				printf("New request\n");
				add_to_queue((int) rec_buf[1], (int) rec_buf[2]);
				break;
			
			case CANDIDATE:
				if ((int) rec_buf[1]) {
					printf("Taking role as master candidate\n");
					master_init(sock);
					set_role(MASTER_CAND);
				}
				else {
					printf("Taking role as slave\n");
					set_role(SLAVE);
				}
				break;
						
			case (-1):
				printf("Flag is not set\n");
				break;
				
		}
		
	}
	return NULL;
}

void* ping_master(void* s) {	
	int socket = (int) s;
	char msg[BUFLEN_TCP];
	msg[0] = (char) PING;
	while(1) {
		//printf("-> Ping...\n");
		write(socket, msg, BUFLEN_TCP);	
		if(ping_tries == PING_ATTEMPTS) {
			printf("Lost connection to master\n");
			break;
		}
		ping_tries++;
		sleep(TCP_PING_INTERVAL);
	}
	
	printf("No pong response!\n");
	net_init();
	return NULL;
	
}

void* tcp_listen(void* timeout_id) {
	struct sockaddr_in socklisten_addr;
	struct sockaddr_in master_addr;
	int sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
	socklen_t addrlen;
	
	if (sock == -1) {
		perror("socket");
		exit(1);
	}
	
	memset(&socklisten_addr, 0, sizeof socklisten_addr);
	socklisten_addr.sin_family = AF_INET;
	socklisten_addr.sin_port = htons(PORT);
	socklisten_addr.sin_addr.s_addr = INADDR_ANY;
	
	int reuse = 1;
	if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof reuse) == -1) {
		perror("setsockopt");
		close(sock);
		exit(1);
	}
	
	if (bind(sock, (struct sockaddr *)&socklisten_addr, sizeof socklisten_addr) == -1) {
		perror("bind");
		close(sock);
		//exit(1);
	}
	
	if (listen(sock, 1) < 0) {
		perror("listen");
		exit(1);
	}
	
	printf("Waiting to accept...\n");	
	addrlen = sizeof master_addr;		
	int sock_master = accept(sock, (struct sockaddr *)&master_addr, &addrlen);
	
	printf("Connected to master\n");
	set_master_sock(sock_master);
	set_role(SLAVE);

	pthread_t read_th, ping_th;
	pthread_create(&read_th, NULL, tcp_read, (void*) sock_master);
	pthread_create(&ping_th, NULL, ping_master, (void*) sock_master);

	printf("Network initialization complete.\n");
	
	close(sock);
	
			
	return NULL;
	
}


void tcp_connect(char ip[]) {
	struct sockaddr_in sockconn_addr;
	int sockid = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

	if (sockid == -1) {
		perror("socket");
		exit(1);
	}
	
	memset(&sockconn_addr, 0, sizeof sockconn_addr);
	sockconn_addr.sin_family = AF_INET;
	sockconn_addr.sin_port = htons(PORT);
	
	if(inet_pton(AF_INET, ip, &sockconn_addr.sin_addr) == 0) {
		perror("Not a valid ip");
		close(sockid);
		exit(1);
	}
	
	printf("Connecting to %s ...\n", ip);
	if (connect(sockid, (struct sockaddr *)&sockconn_addr, sizeof sockconn_addr) == -1) {
		perror("connect");
		close(sockid);
		exit(1);
	}
	
	printf("Connected to slave with socket ID %d\n", sockid);
	add_slave(sockid);
	
	pthread_t read_th;
	pthread_create(&read_th, NULL, tcp_read, (void*) sockid);
}

void udp_broadcast() {
	int sock;
	struct sockaddr_in broadcast_addr;
	struct hostent *he;
	int broadcast = 1;
	
	char dummymsg[BUFLEN_UDP] = "\0"; 
	
	if ((he = gethostbyname(IPGROUP)) == NULL) {
		perror("gethostbyname");
		exit(1);
	}
	
	if ((sock = socket(AF_INET, SOCK_DGRAM, 0)) == -1) {
		perror("socket");
		exit(1);
	}
	
	// Allow broadcast
	if (setsockopt(sock, SOL_SOCKET, SO_BROADCAST, &broadcast, sizeof broadcast) == -1) {
		perror("setsockopt (SO_BROADCAST)");
		close(sock);
		exit(1);
	}
	
	broadcast_addr.sin_family = AF_INET;
	broadcast_addr.sin_port = htons(PORT);
	broadcast_addr.sin_addr = *((struct in_addr *)he->h_addr);
	memset(broadcast_addr.sin_zero, '\0', sizeof broadcast_addr.sin_zero);
	
	
	printf("Broadcasting to %s\n", IPGROUP);
	if (sendto(sock, dummymsg, strlen(dummymsg), 0, (struct sockaddr *)&broadcast_addr, sizeof broadcast_addr) == -1) {
		perror("sendto");
		exit(1);
	}
	
	close(sock);
}

void* udp_listen() {
	int sock;
	struct addrinfo hints;
	struct addrinfo *result, *rp;
	struct sockaddr_in slave_addr;
	char buf[BUFLEN_UDP];
	socklen_t addrlen = sizeof(struct sockaddr);
	
	memset(&hints, 0, sizeof hints);
	hints.ai_family = AF_INET;
	hints.ai_socktype = SOCK_DGRAM;
	hints.ai_flags = AI_PASSIVE;
	
	if (getaddrinfo(NULL, PORTSTR, &hints, &result) != 0) {
		perror("getaddrinfo");
		exit(1);
	}
	
	for (rp = result; rp != NULL; rp = rp->ai_next) {
		sock = socket(rp->ai_family, rp->ai_socktype, 0);
		
		if (sock == -1)
			continue;
		
		if (bind(sock, rp->ai_addr, rp->ai_addrlen) == 0)
			break;
			
		close(sock);
	}
	
	freeaddrinfo(result);
	printf("Waiting for slaves...\n");
	
	while(1) {
		
		if (recvfrom(sock, buf, BUFLEN_UDP, 0, (struct sockaddr *)&slave_addr, &addrlen) == -1) {
			perror("recvfrom");
			exit(1);
		}
		
		printf("Found new slave\n");
		tcp_connect(inet_ntoa(slave_addr.sin_addr));
	}
	return NULL;	
}

void* timeout(void* tcp_listen_id) {
	sleep(TCP_LISTEN_TIMEOUT);
	
	if (get_role() == MASTER) {
		printf("Master is not present\nStarting as master...\n");
		master_init(-1);
		pthread_cancel((pthread_t) tcp_listen_id);
		printf("Start listening for slave broadcasts through UDP\n");
		pthread_t udp_listen_th;
		pthread_create(&udp_listen_th, NULL, udp_listen, NULL);
		printf("Network initialization complete.\n");
		
	}
		
	return NULL;	
}

void net_init() {
	printf("Starting search for master\n");
	
	if (get_role() == MASTER_CAND) {
		set_role(MASTER);
		set_master_flag();
		printf("I am master candidate. Taking role as master...\n");
		pthread_t udp_listen_th;
		pthread_create(&udp_listen_th, NULL, udp_listen, NULL);
	}
	else {
		pthread_t tcp_listen_th, timeout_th;	
		pthread_create(&tcp_listen_th, NULL, tcp_listen, (void*) timeout_th);
		pthread_create(&timeout_th, NULL, timeout, (void*) tcp_listen_th);
	
		udp_broadcast();	

		pthread_join(timeout_th, NULL);
	}
}