X-Git-Url: https://git.tokkee.org/?a=blobdiff_plain;f=src%2Ffrontend%2Fsock.c;h=11fa936686da75e169445842840cad502615d676;hb=f529f86b7b05704f3ede80f06221b556620a8b75;hp=d157853bb64bc4b7489af9beb8c12dff5ab0e853;hpb=60b7521d96faf3dc97118012e80667f1286c17c4;p=sysdb.git diff --git a/src/frontend/sock.c b/src/frontend/sock.c index d157853..11fa936 100644 --- a/src/frontend/sock.c +++ b/src/frontend/sock.c @@ -28,13 +28,14 @@ #include "sysdb.h" #include "core/error.h" #include "core/object.h" +#include "frontend/connection.h" #include "frontend/sock.h" #include "utils/channel.h" #include "utils/llist.h" +#include "utils/strbuf.h" #include - #include #include @@ -62,14 +63,13 @@ */ typedef struct { - int fd; + sdb_object_t super; + + /* connection and client information */ struct sockaddr_storage client_addr; socklen_t client_addr_len; -} connection_t; -typedef struct { - sdb_object_t super; - connection_t conn; + sdb_conn_t conn; } connection_obj_t; #define CONN(obj) ((connection_obj_t *)(obj)) @@ -181,6 +181,7 @@ listener_destroy(listener_t *listener) if (listener->sock_fd >= 0) close(listener->sock_fd); + listener->sock_fd = -1; if (listener->address) free(listener->address); @@ -232,6 +233,48 @@ listener_create(sdb_fe_socket_t *sock, const char *address) return listener; } /* listener_create */ +static int +listener_listen(listener_t *listener) +{ + assert(listener); + + /* try to reopen */ + if (listener->sock_fd < 0) + if (listener_impls[listener->type].opener(listener)) + return -1; + assert(listener->sock_fd >= 0); + + if (listen(listener->sock_fd, /* backlog = */ 32)) { + char buf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: Failed to listen on socket %s: %s", + listener->address, sdb_strerror(errno, buf, sizeof(buf))); + return -1; + } + return 0; +} /* listener_listen */ + +static void +listener_close(listener_t *listener) +{ + assert(listener); + + if (listener->sock_fd < 0) + return; + + close(listener->sock_fd); + listener->sock_fd = -1; +} /* listener_close */ + +static void +socket_close(sdb_fe_socket_t *sock) +{ + size_t i; + + assert(sock); + for (i = 0; i < sock->listeners_num; ++i) + listener_close(sock->listeners + i); +} /* socket_close */ + /* * private data types */ @@ -239,20 +282,27 @@ listener_create(sdb_fe_socket_t *sock, const char *address) static int connection_init(sdb_object_t *obj, va_list ap) { - connection_t *conn; + connection_obj_t *conn; int sock_fd; int sock_fl; assert(obj); - conn = &CONN(obj)->conn; + conn = CONN(obj); sock_fd = va_arg(ap, int); + CONN(obj)->conn.buf = sdb_strbuf_create(/* size = */ 128); + if (! CONN(obj)->conn.buf) { + sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate a read buffer " + "for a new remote connection"); + return -1; + } + conn->client_addr_len = sizeof(conn->client_addr); - conn->fd = accept(sock_fd, (struct sockaddr *)&conn->client_addr, + conn->conn.fd = accept(sock_fd, (struct sockaddr *)&conn->client_addr, &conn->client_addr_len); - if (conn->fd < 0) { + if (conn->conn.fd < 0) { char buf[1024]; sdb_log(SDB_LOG_ERR, "frontend: Failed to accept remote " "connection: %s", sdb_strerror(errno, @@ -266,35 +316,44 @@ connection_init(sdb_object_t *obj, va_list ap) return -1; } - sock_fl = fcntl(conn->fd, F_GETFL); - if (fcntl(conn->fd, F_SETFL, sock_fl | O_NONBLOCK)) { + sock_fl = fcntl(conn->conn.fd, F_GETFL); + if (fcntl(conn->conn.fd, F_SETFL, sock_fl | O_NONBLOCK)) { char buf[1024]; sdb_log(SDB_LOG_ERR, "frontend: Failed to switch connection conn#%i " - "to non-blocking mode: %s", conn->fd, + "to non-blocking mode: %s", conn->conn.fd, sdb_strerror(errno, buf, sizeof(buf))); return -1; } sdb_log(SDB_LOG_DEBUG, "frontend: Accepted connection on fd=%i", - conn->fd); + conn->conn.fd); /* update the object name */ snprintf(obj->name + strlen(CONN_FD_PREFIX), - strlen(CONN_FD_PLACEHOLDER), "%i", conn->fd); + strlen(CONN_FD_PLACEHOLDER), "%i", conn->conn.fd); return 0; } /* connection_init */ static void connection_destroy(sdb_object_t *obj) { - connection_t *conn; + connection_obj_t *conn; + size_t len; assert(obj); - conn = &CONN(obj)->conn; + conn = CONN(obj); + + len = sdb_strbuf_len(conn->conn.buf); + if (len) + sdb_log(SDB_LOG_INFO, "frontend: Discarding incomplete command " + "(%zu byte%s left in buffer)", len, len == 1 ? "" : "s"); - sdb_log(SDB_LOG_DEBUG, "frontend: Closing connection on fd=%i", conn->fd); - close(conn->fd); - conn->fd = -1; + sdb_log(SDB_LOG_DEBUG, "frontend: Closing connection on fd=%i", + conn->conn.fd); + close(conn->conn.fd); + conn->conn.fd = -1; + + sdb_strbuf_destroy(CONN(obj)->conn.buf); } /* connection_destroy */ static sdb_type_t connection_type = { @@ -307,35 +366,6 @@ static sdb_type_t connection_type = { * connection handler functions */ -/* returns negative value on error, 0 on EOF, number of packets else */ -static int -connection_read(int fd) -{ - int n = 0; - - while (42) { - int32_t cmd; - ssize_t status; - - errno = 0; - status = read(fd, &cmd, sizeof(cmd)); - if (status < 0) { - if ((errno == EAGAIN) || (errno == EWOULDBLOCK)) - return n + 1; - return (int)status; - } - else if (! status) /* EOF */ - return 0; - - /* XXX */ - sdb_log(SDB_LOG_DEBUG, "frontend: read command %i from fd=%i", - cmd, fd); - ++n; - } - - return n + 1; -} /* connection_read */ - static void * connection_handler(void *data) { @@ -364,21 +394,22 @@ connection_handler(void *data) continue; } - status = connection_read(conn->conn.fd); + status = (int)sdb_connection_read(&conn->conn); if (status <= 0) { - /* error or EOF → close connection */ + /* error or EOF -> close connection */ sdb_object_deref(SDB_OBJ(conn)); + continue; } - else { - if (sdb_llist_append(sock->open_connections, SDB_OBJ(conn))) { - sdb_log(SDB_LOG_ERR, "frontend: Failed to re-append " - "connection %s to list of open connections", - SDB_OBJ(conn)->name); - } - - /* pass ownership back to list; or destroy in case of an error */ - sdb_object_deref(SDB_OBJ(conn)); + + /* return the connection to the main loop */ + if (sdb_llist_append(sock->open_connections, SDB_OBJ(conn))) { + sdb_log(SDB_LOG_ERR, "frontend: Failed to re-append " + "connection %s to list of open connections", + SDB_OBJ(conn)->name); } + + /* pass ownership back to list; or destroy in case of an error */ + sdb_object_deref(SDB_OBJ(conn)); } return NULL; } /* connection_handler */ @@ -408,6 +439,43 @@ connection_accept(sdb_fe_socket_t *sock, listener_t *listener) return 0; } /* connection_accept */ +static int +socket_handle_incoming(sdb_fe_socket_t *sock, + fd_set *ready, fd_set *exceptions) +{ + sdb_llist_iter_t *iter; + size_t i; + + for (i = 0; i < sock->listeners_num; ++i) { + listener_t *listener = sock->listeners + i; + if (FD_ISSET(listener->sock_fd, ready)) + if (connection_accept(sock, listener)) + continue; + } + + iter = sdb_llist_get_iter(sock->open_connections); + if (! iter) { + sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator " + "for open connections"); + return -1; + } + + while (sdb_llist_iter_has_next(iter)) { + sdb_object_t *obj = sdb_llist_iter_get_next(iter); + + if (FD_ISSET(CONN(obj)->conn.fd, exceptions)) + sdb_log(SDB_LOG_INFO, "Exception on fd %d", + CONN(obj)->conn.fd); + + if (FD_ISSET(CONN(obj)->conn.fd, ready)) { + sdb_llist_iter_remove_current(iter); + sdb_channel_write(sock->chan, &obj); + } + } + sdb_llist_iter_destroy(iter); + return 0; +} /* socket_handle_incoming */ + /* * public API */ @@ -473,18 +541,15 @@ sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) /* XXX: make the number of threads configurable */ pthread_t handler_threads[5]; - if ((! sock) || (! sock->listeners_num) || (! loop)) + if ((! sock) || (! sock->listeners_num) || (! loop) || sock->chan) return -1; FD_ZERO(&sockets); - for (i = 0; i < sock->listeners_num; ++i) { listener_t *listener = sock->listeners + i; - if (listen(listener->sock_fd, /* backlog = */ 32)) { - char buf[1024]; - sdb_log(SDB_LOG_ERR, "frontend: Failed to listen on socket %s: %s", - listener->address, sdb_strerror(errno, buf, sizeof(buf))); + if (listener_listen(listener)) { + socket_close(sock); return -1; } @@ -494,8 +559,10 @@ sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) } sock->chan = sdb_channel_create(1024, sizeof(connection_obj_t *)); - if (! sock->chan) + if (! sock->chan) { + socket_close(sock); return -1; + } memset(&handler_threads, 0, sizeof(handler_threads)); /* XXX: error handling */ @@ -504,26 +571,24 @@ sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) connection_handler, /* arg = */ sock); while (loop->do_loop) { + struct timeval timeout = { 1, 0 }; /* one second */ + sdb_llist_iter_t *iter; + + int max_fd = max_listen_fd; fd_set ready; fd_set exceptions; - int max_fd; int n; - struct timeval timeout = { 1, 0 }; /* one second */ - sdb_llist_iter_t *iter; - FD_ZERO(&ready); FD_ZERO(&exceptions); ready = sockets; - max_fd = max_listen_fd; - iter = sdb_llist_get_iter(sock->open_connections); if (! iter) { sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator " "for open connections"); - return -1; + break; } while (sdb_llist_iter_has_next(iter)) { @@ -546,47 +611,27 @@ sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) sdb_log(SDB_LOG_ERR, "frontend: Failed to monitor sockets: %s", sdb_strerror(errno, buf, sizeof(buf))); - return -1; + break; } - - if (! n) + else if (! n) continue; - for (i = 0; i < sock->listeners_num; ++i) { - listener_t *listener = sock->listeners + i; - if (FD_ISSET(listener->sock_fd, &ready)) - if (connection_accept(sock, listener)) - continue; - } - - iter = sdb_llist_get_iter(sock->open_connections); - if (! iter) { - sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator " - "for open connections"); - return -1; - } - - while (sdb_llist_iter_has_next(iter)) { - sdb_object_t *obj = sdb_llist_iter_get_next(iter); - - if (FD_ISSET(CONN(obj)->conn.fd, &exceptions)) - sdb_log(SDB_LOG_INFO, "Exception on fd %d", - CONN(obj)->conn.fd); - - if (FD_ISSET(CONN(obj)->conn.fd, &ready)) { - sdb_llist_iter_remove_current(iter); - sdb_channel_write(sock->chan, &obj); - } - } - sdb_llist_iter_destroy(iter); + /* handle new and open connections */ + if (socket_handle_incoming(sock, &ready, &exceptions)) + break; } + socket_close(sock); + sdb_log(SDB_LOG_INFO, "frontend: Waiting for connection handler threads " "to terminate"); if (! sdb_channel_shutdown(sock->chan)) for (i = 0; i < SDB_STATIC_ARRAY_LEN(handler_threads); ++i) pthread_join(handler_threads[i], NULL); /* else: we tried our best; let the operating system clean up */ + + sdb_channel_destroy(sock->chan); + sock->chan = NULL; return 0; } /* sdb_fe_sock_listen_and_server */