summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/client.c18
-rw-r--r--src/server.c394
2 files changed, 347 insertions, 65 deletions
diff --git a/src/client.c b/src/client.c
index 3d95287..5b87c37 100644
--- a/src/client.c
+++ b/src/client.c
@@ -52,12 +52,20 @@ int main()
err(EXIT_FAILURE, "failed writing to socket");
}
- ret = read(tcp_sock, &cb, 1);
- if (ret == -1) {
- err(EXIT_FAILURE, "failed reading from socket");
- }
+ if (cb == '\n') {
+ char buf[4096];
+ ret = read(tcp_sock, buf, 4096);
+ if (ret == -1) {
+ printf("Error\n");
+ break;
+ }
- fprintf(stdout, "%c", cb);
+ if (ret == 0) {
+ printf("Closed socket\n");
+ break;
+ }
+ printf("%s\n", buf);
+ }
}
close(tcp_sock);
diff --git a/src/server.c b/src/server.c
index 468c4d6..f643386 100644
--- a/src/server.c
+++ b/src/server.c
@@ -6,14 +6,56 @@
#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;
- /* some other stuff */
+
+ struct strbuf strbuf;
+
+ enum server_cmd cmd;
+ union {
+ struct meta_parse_cmd meta_parse;
+ struct sync_cmd sync;
+ } u;
};
#define VEC_NAME server_conns
@@ -31,7 +73,8 @@ struct server_ctx {
int timeout;
};
-/* buf is sized by count */
+/* 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) {
@@ -47,14 +90,14 @@ static ssize_t readx(int fd, void *buf, size_t count)
}
/* we've read as many bytes as are currently available */
- return r;
+ return r == 0 ? -1 : r;
}
/* should never be reached, but keep here to avoid warnings */
return 0;
}
-static ssize_t writex(int fd, void *buf, size_t count)
+static ssize_t writex(int fd, const void *buf, size_t count)
{
while (1) {
ssize_t r = write(fd, buf, count);
@@ -62,10 +105,6 @@ static ssize_t writex(int fd, void *buf, size_t count)
if (errno == EINTR)
/* write interrupted, try again */
continue;
-
- if (errno == EWOULDBLOCK)
- /* no more data to read for now */
- return 0;
}
return r;
@@ -75,6 +114,11 @@ static ssize_t writex(int fd, void *buf, size_t count)
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;
@@ -108,9 +152,12 @@ static int new_connection(struct server_ctx *ctx, int new_conn_fd)
struct server_conn conn = {
.fd = fd,
.addr = addr,
- .addrlen = addrlen
+ .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;
}
@@ -120,8 +167,9 @@ static int new_connection(struct server_ctx *ctx, int new_conn_fd)
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);
+ assert(server_conns_len(&ctx->conns) >= 1);
if (server_conns_len(&ctx->conns) == 1) {
server_conns_shrink(&ctx->conns, 0);
@@ -134,6 +182,257 @@ static int kill_connection(struct server_ctx *ctx, struct server_conn *conn)
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 = {
@@ -168,6 +467,9 @@ static int event_loop(int tcp_sock)
* 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) {
@@ -190,9 +492,6 @@ static int event_loop(int tcp_sock)
}
/* otherwise, there's some event we should react to */
- size_t l = conn_polls_len(&ctx.polls);
- assert(l >= 1);
-
/* is it a new connection? */
struct pollfd *new_conn = conn_polls_at(&ctx.polls, 0);
if (new_conn->revents != 0) {
@@ -202,61 +501,34 @@ static int event_loop(int tcp_sock)
new_conn->revents = 0;
}
- for (size_t poll_idx = 1; poll_idx < l; ++poll_idx) {
- size_t conn_idx = poll_idx - 1;
-
- struct pollfd *pollfd = conn_polls_at(&ctx.polls, poll_idx);
+ 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);
- if (pollfd->revents & (POLLHUP | POLLNVAL | POLLERR)) {
- kill_connection(&ctx, conn);
- continue;
+ 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();
}
- if (!(pollfd->revents & POLLIN))
- continue;
-
- /* if there's no error, only POLLIN should be set, as it's the one we
- * requested */
- assert((pollfd->revents & POLLIN) == POLLIN);
- while (1) {
- /* for testing, just echo everything */
- char buf[4096];
- ssize_t r = readx(conn->fd, buf, 4096);
- if (r == -1) {
- /* no more data to read for now */
- if (errno == EWOULDBLOCK)
- break;
-
- /* something funky with the socket, kill the
- * connection and let the client try again later
- * if it wishes to */
- kill_connection(&ctx, conn);
- break;
- }
-
- ssize_t w = writex(conn->fd, buf, r);
- if (w == -1) {
- /* likewise, socket is funked up, kill
- * connection */
- kill_connection(&ctx, conn);
- break;
- }
-
- /* if readx or writex are broken, this shouldn't
- * hold */
- assert(w == r);
+ switch (op) {
+ case LARK_KILL:
+ kill_connection(&ctx, conn);
+ /* kill_connection placed last conn at this idx,
+ * retry it */
+ conn_idx--;
+ break;
- /* if we filled the buffer, check if there's
- * more to read */
- if (r == 4096)
- continue;
+ case LARK_NEW:
+ /* reset union for next command */
+ memset(&conn->u, 0, sizeof(conn->u));
+ break;
- /* otherwise, break out of echo */
+ case LARK_POLL:
+ /* nothing to do, timeout request should've been
+ * set if it matters */
break;
}
-
- pollfd->revents = 0;
}
}
@@ -269,6 +541,8 @@ 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");
generated by cgit v1.3 (git 2.53.0) at 2026-08-27 00:19:27 +0000