open-card-table/oct_networking.c

#include <stdio.h>
#include <stdlib.h>
#include <netdb.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <lua5.3/lualib.h>
#include <lua5.3/lauxlib.h>
#include <string.h>
#include <unistd.h>
#include <assert.h>

#include "oct_log.h"
#include "oct_networking.h"

struct {
	struct oct_network_q recv_queue;
	struct oct_network_q send_queue;

	int sfd;
} oct_network_node;

int oct_network_q_enqueue(struct oct_network_q* q, char* addr, char* port, char* msg) {
	struct oct_network_q_entry* e = (struct oct_network_q_entry*)malloc(sizeof(struct oct_network_q_entry));
	if (!e) {
		OCT_LOG_ERROR("Could not initialize new send queue entry");
		return 0;
	}
	strncpy(e->message.addr, addr, NI_MAXHOST);
	strncpy(e->message.port, port, NI_MAXSERV);
	strncpy(e->message.buffer, msg, BUFFER_SIZE);
	e->next = NULL;
	e->prev = NULL;

	// If the queue is empty
	if (q->size == 0) {
		q->first = e;
		q->last = e;
	}
	else {
		e->next = q->first;
		q->first = e;
		e->next->prev = e;
	}

	q->size++;
	return 1;
}

struct oct_network_q_message oct_network_q_dequeue(struct oct_network_q* q) {
	// Nothing left in queue signalled by a struct with the following addr, port, buffer
	struct oct_network_q_message ret;
	strncpy(ret.addr, "0", 2);
	strncpy(ret.port, "0", 2);
	strncpy(ret.buffer, "0", 2);
	
	if (q->size == 0) {
		return ret;
	}
	// Otherwise, first copy fields over
	strncpy(ret.addr, q->last->message.addr, NI_MAXHOST);
	strncpy(ret.port, q->last->message.port, NI_MAXSERV);
	strncpy(ret.buffer, q->last->message.buffer, BUFFER_SIZE);
	
	// Then, remove the last element
	struct oct_network_q_entry* temp = q->last;
	q->last = q->last->prev;
	if (q->last) {
		q->last->next = NULL;
	}
	free(temp);

	q->size--;
	return ret;
}

void oct_network_q_init(struct oct_network_q* q) {
	q->first = NULL;
	q->last = NULL;
	q->size = 0;
}

void oct_network_q_deinit(struct oct_network_q* q) {
	while (q->first) {
		oct_network_q_dequeue(q);
	}
}

// Address book currently implemented as a doubly-linked list with O(1) inserts, O(n) accesses
// Ideal implementation is hash table with O(1) inserts, accesses
// However, we'll see if this works for now
struct {
	unsigned int size;
	struct oct_network_ab_entry* first;
	struct oct_network_ab_entry* last;
} oct_network_ab;

// Returns a live sfd, need to close eventually
static int oct_network_init_socket(char* addr, char* port) {
	int s;
	int sfd;
	struct addrinfo hints;
	struct addrinfo* result;
	struct addrinfo* rp;
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = AF_UNSPEC;    /* Allow IPv4 or IPv6 */
	hints.ai_socktype = SOCK_DGRAM; /* Datagram socket */
	hints.ai_flags = 0;
	hints.ai_protocol = 0;          /* Any protocol */
	//s = getaddrinfo(oct_network_node.send_addr, oct_network_node.send_port, &hints, &result);
	s = getaddrinfo(addr, port, &hints, &result);
	if (s != 0) {
		OCT_LOG_WARNING("getaddrinfo failed on %s:%s\n", addr, port);
		return -1;
	}

	/* getaddrinfo() returns a list of address structures.
	   Try each address until we successfully connect(2).
	   If socket(2) (or connect(2)) fails, we (close the socket
	   and) try the next address. */

	for (rp = result; rp != NULL; rp = rp->ai_next) {

		sfd = socket(rp->ai_family, rp->ai_socktype,
					 rp->ai_protocol);
		if (sfd == -1)
			continue;

		if (connect(sfd, rp->ai_addr, rp->ai_addrlen) != -1)
			break;                  /* Success */

		close(sfd);
	}
	freeaddrinfo(result);           /* No longer needed */
	if (rp == NULL) {
		OCT_LOG_WARNING("Could not connect to %s:%s", addr, port);
		return -1;
	}
	return sfd;
}

int oct_network_node_init(char* port, lua_State* L) {
	// Initialize listening socket
	struct addrinfo hints;
	memset(&hints, 0, sizeof(hints));
	hints.ai_family = AF_UNSPEC; // ipv4 or ipv6
	hints.ai_socktype = SOCK_DGRAM; // datagram
	hints.ai_flags = AI_PASSIVE; // any IP address
	hints.ai_protocol = 0; // any protocol
	hints.ai_canonname = NULL;
	hints.ai_addr = NULL;
	hints.ai_next = NULL;

	struct addrinfo* result;
	int s = getaddrinfo(NULL, port, &hints, &result);
	if (s != 0) {
		//fprintf(stderr, "getaddrinfo: %s\n", gai_strerror(s));
		return 0;
	}

	/* getaddrinfo() returns a list of address structures.
	   Try each address until we successfully bind(2).
	   If socket(2) (or bind(2)) fails, we (close the socket
	   and) try the next address. */

	struct addrinfo *rp;

	for (rp = result; rp != NULL; rp = rp->ai_next) {
		oct_network_node.sfd = socket(rp->ai_family, rp->ai_socktype,
					 rp->ai_protocol);
		if (oct_network_node.sfd == -1)
			continue;

		if (bind(oct_network_node.sfd, rp->ai_addr, rp->ai_addrlen) == 0)
			break;                  /* Success */

		close(oct_network_node.sfd);
	}

	freeaddrinfo(result);
	if (!rp) {
		return 0;
	}
	OCT_LOG_INFO("Listening on 127.0.0.1 port %s", port);

	// Initialize send queue
	oct_network_q_init(&oct_network_node.send_queue);
	oct_network_q_init(&oct_network_node.recv_queue);

	// Push lua functions
	lua_pushcfunction(L, oct_network_recv_msg_lua);
	lua_setglobal(L, "oct_recv");
	lua_pushcfunction(L, oct_network_send_msg_lua);
	lua_setglobal(L, "oct_send");
	
	return 1;
}

void oct_network_node_deinit() {
	oct_network_ab_deinit();
	oct_network_q_deinit(&oct_network_node.send_queue);
	oct_network_q_deinit(&oct_network_node.recv_queue);
}


// This runs every loop, filling the receive buffer if needed
int oct_network_recv_msgs() {
	struct sockaddr_storage peer_addr;
	socklen_t peer_addrlen = sizeof(peer_addr);
	char tmp_addr[NI_MAXHOST];
	char tmp_port[NI_MAXSERV];
	char tmp_buffer[BUFFER_SIZE];

	// Receive up to OCT_NETWORK_MAX_RECVS messages
	// recvfrom returns -1 if nothing was received
	// Need MSG_DONTWAIT flag for nonblocking
	int i = 0;
	ssize_t msg_size = recvfrom(oct_network_node.sfd, tmp_buffer, BUFFER_SIZE-1, MSG_DONTWAIT, (struct sockaddr *) &peer_addr, &peer_addrlen); 
	while ((msg_size > 0) && (i < OCT_NETWORK_MAX_RECVS)) {
		// recvfrom does NOT place the null byte - place it ourselves
		tmp_buffer[msg_size] = '\0';
		OCT_LOG_DEBUG("Received message!");
		int s = getnameinfo((struct sockaddr *) &peer_addr,
							peer_addrlen, tmp_addr, NI_MAXHOST, tmp_port, NI_MAXSERV, NI_NUMERICSERV);
		if (s != 0) {
			OCT_LOG_WARNING("Received message from unknown host: %s", tmp_buffer);
			return -1;
		}
		oct_network_q_enqueue(&oct_network_node.recv_queue, tmp_addr, tmp_port, tmp_buffer);
		i++;
		msg_size = recvfrom(oct_network_node.sfd, tmp_buffer, BUFFER_SIZE-1, MSG_DONTWAIT, (struct sockaddr *) &peer_addr, &peer_addrlen);
	}
	return i;
}


int oct_network_send_msgs() {
	while (oct_network_node.send_queue.size != 0) {
		struct oct_network_q_message m = oct_network_q_dequeue(&oct_network_node.send_queue);
		struct oct_network_ab_entry* e;
		e = oct_network_ab_find(m.addr, m.port);
		if (!e) {
			OCT_LOG_DEBUG("Not in address book: %s:%s", m.addr, m.port);
			e = oct_network_ab_insert(m.addr, m.port);
			if (!e) {
				// Give up...
				OCT_LOG_ERROR("Error: Could not insert address book entry");
				return 0;
			}
		}
		else {
			OCT_LOG_DEBUG("Found in address book: %s:%s", m.addr, m.port);
	}	
		int sfd = e->sfd;
		
		ssize_t length = strlen(m.buffer);
		OCT_LOG_DEBUG("Sending message: %s", m.buffer);
		if (write(sfd, m.buffer, length) != length) {
			OCT_LOG_WARNING("Partial write when sending message");
			return -1;
		}
	}
	return 1;
}

int oct_network_recv_msg_lua(lua_State* L) {
	if (oct_network_node.recv_queue.size != 0) {
		struct oct_network_q_message m = oct_network_q_dequeue(&oct_network_node.recv_queue);
		lua_pushstring(L, m.buffer);
		lua_pushstring(L, m.addr);
		lua_pushstring(L, m.port);
	}
	else {
		lua_pushstring(L, "");
		lua_pushstring(L, "");
		lua_pushstring(L, "");
	}
	return 3;
}
int oct_network_send_msg_lua(lua_State* L) {
	char tmp_addr[NI_MAXHOST];
	char tmp_port[NI_MAXSERV];
	char tmp_buffer[BUFFER_SIZE];
	strncpy(tmp_buffer, luaL_checkstring(L, -3), BUFFER_SIZE);
	strncpy(tmp_addr, luaL_checkstring(L, -2), NI_MAXHOST);
	strncpy(tmp_port, luaL_checkstring(L, -1), NI_MAXSERV);

	oct_network_q_enqueue(&oct_network_node.send_queue, tmp_addr, tmp_port, tmp_buffer);
	return 0;
}

void oct_network_ab_init() {
	oct_network_ab.size = 0;
	oct_network_ab.first = NULL;
	oct_network_ab.last = NULL;
}

struct oct_network_ab_entry* oct_network_ab_insert(char* addr, char* port) {
	// First create the entry
	struct oct_network_ab_entry* e = (struct oct_network_ab_entry*)malloc(sizeof(struct oct_network_ab_entry));
	if (!e) {
		OCT_LOG_ERROR("Could not allocate space for address book entry for %s:%s", addr, port);
	}
	
	// Then, create the socket
	e->sfd = oct_network_init_socket(addr, port);
	if (e->sfd < 0) {
		free(e);
		return NULL;
	}
	OCT_LOG_INFO("Created new socket for %s:%s", addr, port);
	e->next = NULL;
	e->prev = oct_network_ab.last;
	strncpy(e->addr, addr, NI_MAXHOST);
	strncpy(e->port, port, NI_MAXSERV);
	if (oct_network_ab.size == 0) {
		oct_network_ab.first = e;
	}
	else {
		oct_network_ab.last->next = e;
	}
	oct_network_ab.last = e;
	oct_network_ab.size++;
	
	return e;
}

int oct_network_ab_remove(struct oct_network_ab_entry* e) {
	// First, remove node from linked list and fix
	if (e == oct_network_ab.first) {
		oct_network_ab.first = e->next;
		if (e->next) {
			e->next->prev = NULL;
		}
	}
	else if (e == oct_network_ab.last) {
		oct_network_ab.last = e->prev;
		e->prev->next = NULL;
	}
	else {
		e->next->prev = e->prev;
		e->prev->next = e->next;
	}
	// Finally, get rid of node
	close(e->sfd);
	free(e);
	oct_network_ab.size--;

	return 1;
}

struct oct_network_ab_entry* oct_network_ab_find(char* addr, char* port) {
	struct oct_network_ab_entry* crawler = oct_network_ab.first;
	while (crawler != NULL) {
		if (strncmp(crawler->addr, addr, NI_MAXHOST) == 0 &&
			strncmp(crawler->port, port, NI_MAXSERV) == 0) {
			return crawler;
		}
		crawler = crawler->next;
	}	
	return NULL;
}

void oct_network_ab_deinit() {
	while (oct_network_ab.first) {
		oct_network_ab_remove(oct_network_ab.first);
	}
}