#include #include #include #include #include #include #include #include #include #include #include #include /* arbitrary */ #define MAX_CMD_LEN 1024 enum que_op { LARK_KILL, LARK_NEW, LARK_POLL, }; #define VEC_NAME strbuf #define VEC_TYPE char #include /* 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 #define VEC_NAME conn_polls #define VEC_TYPE struct pollfd #include 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; }