X-Git-Url: https://git.tokkee.org/?p=sysdb.git;a=blobdiff_plain;f=src%2Ffrontend%2Fconnection.c;h=ce16a8b900085ecbac3eb5ef82b4af466400dcca;hp=69c71edbd3e6f7ac9e5876841d9237f738267ab5;hb=f78d30d42f9fe3b58f1c2132cafe4c8687b8bd46;hpb=72a63cfb0e65bb575889b8dbee6648cafd6a52f2 diff --git a/src/frontend/connection.c b/src/frontend/connection.c index 69c71ed..ce16a8b 100644 --- a/src/frontend/connection.c +++ b/src/frontend/connection.c @@ -29,7 +29,6 @@ #include "core/object.h" #include "core/plugin.h" #include "frontend/connection-private.h" -#include "frontend/parser.h" #include "utils/error.h" #include "utils/strbuf.h" #include "utils/proto.h" @@ -117,6 +116,7 @@ connection_init(sdb_object_t *obj, va_list ap) conn->cmd = CONNECTION_IDLE; conn->cmd_len = 0; + conn->skip_len = 0; /* update the object name */ snprintf(obj->name + strlen(CONN_FD_PREFIX), @@ -140,12 +140,15 @@ connection_destroy(sdb_object_t *obj) "(%zu byte%s left in buffer)", len, len == 1 ? "" : "s"); } - sdb_log(SDB_LOG_DEBUG, "frontend: Closing connection on fd=%i", - conn->fd); + sdb_log(SDB_LOG_DEBUG, "frontend: Closing connection %s", obj->name); if (conn->fd >= 0) close(conn->fd); conn->fd = -1; + if (conn->username) + free(conn->username); + conn->username = NULL; + sdb_strbuf_destroy(conn->buf); conn->buf = NULL; sdb_strbuf_destroy(conn->errbuf); @@ -185,7 +188,13 @@ sdb_conn_ctx_init(void) static void sdb_conn_set_ctx(sdb_conn_t *conn) { + sdb_conn_t *old; + sdb_conn_ctx_init(); + + old = pthread_getspecific(conn_ctx_key); + if (old) + sdb_object_deref(SDB_OBJ(old)); if (conn) sdb_object_ref(SDB_OBJ(conn)); pthread_setspecific(conn_ctx_key, conn); @@ -250,112 +259,34 @@ command_handle(sdb_conn_t *conn) int status = -1; assert(conn && (conn->cmd != CONNECTION_IDLE)); + assert(! conn->skip_len); sdb_log(SDB_LOG_DEBUG, "frontend: Handling command %u (len: %u)", conn->cmd, conn->cmd_len); - if ((! conn->username) && (conn->cmd != CONNECTION_STARTUP)) { - const char *errmsg = "Authentication required"; - sdb_connection_send(conn, CONNECTION_ERROR, - (uint32_t)strlen(errmsg), errmsg); - return -1; - } - - /* reset */ - sdb_strbuf_sprintf(conn->errbuf, ""); - - switch (conn->cmd) { - case CONNECTION_PING: - status = sdb_connection_ping(conn); - break; - case CONNECTION_STARTUP: - status = sdb_fe_session_start(conn); - break; - - case CONNECTION_QUERY: - { - sdb_llist_t *parsetree; - sdb_conn_node_t *node = NULL; - - parsetree = sdb_fe_parse(sdb_strbuf_string(conn->buf), - (int)conn->cmd_len); - if (! parsetree) { - sdb_log(SDB_LOG_ERR, "frontend: Failed to parse query '%s'", - sdb_strbuf_string(conn->buf)); - status = -1; - break; - } - - switch (sdb_llist_len(parsetree)) { - case 0: - /* skipping empty command */ - break; - case 1: - node = SDB_CONN_NODE(sdb_llist_get(parsetree, 0)); - break; - - default: - sdb_log(SDB_LOG_WARNING, "frontend: Ignoring %d command%s " - "in multi-statement query '%s'", - sdb_llist_len(parsetree) - 1, - sdb_llist_len(parsetree) == 2 ? "" : "s", - sdb_strbuf_string(conn->buf)); - node = SDB_CONN_NODE(sdb_llist_get(parsetree, 0)); - } - - if (node) { - status = sdb_fe_exec(conn, node); - sdb_object_deref(SDB_OBJ(node)); - } - - sdb_llist_destroy(parsetree); - break; - } - - case CONNECTION_FETCH: - status = sdb_fe_fetch(conn, sdb_strbuf_string(conn->buf)); - break; - case CONNECTION_LIST: - status = sdb_fe_list(conn); - break; - case CONNECTION_LOOKUP: - { - sdb_store_matcher_t *m; - - m = sdb_fe_parse_matcher(sdb_strbuf_string(conn->buf), - (int)conn->cmd_len); - if (! m) { - sdb_log(SDB_LOG_ERR, "frontend: Failed to parse expression '%s'", - sdb_strbuf_string(conn->buf)); - status = -1; - break; - } - - status = sdb_fe_lookup(conn, m); - sdb_object_deref(SDB_OBJ(m)); - break; - } - - default: - { - sdb_log(SDB_LOG_WARNING, "frontend: Ignoring invalid command %#x", - conn->cmd); - sdb_strbuf_sprintf(conn->errbuf, "Invalid command %#x", conn->cmd); - status = -1; - break; - } + if (conn->cmd == CONNECTION_PING) + status = sdb_connection_ping(conn); + else if (conn->cmd == CONNECTION_STARTUP) + status = sdb_fe_session_start(conn); + else if (conn->cmd == CONNECTION_QUERY) + status = sdb_fe_query(conn); + else if (conn->cmd == CONNECTION_FETCH) + status = sdb_fe_fetch(conn); + else if (conn->cmd == CONNECTION_LIST) + status = sdb_fe_list(conn); + else if (conn->cmd == CONNECTION_LOOKUP) + status = sdb_fe_lookup(conn); + else { + sdb_log(SDB_LOG_WARNING, "frontend: Ignoring invalid command %#x", + conn->cmd); + sdb_strbuf_sprintf(conn->errbuf, "Invalid command %#x", conn->cmd); + status = -1; } if (status) sdb_connection_send(conn, CONNECTION_ERROR, (uint32_t)sdb_strbuf_len(conn->errbuf), sdb_strbuf_string(conn->errbuf)); - - /* remove the command from the buffer */ - if (conn->cmd_len) - sdb_strbuf_skip(conn->buf, 0, conn->cmd_len); - conn->cmd = CONNECTION_IDLE; - conn->cmd_len = 0; return status; } /* command_handle */ @@ -363,17 +294,42 @@ command_handle(sdb_conn_t *conn) static int command_init(sdb_conn_t *conn) { - size_t len; + const char *errmsg = NULL; assert(conn && (conn->cmd == CONNECTION_IDLE) && (! conn->cmd_len)); + if (conn->skip_len) + return -1; + + /* reset */ + sdb_strbuf_clear(conn->errbuf); + conn->cmd = connection_get_int32(conn, 0); conn->cmd_len = connection_get_int32(conn, sizeof(uint32_t)); - len = 2 * sizeof(uint32_t); - if (conn->cmd == CONNECTION_IDLE) - len += conn->cmd_len; - sdb_strbuf_skip(conn->buf, 0, len); + sdb_strbuf_skip(conn->buf, 0, 2 * sizeof(uint32_t)); + + if ((! conn->username) && (conn->cmd != CONNECTION_STARTUP)) + errmsg = "Authentication required"; + else if (conn->cmd == CONNECTION_IDLE) + errmsg = "Invalid command 0"; + + if (errmsg) { + size_t len = sdb_strbuf_len(conn->buf); + + sdb_strbuf_sprintf(conn->errbuf, "%s", errmsg); + sdb_connection_send(conn, CONNECTION_ERROR, + (uint32_t)strlen(errmsg), errmsg); + conn->skip_len += conn->cmd_len; + conn->cmd = CONNECTION_IDLE; + conn->cmd_len = 0; + + if (len > conn->skip_len) + len = conn->skip_len; + sdb_strbuf_skip(conn->buf, 0, len); + conn->skip_len -= len; + /* connection_read will handle anything else */ + } return 0; } /* command_init */ @@ -402,7 +358,19 @@ connection_read(sdb_conn_t *conn) else if (! status) /* EOF */ break; + if (conn->skip_len) { + size_t len = (size_t)status < conn->skip_len + ? (size_t)status : conn->skip_len; + sdb_strbuf_skip(conn->buf, 0, len); + conn->skip_len -= len; + } + n += status; + + /* give the main loop a chance to execute commands (and free up buffer + * space) on large amounts of incoming traffic */ + if (n > 1024 * 1024) + break; } return n; @@ -434,6 +402,14 @@ sdb_connection_accept(int fd) void sdb_connection_close(sdb_conn_t *conn) { + if (! conn) + return; + + /* close the connection even if someone else still references it */ + if (conn->fd >= 0) + close(conn->fd); + conn->fd = -1; + sdb_object_deref(SDB_OBJ(conn)); } /* sdb_connection_close */ @@ -451,9 +427,16 @@ sdb_connection_read(sdb_conn_t *conn) && (sdb_strbuf_len(conn->buf) >= 2 * sizeof(int32_t))) command_init(conn); if ((conn->cmd != CONNECTION_IDLE) - && (sdb_strbuf_len(conn->buf) >= conn->cmd_len)) + && (sdb_strbuf_len(conn->buf) >= conn->cmd_len)) { command_handle(conn); + /* remove the command from the buffer */ + if (conn->cmd_len) + sdb_strbuf_skip(conn->buf, 0, conn->cmd_len); + conn->cmd = CONNECTION_IDLE; + conn->cmd_len = 0; + } + if (status <= 0) break;