X-Git-Url: https://git.tokkee.org/?a=blobdiff_plain;f=src%2Ffrontend%2Fsock.c;h=cac49e1db1648b93cc7cd55d933383dd274103f9;hb=c2b7616c8a80c5de6356b675d55a4ee36415fd7b;hp=063b2ca3d1d6528d0a22c7304faab65408e398c2;hpb=1e227f75832c867d9174132dc7b9ffe0a4d9f96b;p=sysdb.git diff --git a/src/frontend/sock.c b/src/frontend/sock.c index 063b2ca..cac49e1 100644 --- a/src/frontend/sock.c +++ b/src/frontend/sock.c @@ -27,14 +27,18 @@ #include "sysdb.h" #include "core/error.h" +#include "core/object.h" +#include "frontend/connection-private.h" #include "frontend/sock.h" #include "utils/channel.h" +#include "utils/llist.h" +#include "utils/strbuf.h" #include - #include +#include #include #include @@ -52,12 +56,6 @@ * private data types */ -typedef struct { - int fd; - struct sockaddr_storage client_addr; - socklen_t client_addr_len; -} connection_t; - typedef struct { char *address; int type; @@ -75,6 +73,12 @@ typedef struct { struct sdb_fe_socket { listener_t *listeners; size_t listeners_num; + + sdb_llist_t *open_connections; + + /* channel used for communication between main + * and connection handler threads */ + sdb_channel_t *chan; }; /* @@ -90,8 +94,8 @@ open_unix_sock(listener_t *listener) listener->sock_fd = socket(AF_UNIX, SOCK_STREAM, 0); if (listener->sock_fd < 0) { char buf[1024]; - sdb_log(SDB_LOG_ERR, "sock: Failed to open UNIX socket: %s", - sdb_strerror(errno, buf, sizeof(buf))); + sdb_log(SDB_LOG_ERR, "frontend: Failed to open UNIX socket %s: %s", + listener->address, sdb_strerror(errno, buf, sizeof(buf))); return -1; } @@ -103,8 +107,8 @@ open_unix_sock(listener_t *listener) status = bind(listener->sock_fd, (struct sockaddr *)&sa, sizeof(sa)); if (status) { char buf[1024]; - sdb_log(SDB_LOG_ERR, "sock: Failed to bind to UNIX socket: %s", - sdb_strerror(errno, buf, sizeof(buf))); + sdb_log(SDB_LOG_ERR, "frontend: Failed to bind to UNIX socket %s: %s", + listener->address, sdb_strerror(errno, buf, sizeof(buf))); return -1; } return 0; @@ -160,6 +164,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); @@ -172,25 +177,37 @@ listener_create(sdb_fe_socket_t *sock, const char *address) int type; type = get_type(address); - if (type < 0) + if (type < 0) { + sdb_log(SDB_LOG_ERR, "frontend: Unsupported address type specified " + "in listen address '%s'", address); return NULL; + } listener = realloc(sock->listeners, sock->listeners_num * sizeof(*sock->listeners)); - if (! listener) + if (! listener) { + char buf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s", + sdb_strerror(errno, buf, sizeof(buf))); return NULL; + } + sock->listeners = listener; listener = sock->listeners + sock->listeners_num; listener->sock_fd = -1; listener->address = strdup(address); if (! listener->address) { + char buf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s", + sdb_strerror(errno, buf, sizeof(buf))); listener_destroy(listener); return NULL; } listener->type = type; if (listener_impls[type].opener(listener)) { + /* prints error */ listener_destroy(listener); return NULL; } @@ -199,6 +216,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 */ + /* * connection handler functions */ @@ -206,17 +265,18 @@ listener_create(sdb_fe_socket_t *sock, const char *address) static void * connection_handler(void *data) { - sdb_channel_t *chan = data; + sdb_fe_socket_t *sock = data; - assert(chan); + assert(sock); while (42) { struct timespec timeout = { 0, 500000000 }; /* .5 seconds */ - connection_t conn; + sdb_conn_t *conn; int status; errno = 0; - status = sdb_channel_select(chan, NULL, &conn, NULL, NULL, &timeout); + status = sdb_channel_select(sock->chan, /* read */ NULL, &conn, + /* write */ NULL, NULL, &timeout); if (status) { char buf[1024]; @@ -225,27 +285,89 @@ connection_handler(void *data) if (errno == EBADF) /* channel shut down */ break; - sdb_log(SDB_LOG_ERR, "sock: Failed to read from channel: %s", + sdb_log(SDB_LOG_ERR, "frontend: Failed to read from channel: %s", sdb_strerror(errno, buf, sizeof(buf))); continue; } - if (conn.fd < 0) + status = (int)sdb_connection_read(conn); + if (status <= 0) { + /* error or EOF -> close connection */ + sdb_object_deref(SDB_OBJ(conn)); continue; + } - if (conn.client_addr.ss_family != AF_UNIX) { - sdb_log(SDB_LOG_ERR, "sock: Accepted connection using unexpected " - "family type %d", conn.client_addr.ss_family); - continue; + /* 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); } - /* XXX */ - sdb_log(SDB_LOG_INFO, "Accepted connection on fd=%i\n", conn.fd); - close(conn.fd); + /* pass ownership back to list; or destroy in case of an error */ + sdb_object_deref(SDB_OBJ(conn)); } return NULL; } /* connection_handler */ +static int +connection_accept(sdb_fe_socket_t *sock, listener_t *listener) +{ + sdb_object_t *obj; + int status; + + obj = SDB_OBJ(sdb_connection_accept(listener->sock_fd)); + if (! obj) + return -1; + + status = sdb_llist_append(sock->open_connections, obj); + if (status) + sdb_log(SDB_LOG_ERR, "frontend: Failed to append " + "connection %s to list of open connections", + obj->name); + + /* hand ownership over to the list; or destroy in case of an error */ + sdb_object_deref(obj); + return status; +} /* 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)->fd, exceptions)) + sdb_log(SDB_LOG_INFO, "Exception on fd %d", + CONN(obj)->fd); + + if (FD_ISSET(CONN(obj)->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 */ @@ -258,6 +380,12 @@ sdb_fe_sock_create(void) sock = calloc(1, sizeof(*sock)); if (! sock) return NULL; + + sock->open_connections = sdb_llist_create(); + if (! sock->open_connections) { + sdb_fe_sock_destroy(sock); + return NULL; + } return sock; } /* sdb_fe_sock_create */ @@ -274,6 +402,10 @@ sdb_fe_sock_destroy(sdb_fe_socket_t *sock) } if (sock->listeners) free(sock->listeners); + sock->listeners = NULL; + + sdb_llist_destroy(sock->open_connections); + sock->open_connections = NULL; free(sock); } /* sdb_fe_sock_destroy */ @@ -294,98 +426,104 @@ sdb_fe_sock_add_listener(sdb_fe_socket_t *sock, const char *address) int sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) { - sdb_channel_t *chan; fd_set sockets; - int max_fd = 0; + int max_listen_fd = 0; size_t i; /* 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, "sock: Failed to listen on socket %s: %s", - listener->address, sdb_strerror(errno, buf, sizeof(buf))); + if (listener_listen(listener)) { + socket_close(sock); return -1; } FD_SET(listener->sock_fd, &sockets); - if (listener->sock_fd > max_fd) - max_fd = listener->sock_fd; + if (listener->sock_fd > max_listen_fd) + max_listen_fd = listener->sock_fd; } - chan = sdb_channel_create(1024, sizeof(connection_t)); - if (! chan) + sock->chan = sdb_channel_create(1024, sizeof(sdb_conn_t *)); + if (! sock->chan) { + socket_close(sock); return -1; + } memset(&handler_threads, 0, sizeof(handler_threads)); /* XXX: error handling */ for (i = 0; i < SDB_STATIC_ARRAY_LEN(handler_threads); ++i) pthread_create(&handler_threads[i], /* attr = */ NULL, - connection_handler, /* arg = */ chan); + connection_handler, /* arg = */ sock); while (loop->do_loop) { - fd_set ready = sockets; + 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 n; - struct timeval timeout = { 1, 0 }; /* one second */ + FD_ZERO(&ready); + FD_ZERO(&exceptions); + + ready = sockets; + + iter = sdb_llist_get_iter(sock->open_connections); + if (! iter) { + sdb_log(SDB_LOG_ERR, "frontend: Failed to acquire iterator " + "for open connections"); + break; + } + + while (sdb_llist_iter_has_next(iter)) { + sdb_object_t *obj = sdb_llist_iter_get_next(iter); + FD_SET(CONN(obj)->fd, &ready); + FD_SET(CONN(obj)->fd, &exceptions); + + if (CONN(obj)->fd > max_fd) + max_fd = CONN(obj)->fd; + } + sdb_llist_iter_destroy(iter); errno = 0; - n = select(max_fd + 1, &ready, NULL, NULL, &timeout); + n = select(max_fd + 1, &ready, NULL, &exceptions, &timeout); if (n < 0) { char buf[1024]; if (errno == EINTR) continue; - sdb_log(SDB_LOG_ERR, "sock: Failed to monitor sockets: %s", + 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)) { - connection_t conn; - - memset(&conn, 0, sizeof(conn)); - conn.client_addr_len = sizeof(conn.client_addr); - - conn.fd = accept(listener->sock_fd, - (struct sockaddr *)&conn.client_addr, - &conn.client_addr_len); - - if (conn.fd < 0) { - char buf[1024]; - sdb_log(SDB_LOG_ERR, "sock: Failed to accept remote " - "connection: %s", sdb_strerror(errno, - buf, sizeof(buf))); - continue; - } - - sdb_channel_write(chan, &conn); - } - } + /* handle new and open connections */ + if (socket_handle_incoming(sock, &ready, &exceptions)) + break; } - sdb_log(SDB_LOG_INFO, "sock: Waiting for connection handler threads " + socket_close(sock); + + sdb_log(SDB_LOG_INFO, "frontend: Waiting for connection handler threads " "to terminate"); - if (! sdb_channel_shutdown(chan)) + 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 */