path: root/src/server.c

#include <err.h>
#include <poll.h>
#include <stdio.h>
#include <errno.h>
#include <fcntl.h>
#include <ctype.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <syslog.h>
#include <stddef.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_NEXT,
	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_HEADER = 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 fdin;
	int fdout;
	struct sockaddr addr;
	socklen_t addrlen;

	struct strbuf strbuf;
	size_t nextlen;

	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 sock;
	bool oneshot;
	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 bool streq(const char *s1, const char *s2)
{
	return strcmp(s1, s2) == 0;
}

static int new_connection(struct server_ctx *ctx, int new_conn_fd)
{
	assert(!ctx->oneshot);

	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 = {
		.fdin      = fd,
		.fdout     = fd,
		.addr      = addr,
		.addrlen   = addrlen,
		.cmd       = META_PARSE_CMD,
		.strbuf    = strbuf_create(0),
		.nextlen   = 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->fdin);
	if (conn->fdout != conn->fdin)
		close(conn->fdout);

	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 for how
 * large the buffer is currently */
static ssize_t reread(struct server_conn *conn, size_t c)
{
	assert(c != 0);

	struct strbuf *strbuf = &conn->strbuf;

	size_t n = strbuf->n;
	strbuf_reserve(strbuf, n + c);

	ssize_t r = readx(conn->fdin, strbuf->buf + n, c);
	if (r < 0)
		return r;

	strbuf_shrink(strbuf, n + r);
	return n + r;
}

/* read a full command, using at most max bytes.
 * < 0 on error (including no command found in the allotted byte count),
 *   0 if no command found yet,
 * > 0 when command is available, indicating command length (including newline)
 *
 * Note that this does not respect timeouts, and each command that uses this
 * should keep track of its own timing.
 */
static ssize_t reread_cmd(struct server_conn *conn, size_t max)
{
	struct strbuf *strbuf = &conn->strbuf;

	/* previous occupancy */
	size_t n = strbuf->n;
	ssize_t r = reread(conn, max);
	if (r < 0)
		return r;

	if (r == (ssize_t)n)
		return 0;

	/* parse the new bytes and see if we have an end-of-command marker */
	char *buf = strbuf->buf;
	for (ssize_t i = n; i < r; ++i) {
		if (i > (ssize_t)max)
			return -1;

		if (buf[i] == '\n')
			return i + 1;
	}

	return 0;
}

/* read up to n bytes.
 * < 0 on error
 *   0 if not enough bytes yet
 * > 0 to indicate how many bytes are available (should always be n)
 */
static ssize_t reread_n(struct server_conn *conn, size_t n)
{
	if (conn->strbuf.n >= n)
		return n;

	ssize_t r = reread(conn, n - conn->strbuf.n);
	if (r < 0)
		return r;

	if (r < (ssize_t)n)
		return 0;

	return r;
}

/* 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;
}

/* return 0 on success, !0 on failure */
static int parse_cmd(char *buf, char **cmd, size_t *len)
{
	char *keystart = buf;
	while (*keystart && isblank(*keystart))
		keystart++;

	/* empty string */
	if (!*keystart)
		return -1;

	char *keyend = keystart + 1;
	while (*keyend && isalnum(*keyend))
		keyend++;

	/* nothing after key */
	if (!*keyend)
		return -1;

	/* split off cmd from rest */
	*keyend = '\0';
	if (cmd)
		*cmd = keystart;

	if (!len)
		return 0;

	char *lenstart = keyend + 1;
	while (*lenstart && isblank(*lenstart))
		lenstart++;

	/* space after key but no len value */
	if (!*lenstart)
		return -1;

	char *lenend = lenstart + 1;
	while (*lenend && isdigit(*lenend))
		lenend++;

	if (!*lenend) {
		/* end of string, nothing after digits but there doesn't need to be */
		char *check = NULL;
		*len = strtoull(lenstart, &check, 10);

		/* sanity check of sorts, if they're equal we return 0 (which we
		 * want), otherwise -1 */
		return check == lenend ? 0 : -1;
	}

	/* something following length, trailing spaces are allowed but nothing
	 * else */
	char *linend = lenend + 1;
	while (*linend && isblank(*linend))
		linend++;

	/* there was other stuff besides spaces, error */
	if (*linend)
		return -1;

	*lenend = '\0';

	char *check = NULL;
	*len = strtoull(lenstart, &check, 10);
	return check == lenend ? 0 : -1;
}

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 enum que_op meta_parse_cmd(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 (should) have a command and its nextlen */
	char *cmd = NULL;
	conn->strbuf.buf[r - 1] = '\0';
	if (parse_cmd(conn->strbuf.buf, &cmd, &conn->nextlen)) {
		syslog(LOG_DEBUG, "invalid command %s, killing connection\n", conn->strbuf.buf);
		return LARK_KILL;
	}

	if (strneq(conn->strbuf.buf, "SYNC", 4)) {
		syslog(LOG_DEBUG, "parsed SYNC command start\n");
		conn->cmd = SYNC_CMD;
		strbuf_shift(&conn->strbuf, r);
		return LARK_NEW;
	}

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

/* could also take the length of the buf? */
typedef int (*kv_cb_t)(char *key, char *value, void *data);
static int parse_kv(char *buf, int (*kv_cb)(char *key, char *value, void *data), void *data)
{
	if (!buf)
		return 0;

	char *start = buf;
	while (1) {
		char *keystart = start;
		/* skip leading whitespace */
		while (*keystart && isblank(*keystart))
			keystart++;

		/* end of string, nothing more to read */
		if (!*keystart)
			return 0;

		/* end of blank line, skip to next */
		if (*keystart == '\n') {
			start++;
			continue;
		}

		/* start of empty line just a comment, skip to next line if one exists */
		if (*keystart == '#') {
			while (*keystart && *keystart != '\n')
				keystart++;

			/* end of string */
			if (!*keystart)
				return 0;

			/* skip line-ending '\n' */
			start = keystart + 1;
			continue;
		}

		/* start now points at the first character of a key. Next, parse
		 * key forward a bit to check that all characters are valid */
		char *keyend = keystart;
		while (*keyend && (isalnum(*keyend) || *keyend == '-' || *keyend == '_'))
			keyend++;

		/* key ends a line, illegal */
		if (!*keyend)
			return -1;

		/* key contains illegal characters, as we didn't end up on empty
		 * space or the '=' key */
		if (!(isblank(*keyend) || *keyend == '='))
			return -1;

		/* next, find out where value starts */
		char *valstart = keyend;

		/* skip forward until '=' sign */
		while (*valstart && *valstart != '=')
			valstart++;

		/* plain key on a line, illegal */
		if (!*valstart)
			return -1;

		/* skip over '=' sign */
		valstart++;

		/* skip forward until actual start of value, when blanks stop */
		while (*valstart && isblank(*valstart))
			valstart++;

		/* find end of value. Note that empty values are allowed,
		 * whether they're empty by end of string or end of line or the
		 * start of a comment*/
		char *valend = valstart;
		char *runner = valend;
		while (*runner && *runner != '#' && *runner != '\n') {
			/* keep track of where last non-whitespace character is
			 * since it tells us where the value ends */
			if (!isblank(*runner))
				valend = runner + 1;

			/* skip escape characters */
			if (*runner == '\\')
				runner++;

			/* but don't skip end of line of end of string */
			if (!*runner || *runner == '\n')
				break;

			runner++;
		}

		/* prepare start of next line */
		while (*runner && *runner != '\n')
			runner++;

		if (*runner == '\n')
			runner++;

		start = runner;

		/* present parsed values to user */
		*keyend = '\0';
		*valend = '\0';

		/* unescape backslashes */
		long shift = 0;
		for (ptrdiff_t i = 0; i < valend - valstart + 1; ++i) {
			if (valstart[i] == '\\') {
				shift++;
				continue;
			}

			if (shift == 0)
				continue;

			valstart[i - shift] = valstart[i];
		}

		int r = kv_cb(keystart, valstart, data);
		if (r)
			return r;
	}

	return 0;
}

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

static int sync_kv(char *key, char *value, struct sync_cmd *cmd)
{
	if (streq(key, "START")) {
		char *end = NULL;
		cmd->start = strtol(value, &end, 0);
		if (end == value || *end != '\0') {
			syslog(LOG_DEBUG, "not an integer value: %s = %s", key, value);
			return -1;
		}

		if (cmd->start < 0) {
			syslog(LOG_DEBUG, "illegal integer value: %ld", cmd->start);
			return -1;
		}

		return 0;
	}

	if (streq(key, "END")) {
		char *end = NULL;
		cmd->end = strtol(value, &end, 0);
		if (end == value || *end != '\0') {
			syslog(LOG_DEBUG, "not an integer value: %s = %s", key, value);
			return -1;
		}

		if (cmd->end < 0 && cmd->end != -1) {
			syslog(LOG_DEBUG, "illegal integer value: %ld", cmd->end);
			return -1;
		}

		return 0;
	}

	if (streq(key, "THREAD")) {
		/** @todo check validness, copy default value, etc. */
		cmd->thread = strdup(value);
		return 0;
	}

	syslog(LOG_DEBUG, "unknown parameter to sync: %s", key);
	return -1;
}

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_HEADER: {
		ssize_t r = reread_n(conn, conn->nextlen);
		if (r < 0) {
			syslog(LOG_DEBUG, "client messed up connection");
			return LARK_KILL;
		}

		/** @todo respect timeout of some kind? */
		if (r == 0)
			return LARK_POLL;

		/* headr must include trailing newline, so we can be sure that a
		 * valid header will remain valid if we null-terminate the
		 * header */
		conn->strbuf.buf[r - 1] = '\0';

		/* parse ini-style header */

		/* some default values, arguably not sensible to always sync a
		 * whole thread but eh */
		cmd->start = 0;
		cmd->end   = -1;
		cmd->thread = NULL;

		if (parse_kv(conn->strbuf.buf, (kv_cb_t)sync_kv, cmd)) {
			syslog(LOG_DEBUG, "failed parsing SYNC header");
			destroy_sync(cmd);
			return LARK_KILL;
		}

		strbuf_shift(&conn->strbuf, r);

		syslog(LOG_DEBUG, "got valid sync header\n");
		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->fdout, 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 sock, bool oneshot)
{
	struct server_ctx ctx = {
		.sock      = sock,
		.oneshot   = oneshot,
		.conns     = server_conns_create(0),
		.polls     = conn_polls_create(0),
		.timeout   = -1
	};

	/* set up a connection */
	if (oneshot) {
		struct server_conn conn = {
			.fdin    = sock,
			.fdout   = sock == STDIN_FILENO ? STDOUT_FILENO : sock,
			.addr    = {},
			.addrlen = 0,
			.cmd     = META_PARSE_CMD,
			.strbuf  = strbuf_create(0),
			.nextlen = 0
		};

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

	int ret = 0;
	while (1) {
		if (ctx.oneshot && server_conns_len(&ctx.conns) == 0) {
			ret = 0;
			break;
		}

		/* build up set of connections to poll */
		conn_polls_reset(&ctx.polls);

		/* no need to poll if we're in oneshot mode, but keep pollfd
		 * around to simplify index calculations */
		struct pollfd tcp_poll = {
			.fd      = ctx.oneshot ? -1 : ctx.sock,
			.events  = POLLIN,
			.revents = 0
		};
		conn_polls_append(&ctx.polls, tcp_poll);

		foreach(server_conns, c, &ctx.conns) {
			struct pollfd conn_poll = {
				.fd      = c->fdin,
				.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(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_NEXT:
				set_timeout(&ctx, 0);
				break;

			case LARK_NEW:
				set_timeout(&ctx, 0);
				/* 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, opt;

	const char *single_conn = NULL;
	while ((opt = getopt(argc, argv, "s:")) != -1) {
		switch (opt) {
		case 's':
			single_conn = optarg;
			break;

		default:
			fprintf(stderr, "usage: %s [-s socket or - for stdin]\n",
					argv[0]);
			exit(EXIT_FAILURE);
		}
	}

	openlog(NULL, LOG_PERROR, LOG_USER);

	int sock = -1;
	if (single_conn) {
		if (streq(single_conn, "-"))
			sock = STDIN_FILENO;
		else
			sock = open(single_conn, O_RDONLY);

		if (sock == -1)
			err(EXIT_FAILURE, "failed opening single connection");

	} else {
		int sock = socket(AF_INET, SOCK_STREAM, 0);
		if (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(sock, (struct sockaddr *)&addr, sizeof(addr)) == -1)
			err(EXIT_FAILURE, "failed binding tcp socket");

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

	ret = event_loop(sock, single_conn != NULL);
	close(sock);
	return ret;
}