path: root/src/daemon/net.c

/*
 * Usurpation – newtwork logic
 *
 * Copyright (C) 2019 Gediminas Jakutis
 * Copyright (C) 2019 Paulius Ratkevičius
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; version 2.1
 * of the License.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
 */

#include <rin/time.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/udp.h>
#include <arpa/inet.h>
#include <poll.h>
#include <unistd.h>
#include <pthread.h>
#include <time.h>
#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <limits.h>
#include "settings.h"
#include "net.h"
#include "utils.h"

struct netstate {
	struct timespec lastreply;
	struct timespec start_time;
	int nd;
	int sock;
	unsigned short int port;
	pthread_t listner;
	char *data;
	size_t bufsize;
	ssize_t recvsize;
	unsigned int available;
	int sock_status;
	int buffer_status;
	pthread_mutex_t mutex;
	struct sockaddr_in clientaddr;
};

static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
static struct netstate *state;
static ssize_t count;
static size_t available_count;

static void *dolisten(void * state);
static int getpacket(char *data, size_t buffsize, ssize_t *recvbufsize, int sock, struct sockaddr_in *sender);

int net_init(const unsigned short int port)
{
	struct sockaddr_in bindaddress = {0};
	int ret;
	ssize_t i;
	int sizechanged = 0;

	pthread_mutex_lock(&mutex);
	if (available_count) {
		for (i = 0; !state[i].available; ++i) {
			if (i == count) {
				ret = ERROR;
				goto out;
			}
		}
	} else {
		if (++count == INT_MAX) {
			ret = ERROR;
			goto out;
		}

		i = count - 1;
		state = realloc(state, sizeof(struct netstate) * count);
		sizechanged = 1;
	}
	state[i].available = 0;
	bindaddress.sin_family = AF_INET;
	bindaddress.sin_port = htons(port);
	state[i].sock  = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
	if (bind(state[i].sock, (struct sockaddr*) &bindaddress, sizeof(bindaddress))) {
		ret = ERROR;
		goto out;
	}

	state[i].lastreply.tv_sec -= 5;
	state[i].nd = i + 1;
	state[i].port = port;
	state[i].data = malloc(MTU);
	memset(state[i].data, 0, MTU);
	state[i].bufsize = MTU;
	state[i].sock_status = NONEWDATA;
	state[i].buffer_status = NONEWDATA;
	clock_gettime(CLOCK_MONOTONIC_RAW, &state[i].lastreply);
	state[i].start_time = state[i].lastreply;
	pthread_mutex_init(&state[i].mutex, NULL);

	if (pthread_create(&state[i].listner, NULL, dolisten, state + i)) {
		ret = ERROR;
		goto out;
	}

	ret = count;
out:
	if (ret == ERROR && sizechanged) {
		state = realloc(state, sizeof(struct netstate) * --count);
	}

	pthread_mutex_unlock(&mutex);
	return ret;
}

int net_close(int nd)
{
	int ret;

	pthread_mutex_lock(&mutex);
	if (nd > count || nd < 1 || state[nd - 1].available) {
		ret = ERROR;
	} else {
		--nd;
		pthread_cancel(state[nd].listner);
		pthread_join(state[nd].listner, NULL);
		close(state[nd].sock);
		free(state[nd].data);
		state[nd].available = 1;
		ret = A_OK;
	}

	pthread_mutex_unlock(&mutex);
	return ret;
}

int net_getlastdata(int nd, char ** const data, size_t *recvsize)
{
	size_t size;
	int ret;

	if (nd > count || nd < 1 || state[--nd].available) {
		ret = ERROR;
	} else if (!(ret = state[nd].buffer_status)) {
		if (!data) {
			ret = ERROR;
			goto out;
		} else if (!(*data)) {
			*data = malloc(MTU);
		}

		size = state[nd].recvsize;
		memset(*data, 0, MTU);
		memcpy(*data, state[nd].data, size);
		*recvsize = size;
	}

	state[nd].buffer_status = NONEWDATA;
out:
	return ret;
}

int net_send(int nd, const char * const buf, size_t buf_size)
{
	int ret = A_OK;

	if (nd > count || nd < 1 || state[nd - 1].available) {
		ret = ERROR;
	} else {
		ret = net_send_addr(nd, buf, buf_size, &state[nd - 1].clientaddr);
	}

	return ret;
}

int net_send_addr(int nd, const char * const buf, size_t buf_size, const struct sockaddr_in * const addr)
{
	int ret = A_OK;
	ssize_t sent;
	struct timespec t;

	if (nd > count || nd < 1 || state[--nd].available) {
		ret = ERROR;
	} else {
		if (setting_verbose() >= INFO) {
			clock_gettime(CLOCK_MONOTONIC_RAW, &t);
			t = rin_time_sub(&t, &state[nd].start_time);
			fprintf(stderr, "Sending packet to %s, timestap: %li.%06lis \n", inet_ntoa(state[nd].clientaddr.sin_addr), t.tv_sec, t.tv_nsec / 1000);
		}

		sent = sendto(	state[nd].sock, buf, buf_size, MSG_DONTWAIT,
				(struct sockaddr *) addr,
				sizeof(*addr));

		if (sent == -1) {
			ret = DEAD;
			if (setting_verbose() >= ERR) {
				fprintf(stderr, "Sending packet to %s failed.\n", inet_ntoa(addr->sin_addr));
			}
		}
	}

	return ret;
}

static void *dolisten(void * state)
{
	struct timespec now;
	struct timespec delta;
	struct timespec wait = {0, 1 * 1000 * 1000};	/* 1 ms */
	struct timespec timeout = {60, 0};		/* 1 minute */
	struct netstate *st;
	int cancelstate;
	int oldstatus = DEAD;
	char *ipstring;

	st = state;

	do {
		pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, &cancelstate);
		pthread_mutex_lock(&st->mutex);
		st->sock_status = getpacket(st->data, st->bufsize, &st->recvsize, st->sock, &st->clientaddr);

		clock_gettime(CLOCK_MONOTONIC_RAW, &now);
		if (!st->sock_status) {
			st->lastreply = now;
			st->buffer_status = A_OK;

			if (setting_verbose() >= INFO) {
				ipstring = inet_ntoa(st->clientaddr.sin_addr);
				delta = rin_time_sub(&now, &st->start_time);
				fprintf(stderr, "packet from %s received at %li.%06li\n", ipstring, delta.tv_sec, delta.tv_nsec / 1000);
			}
		}

		/* timeout if no packets in over minute */
		delta = rin_time_sub(&now, &st->lastreply);
		if (rin_time_cmp_more(&delta, &timeout)) {
			st->sock_status = DEAD;
		};

		if (oldstatus != st->sock_status && st->sock_status != NONEWDATA) {
			oldstatus = st->sock_status;
			if (st->sock_status == DEAD && setting_verbose() >= INFO) {
				delta = rin_time_sub(&now, &st->start_time);
				fprintf(stderr, "Connection with the client has been lost. Last reply was: %li.%06lis ago\n", delta.tv_sec, delta.tv_nsec / 1000);
			}
		}

		pthread_mutex_unlock(&st->mutex);
		pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &cancelstate);
		pthread_testcancel();

		nanosleep(&wait, NULL);

	} while (1);

	return NULL;
}

static int getpacket(char *data, size_t buffsize, ssize_t *recvbufsize, int sock, struct sockaddr_in *sender)
{
	static struct pollfd pfd = {0};
	int ret;
	socklen_t sender_len = sizeof(*sender);

	if (!pfd.fd) {
		pfd.fd = sock;
		pfd.events = POLLIN;
		pfd.revents = 0;
	}

	ret = poll(&pfd, 1, 0);
	if (ret < 0) {
		/* NOOP */
	} else if (!ret) {
		ret = NONEWDATA;
	} else if (pfd.revents & POLLIN) {
		*recvbufsize = recvfrom(sock, data, buffsize, MSG_DONTWAIT, (struct sockaddr *) sender, &sender_len);
		if (*recvbufsize < 0) {
			ret = ERR;
		} else {
			ret = A_OK;
		}
	} else {
		ret = NONEWDATA;
	}
	return ret;
}