path: root/src/server.c

#include <err.h>
#include <poll.h>
#include <stdio.h>
#include <errno.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <syslog.h>
#include <stdbool.h>
#include <sys/socket.h>
#include <netinet/in.h>

/* arbitrary */
#define MAX_CMD_LEN 1024

enum que_op {
	LARK_KILL,
	LARK_NEW,
	LARK_POLL,
};

#define VEC_NAME strbuf
#define VEC_TYPE char
#include <conts/vec.h>

/* doesn't use any state */
struct meta_parse_cmd {};

enum sync_state {
	SYNC_INIT = 0,
	SYNC_SEND
};

struct sync_cmd {
	enum sync_state state;
	long start;
	long end;
	char *thread;
};

enum server_cmd {
	META_PARSE_CMD,
	SYNC_CMD,
};

struct server_conn {
	int fd;
	struct sockaddr addr;
	socklen_t addrlen;

	struct strbuf strbuf;

	enum server_cmd cmd;
	union {
		struct meta_parse_cmd meta_parse;
		struct sync_cmd sync;
	} u;
};

#define VEC_NAME server_conns
#define VEC_TYPE struct server_conn
#include <conts/vec.h>

#define VEC_NAME conn_polls
#define VEC_TYPE struct pollfd
#include <conts/vec.h>

struct server_ctx {
	int tcp_sock;
	struct server_conns conns;
	struct conn_polls polls;
	int timeout;
};

/* buf is sized by count. < 0 for error, 0 for no data available, > 0 for how
 * much data was read */
static ssize_t readx(int fd, void *buf, size_t count)
{
	while (1) {
		ssize_t r = read(fd, buf, count);
		if (r == -1) {
			if (errno == EINTR)
				/* read interrupted, try again */
				continue;

			if (errno == EWOULDBLOCK)
				/* no more data to read for now */
				return 0;
		}

		/* we've read as many bytes as are currently available */
		return r == 0 ? -1 : r;
	}

	/* should never be reached, but keep here to avoid warnings */
	return 0;
}

static ssize_t writex(int fd, const void *buf, size_t count)
{
	while (1) {
		ssize_t r = write(fd, buf, count);
		if (r == -1) {
			if (errno == EINTR)
				/* write interrupted, try again */
				continue;
		}

		return r;
	}

	/* should never be reached, but keep here to avoid warnings */
	return 0;
}

static bool strneq(const char *s1, const char *s2, size_t n)
{
	return strncmp(s1, s2, n) == 0;
}

static int new_connection(struct server_ctx *ctx, int new_conn_fd)
{
	struct sockaddr addr;
	socklen_t addrlen;

	int ret = 0;

	/** @todo retry count? */
	while (1) {
		ret = accept(new_conn_fd, &addr, &addrlen);
		if (ret == -1) {
			/* try again */
			if (errno == EINTR)
				continue;

			/* some other error, throw it at the user */
			return ret;
		}

		/* we have a valid fd */
		break;
	}

	int fd = ret;
	ret = fcntl(fd, F_SETFL, O_NONBLOCK);
	if (ret == -1) {
		close(fd);
		return ret;
	}

	struct server_conn conn = {
		.fd      = fd,
		.addr    = addr,
		.addrlen = addrlen,
		.cmd     = META_PARSE_CMD,
		.strbuf  = strbuf_create(0)
	};

	memset(&conn.u, 0, sizeof(conn.u));
	server_conns_append(&ctx->conns, conn);
	return 0;
}

/* I'll assume that each individual subcommant has ensured all its resources are
 * freed by now. Invalidates the conn pointer! */
static int kill_connection(struct server_ctx *ctx, struct server_conn *conn)
{
	close(conn->fd);
	strbuf_destroy(&conn->strbuf);

	assert(server_conns_len(&ctx->conns) >= 1);

	if (server_conns_len(&ctx->conns) == 1) {
		server_conns_shrink(&ctx->conns, 0);
		return 0;
	}

	/* overwrite our location in the vector with whaterver is at the back of
	 * the vector */
	*conn = *server_conns_pop(&ctx->conns);
	return 0;
}

/* re-entrant read, returns < 0 on error, 0 on not enough data and > 0 showing the
 * length of the cmd (including newline) */
static ssize_t reread_cmd(struct server_conn *conn, size_t max)
{
	struct strbuf *strbuf = &conn->strbuf;

	/* by default, read up to a page of data at a time */
	size_t n = strbuf->n;

	/* grow to hold up to 4096 new chars */
	strbuf_reserve(strbuf, n + 4096);

	/* read up to 4096 new chars */
	ssize_t r = readx(conn->fd, strbuf->buf + n, 4096);
	if (r == 0) {
		/* shrink back to original size and try again later */
		strbuf_shrink(strbuf, n);
		return 0;
	}

	/* some unforeseen error occured */
	if (r < 0)
		return r;

	/* set to actual size */
	strbuf_shrink(strbuf, n + r);

	/* look for newline */
	for (ssize_t i = 0; i < r; ++i) {
		/* read bytes go above allowed limit */
		if (n + i > max)
			return -1;

		if (strbuf->buf[n + i] == '\n')
			return r;
	}

	/* no command as of yet, but try again later */
	return 0;
}

/* set timeout to new smallest value, taking infinite timeout (-1) into account.
 * */
static void set_timeout(struct server_ctx *ctx, int timeout)
{
	if (ctx->timeout == -1) {
		ctx->timeout = timeout;
		return;
	}

	/* ctx->timeout != -1, meaning there's a timeout set and it's not
	 * infinite, so prefer the existing value */
	if (timeout == -1)
		return;

	if (timeout < ctx->timeout)
		ctx->timeout = timeout;
}

static enum que_op meta_parse_cmd(struct server_ctx *ctx, struct server_conn *conn)
{
	int r = reread_cmd(conn, MAX_CMD_LEN);
	if (r < 0) {
		syslog(LOG_DEBUG,
				"error while reading cmd, "
				"client disconnected or sent garbage data\n");
		return LARK_KILL;
	}

	if (r == 0) {
		syslog(LOG_DEBUG, "not enough data in socket to parse cmd with\n");
		return LARK_POLL;
	}

	/* r > 0, we have a command and its length */
	if (strneq(conn->strbuf.buf, "SYNC", 4)) {
		syslog(LOG_DEBUG, "parsed SYNC command start\n");
		conn->cmd = SYNC_CMD;

		/* we want to immediately run the next command since we know
		 * that we've placed it into the strbuf */
		set_timeout(ctx, 0);
		return LARK_NEW;
	}

	syslog(LOG_DEBUG, "client is rarted\n");
	return LARK_KILL;
}

static void strbuf_shift(struct strbuf *strbuf, size_t i)
{
	assert(i <= strbuf->n);
	memmove(strbuf->buf, strbuf->buf + i, strbuf->n - i);
	strbuf_shrink(strbuf, strbuf->n - i);
}

static int sync_parse(struct sync_cmd *cmd, struct strbuf *strbuf)
{
	char *buf = strbuf->buf;
	size_t max = strbuf->n;

	/* default values */
	cmd->start  =  0;
	cmd->end    = -1;
	cmd->thread = NULL;

	size_t i = 0;

	size_t start = 0, split = 0, end = 0;
	for (i = 0; i < max; ++i) {
		if (buf[i] == '=') {
			split = i;
			continue;
		}

		if (buf[i] == '\\') {
			/* skip escaped character */
			++i;
			continue;
		}

		/* regular characters, don't do anything special with them */
		if (buf[i] != ' ' && buf[i] != '\n')
			continue;

		buf[i] = '\0';

		/* we have something like STRING or STRING=STRING, possibly with
		 * escaped characters */
		end = i;

		/* special case, consequtive spaces */
		if (end == start) {
			start = i + 1;
			split = start;
			continue;
		}

		if (split == start) {
			/* we're just a single key, like STRING */
			const char *key = &buf[start];
			size_t keylen = end - start;

			if (strneq(key, "SYNC", keylen)) {
				/* nothing to do, which technically speaking
				 * allows the command name to be repeated arbitratily often */
			}
			else {
				syslog(LOG_DEBUG, "illegal flag name: %s\n", key);
				return -1;
			}

		} else {
			/* we're something like STRING=STRING */
			const char *key = &buf[start];
			size_t keylen = split - start;

			const char *value = &buf[split + 1];
			size_t valuelen = end - (split + 1);

			/* this seems fairly easy to automate, just pass in an
			 * array of structs that describe the name and type of
			 * each kv pair, the user can then make more checks
			 * after parsing */
			if (strneq(key, "START", keylen)) {
				char *endptr = NULL;
				cmd->start = strtol(value, &endptr, 10);
				if (value + valuelen != endptr) {
					syslog(LOG_DEBUG, "not an integer: %s\n", key);
					return -1;
				}

			} else if (strneq(key, "END", keylen)) {
				char *endptr = NULL;
				cmd->start = strtol(value, &endptr, 10);
				if (value + valuelen != endptr) {
					syslog(LOG_DEBUG, "not an integer: %s\n", key);
					return -1;
				}

			} else if (strneq(key, "THREAD", keylen)) {
				cmd->thread = strndup(value, valuelen);

				/** @todo check that the string is a valid
				 * thread reference, name@server */

			} else {
				syslog(LOG_DEBUG, "illegal keyvalue: %s\n", key);
				return -1;
			}
		}


		/* next iter */
		start = i + 1;
		split = start;
	}

	strbuf_shift(strbuf, i);
	return 0;
}

static void destroy_sync(struct sync_cmd *cmd)
{
	free(cmd->thread);
}

static enum que_op sync_cmd(struct server_ctx *ctx, struct server_conn *conn)
{
	struct sync_cmd *cmd = &conn->u.sync;
	switch (cmd->state) {
	case SYNC_INIT: {
		if (sync_parse(cmd, &conn->strbuf))
			return LARK_KILL;

		if (cmd->start < 0) {
			syslog(LOG_DEBUG, "illegal value: SYNC START=%ld\n", cmd->start);
			destroy_sync(cmd);
			return LARK_KILL;
		}

		if (cmd->end < 0 && cmd->end != -1) {
			syslog(LOG_DEBUG, "illegal value: SYNC END=%ld\n", cmd->end);
			destroy_sync(cmd);
			return LARK_KILL;
		}

		if (cmd->thread == NULL) {
			/** @todo copy over our home thread if it exists */
		}

		cmd->state = SYNC_SEND;
		return sync_cmd(ctx, conn);
	}

	case SYNC_SEND: {
		syslog(LOG_DEBUG, "sending sync\n");

		const char buf[] = "From: urmum@amazon.com";
		ssize_t ret = writex(conn->fd, buf, sizeof(buf));
		assert(ret == sizeof(buf));
		conn->cmd = META_PARSE_CMD;
		destroy_sync(cmd);
		return LARK_NEW;
	}
	}

	syslog(LOG_DEBUG, "sync state broken\n");
	return LARK_KILL;
}

static int event_loop(int tcp_sock)
{
	struct server_ctx ctx = {
		.tcp_sock  = tcp_sock,
		.conns     = server_conns_create(0),
		.polls     = conn_polls_create(0),
		.timeout   = -1
	};

	int ret = 0;
	while (1) {
		/* build up set of connections to poll */
		conn_polls_reset(&ctx.polls);

		struct pollfd tcp_poll = {
			.fd      = ctx.tcp_sock,
			.events  = POLLIN,
			.revents = 0
		};
		conn_polls_append(&ctx.polls, tcp_poll);

		foreach(server_conns, c, &ctx.conns) {
			struct pollfd conn_poll = {
				.fd      = c->fd,
				.events  = POLLIN,
				.revents = 0
			};
			conn_polls_append(&ctx.polls, conn_poll);
		}

		/* wait for something to happen on incoming connections, or a
		 * new connection. Eventually might also want to wait on
		 * outgoing connections to become available for writing? */
		ret = poll(ctx.polls.buf, conn_polls_len(&ctx.polls), ctx.timeout);
		/* reset timeout value for next iteration */
		ctx.timeout = -1;

		if (ret == 0) {
			/* timeout */
			if (server_conns_len(&ctx.conns) == 0) {
				/* nothing to do, quit for now */
				break;
			}

			/* some command just timed out, handle it by fallthrough */
		}

		if (ret == -1) {
			/* some kind of error */
			if (errno == EINTR) {
				/* safe to try again */
				continue;
			}

			/* exit loop due to unknown error*/
			err(EXIT_FAILURE, "poll failed");
		}

		/* otherwise, there's some event we should react to */
		/* is it a new connection? */
		struct pollfd *new_conn = conn_polls_at(&ctx.polls, 0);
		if (new_conn->revents != 0) {
			if (new_connection(&ctx, new_conn->fd))
				err(EXIT_FAILURE, "tcp socket exploded");

			new_conn->revents = 0;
		}

		for (size_t conn_idx = 0; conn_idx < server_conns_len(&ctx.conns); ++conn_idx) {
			struct server_conn *conn = server_conns_at(&ctx.conns, conn_idx);

			enum que_op op = LARK_KILL;
			switch (conn->cmd) {
			case META_PARSE_CMD: op = meta_parse_cmd(&ctx, conn); break;
			case SYNC_CMD: op = sync_cmd(&ctx, conn); break;
			default: abort();
			}

			switch (op) {
			case LARK_KILL:
				kill_connection(&ctx, conn);
				/* kill_connection placed last conn at this idx,
				 * retry it */
				conn_idx--;
				break;

			case LARK_NEW:
				/* reset union for next command */
				memset(&conn->u, 0, sizeof(conn->u));
				break;

			case LARK_POLL:
				/* nothing to do, timeout request should've been
				 * set if it matters */
				break;
			}
		}
	}

	conn_polls_destroy(&ctx.polls);
	server_conns_destroy(&ctx.conns);
	return ret;
}

int main(int argc, char *argv[argc])
{
	int ret = 0;

	openlog(NULL, LOG_PERROR, LOG_USER);

	int tcp_sock = socket(AF_INET, SOCK_STREAM, 0);
	if (tcp_sock == -1) {
		err(EXIT_FAILURE, "failed opening tcp socket");
	}

	struct sockaddr_in addr;
	memset(&addr, 0, sizeof(addr));
	addr.sin_family      = AF_INET;
	addr.sin_addr.s_addr = htonl(INADDR_ANY);
	/* should probably come up with a reasonable port at some point */
	addr.sin_port        = htons(24242);

	if (bind(tcp_sock, (struct sockaddr *)&addr, sizeof(addr)) == -1) {
		err(EXIT_FAILURE, "failed binding tcp socket");
	}

	if (listen(tcp_sock, 64) == -1) {
		err(EXIT_FAILURE, "failed listening on tcp socket");
	}

	ret = event_loop(tcp_sock);
	close(tcp_sock);
	return ret;
}