X-Git-Url: https://git.tokkee.org/?a=blobdiff_plain;f=src%2Ffrontend%2Fsock.c;h=5279cbae84fd44d1c05623537a1bbf468860653b;hb=23b12088ee320df7b54f5c81011daf8995e22203;hp=65884ea945bf61fe935b48fe60ab819b453b1eb8;hpb=ec6233c810ce2115eb1ce1b3f10d6833f0ed33ea;p=sysdb.git diff --git a/src/frontend/sock.c b/src/frontend/sock.c index 65884ea..5279cba 100644 --- a/src/frontend/sock.c +++ b/src/frontend/sock.c @@ -25,6 +25,10 @@ * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. */ +#if HAVE_CONFIG_H +# include "config.h" +#endif /* HAVE_CONFIG_H */ + #include "sysdb.h" #include "core/object.h" #include "frontend/connection-private.h" @@ -33,6 +37,7 @@ #include "utils/channel.h" #include "utils/error.h" #include "utils/llist.h" +#include "utils/os.h" #include "utils/strbuf.h" #include @@ -48,8 +53,21 @@ #include #include #include +#include #include +#ifdef HAVE_UCRED_H +# include +#endif +#ifdef HAVE_SYS_UCRED_H +# include +#endif + +#include + +#include +#include +#include #include /* @@ -61,20 +79,29 @@ typedef struct { int type; int sock_fd; + int (*setup)(sdb_conn_t *, void *); } listener_t; typedef struct { int type; const char *prefix; - int (*opener)(listener_t *); + int (*open)(listener_t *); + void (*close)(listener_t *); } fe_listener_impl_t; struct sdb_fe_socket { listener_t *listeners; size_t listeners_num; + /* list of open, idle connections; active connections will be passed on to + * the connection handler threads which place them back after handling + * pending actions; the trigger pipe is used to notify the main thread + * after adding a connection (back) to the list */ sdb_llist_t *open_connections; + int trigger[2]; +#define TRIGGER_READ 0 +#define TRIGGER_WRITE 1 /* channel used for communication between main * and connection handler threads */ @@ -86,8 +113,50 @@ struct sdb_fe_socket { */ static int -open_unix_sock(listener_t *listener) +setup_unixsock(sdb_conn_t *conn, void __attribute__((unused)) *user_data) +{ + uid_t uid; + + struct passwd pw_entry; + struct passwd *result = NULL; + char buf[1024]; + +#ifdef SO_PEERCRED + struct ucred cred; + socklen_t len = sizeof(cred); + + if (getsockopt(conn->fd, SOL_SOCKET, SO_PEERCRED, &cred, &len) + || (len != sizeof(cred))) { + char errbuf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: Failed to determine peer for " + "connection conn#%i: %s", conn->fd, + sdb_strerror(errno, errbuf, sizeof(errbuf))); + return -1; + } + uid = cred.uid; +#else /* SO_PEERCRED */ + sdb_log(SDB_LOG_ERR, "frontend: Failed to determine peer for " + "connection conn#%i: operation not supported", conn->fd); + return -1; +#endif + + memset(&pw_entry, 0, sizeof(pw_entry)); + if (getpwuid_r(uid, &pw_entry, buf, sizeof(buf), &result) || (! result) + || (! (conn->username = strdup(result->pw_name)))) { + char errbuf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: Failed to determine peer for " + "connection conn#%i: %s", conn->fd, + sdb_strerror(errno, errbuf, sizeof(errbuf))); + return -1; + } + return 0; +} /* setup_unixsock */ + +static int +open_unixsock(listener_t *listener) { + char *addr_copy; + char *base_dir; struct sockaddr_un sa; int status; @@ -101,13 +170,31 @@ open_unix_sock(listener_t *listener) memset(&sa, 0, sizeof(sa)); sa.sun_family = AF_UNIX; - strncpy(sa.sun_path, listener->address + strlen("unix:"), - sizeof(sa.sun_path)); + strncpy(sa.sun_path, listener->address, sizeof(sa.sun_path)); - if (unlink(listener->address + strlen("unix:")) && (errno != ENOENT)) { + addr_copy = strdup(listener->address); + if (! addr_copy) { + char errbuf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: strdup failed: %s", + sdb_strerror(errno, errbuf, sizeof(errbuf))); + return -1; + } + base_dir = dirname(addr_copy); + + /* ensure that the directory exists */ + if (sdb_mkdir_all(base_dir, 0777)) { + char errbuf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: Failed to create directory '%s': %s", + base_dir, sdb_strerror(errno, errbuf, sizeof(errbuf))); + free(addr_copy); + return -1; + } + free(addr_copy); + + if (unlink(listener->address) && (errno != ENOENT)) { char errbuf[1024]; sdb_log(SDB_LOG_WARNING, "frontend: Failed to remove stale UNIX " - "socket %s: %s", listener->address + strlen("unix:"), + "socket %s: %s", listener->address, sdb_strerror(errno, errbuf, sizeof(errbuf))); } @@ -118,8 +205,90 @@ open_unix_sock(listener_t *listener) listener->address, sdb_strerror(errno, buf, sizeof(buf))); return -1; } + + listener->setup = setup_unixsock; + return 0; +} /* open_unixsock */ + +static void +close_unixsock(listener_t *listener) +{ + assert(listener); + + if (! listener->address) + return; + + if (listener->sock_fd >= 0) + close(listener->sock_fd); + listener->sock_fd = -1; + + unlink(listener->address); +} /* close_unixsock */ + +static int +open_tcp(listener_t *listener) +{ + struct addrinfo *ai, *ai_list = NULL; + int status; + + assert(listener); + + if ((status = sdb_resolve(SDB_NET_TCP, listener->address, &ai_list))) { + sdb_log(SDB_LOG_ERR, "frontend: Failed to resolve '%s': %s", + listener->address, gai_strerror(status)); + return -1; + } + + for (ai = ai_list; ai != NULL; ai = ai->ai_next) { + char errbuf[1024]; + int reuse = 1; + + listener->sock_fd = socket(ai->ai_family, + ai->ai_socktype, ai->ai_protocol); + if (listener->sock_fd < 0) { + sdb_log(SDB_LOG_ERR, "frontend: Failed to open socket for %s: %s", + listener->address, + sdb_strerror(errno, errbuf, sizeof(errbuf))); + continue; + } + + if (setsockopt(listener->sock_fd, SOL_SOCKET, SO_REUSEADDR, + &reuse, sizeof(reuse)) < 0) { + sdb_log(SDB_LOG_ERR, "frontend: Failed to set socket option: %s", + sdb_strerror(errno, errbuf, sizeof(errbuf))); + close(listener->sock_fd); + listener->sock_fd = -1; + continue; + } + + if (bind(listener->sock_fd, ai->ai_addr, ai->ai_addrlen) < 0) { + char host[1024], port[32]; + getnameinfo(ai->ai_addr, ai->ai_addrlen, host, sizeof(host), + port, sizeof(port), NI_NUMERICHOST | NI_NUMERICSERV); + sdb_log(SDB_LOG_ERR, "frontend: Failed to bind to %s:%s: %s", + host, port, sdb_strerror(errno, errbuf, sizeof(errbuf))); + close(listener->sock_fd); + listener->sock_fd = -1; + continue; + } + break; + } + freeaddrinfo(ai_list); + + if (listener->sock_fd < 0) + return -1; return 0; -} /* open_unix_sock */ +} /* open_tcp */ + +static void +close_tcp(listener_t *listener) +{ + assert(listener); + + if (listener->sock_fd >= 0) + close(listener->sock_fd); + listener->sock_fd = -1; +} /* close_tcp */ /* * private variables @@ -128,16 +297,51 @@ open_unix_sock(listener_t *listener) /* the enum has to be sorted the same as the implementations array * to ensure that the type may be used as index into the array */ enum { - LISTENER_UNIXSOCK = 0, + LISTENER_TCP = 0, /* this is the default */ + LISTENER_UNIXSOCK, }; static fe_listener_impl_t listener_impls[] = { - { LISTENER_UNIXSOCK, "unix", open_unix_sock }, + { LISTENER_TCP, "tcp", open_tcp, close_tcp }, + { LISTENER_UNIXSOCK, "unix", open_unixsock, close_unixsock }, }; /* * private helper functions */ +static int +listener_listen(listener_t *listener) +{ + assert(listener); + + /* try to reopen */ + if (listener->sock_fd < 0) + if (listener_impls[listener->type].open(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_impls[listener->type].close) + listener_impls[listener->type].close(listener); + + if (listener->sock_fd >= 0) + close(listener->sock_fd); + listener->sock_fd = -1; +} /* listener_close */ + static int get_type(const char *address) { @@ -145,9 +349,11 @@ get_type(const char *address) size_t len; size_t i; + if (*address == '/') + return LISTENER_UNIXSOCK; sep = strchr(address, (int)':'); if (! sep) - return -1; + return listener_impls[0].type; assert(sep > address); len = (size_t)(sep - address); @@ -169,18 +375,18 @@ listener_destroy(listener_t *listener) if (! listener) return; - if (listener->sock_fd >= 0) - close(listener->sock_fd); - listener->sock_fd = -1; + listener_close(listener); if (listener->address) free(listener->address); + listener->address = NULL; } /* listener_destroy */ static listener_t * listener_create(sdb_fe_socket_t *sock, const char *address) { listener_t *listener; + size_t len; int type; type = get_type(address); @@ -191,7 +397,7 @@ listener_create(sdb_fe_socket_t *sock, const char *address) } listener = realloc(sock->listeners, - sock->listeners_num * sizeof(*sock->listeners)); + (sock->listeners_num + 1) * sizeof(*sock->listeners)); if (! listener) { char buf[1024]; sdb_log(SDB_LOG_ERR, "frontend: Failed to allocate memory: %s", @@ -202,6 +408,11 @@ listener_create(sdb_fe_socket_t *sock, const char *address) sock->listeners = listener; listener = sock->listeners + sock->listeners_num; + len = strlen(listener_impls[type].prefix); + if ((! strncmp(address, listener_impls[type].prefix, len)) + && (address[len] == ':')) + address += strlen(listener_impls[type].prefix) + 1; + listener->sock_fd = -1; listener->address = strdup(address); if (! listener->address) { @@ -212,8 +423,9 @@ listener_create(sdb_fe_socket_t *sock, const char *address) return NULL; } listener->type = type; + listener->setup = NULL; - if (listener_impls[type].opener(listener)) { + if (listener_impls[type].open(listener)) { /* prints error */ listener_destroy(listener); return NULL; @@ -223,37 +435,19 @@ 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) +socket_clear(sdb_fe_socket_t *sock) { - assert(listener); - - if (listener->sock_fd < 0) - return; + size_t i; - close(listener->sock_fd); - listener->sock_fd = -1; -} /* listener_close */ + assert(sock); + for (i = 0; i < sock->listeners_num; ++i) + listener_destroy(sock->listeners + i); + if (sock->listeners) + free(sock->listeners); + sock->listeners = NULL; + sock->listeners_num = 0; +} /* socket_clear */ static void socket_close(sdb_fe_socket_t *sock) @@ -297,7 +491,7 @@ connection_handler(void *data) continue; } - status = (int)sdb_connection_read(conn); + status = (int)sdb_connection_handle(conn); if (status <= 0) { /* error or EOF -> close connection */ sdb_object_deref(SDB_OBJ(conn)); @@ -310,6 +504,7 @@ connection_handler(void *data) "connection %s to list of open connections", SDB_OBJ(conn)->name); } + write(sock->trigger[TRIGGER_WRITE], "", 1); /* pass ownership back to list; or destroy in case of an error */ sdb_object_deref(SDB_OBJ(conn)); @@ -323,7 +518,8 @@ 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)); + obj = SDB_OBJ(sdb_connection_accept(listener->sock_fd, + listener->setup, NULL)); if (! obj) return -1; @@ -362,9 +558,14 @@ socket_handle_incoming(sdb_fe_socket_t *sock, while (sdb_llist_iter_has_next(iter)) { sdb_object_t *obj = sdb_llist_iter_get_next(iter); - if (FD_ISSET(CONN(obj)->fd, exceptions)) + if (FD_ISSET(CONN(obj)->fd, exceptions)) { sdb_log(SDB_LOG_INFO, "Exception on fd %d", CONN(obj)->fd); + /* close the connection */ + sdb_llist_iter_remove_current(iter); + sdb_object_deref(obj); + continue; + } if (FD_ISSET(CONN(obj)->fd, ready)) { sdb_llist_iter_remove_current(iter); @@ -383,33 +584,60 @@ sdb_fe_socket_t * sdb_fe_sock_create(void) { sdb_fe_socket_t *sock; + int flags; sock = calloc(1, sizeof(*sock)); if (! sock) return NULL; + sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1; sock->open_connections = sdb_llist_create(); if (! sock->open_connections) { sdb_fe_sock_destroy(sock); return NULL; } + + if (pipe(sock->trigger)) { + char errbuf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: Failed to create pipe: %s", + sdb_strerror(errno, errbuf, sizeof(errbuf))); + sdb_fe_sock_destroy(sock); + return NULL; + } + /* TODO: Can we do a zero-byte write as well to trigger select()? + * Linux does not seem to support the I_SWROPT ioctl on a pipe. */ + flags = fcntl(sock->trigger[TRIGGER_WRITE], F_GETFL); + if (fcntl(sock->trigger[TRIGGER_WRITE], F_SETFL, flags | O_NONBLOCK)) { + char errbuf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: Failed to switch pipe to non-blocking " + "mode: %s", sdb_strerror(errno, errbuf, sizeof(errbuf))); + sdb_fe_sock_destroy(sock); + return NULL; + } + flags = fcntl(sock->trigger[TRIGGER_READ], F_GETFL); + if (fcntl(sock->trigger[TRIGGER_READ], F_SETFL, flags | O_NONBLOCK)) { + char errbuf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: Failed to switch pipe to non-blocking " + "mode: %s", sdb_strerror(errno, errbuf, sizeof(errbuf))); + sdb_fe_sock_destroy(sock); + return NULL; + } return sock; } /* sdb_fe_sock_create */ void sdb_fe_sock_destroy(sdb_fe_socket_t *sock) { - size_t i; - if (! sock) return; - for (i = 0; i < sock->listeners_num; ++i) { - listener_destroy(sock->listeners + i); - } - if (sock->listeners) - free(sock->listeners); - sock->listeners = NULL; + socket_clear(sock); + + if (sock->trigger[TRIGGER_WRITE] >= 0) + close(sock->trigger[TRIGGER_WRITE]); + if (sock->trigger[TRIGGER_READ] >= 0) + close(sock->trigger[TRIGGER_READ]); + sock->trigger[TRIGGER_READ] = sock->trigger[TRIGGER_WRITE] = -1; sdb_llist_destroy(sock->open_connections); sock->open_connections = NULL; @@ -430,6 +658,15 @@ sdb_fe_sock_add_listener(sdb_fe_socket_t *sock, const char *address) return 0; } /* sdb_fe_sock_add_listener */ +void +sdb_fe_sock_clear_listeners(sdb_fe_socket_t *sock) +{ + if (! sock) + return; + + socket_clear(sock); +} /* sdb_fe_sock_clear_listeners */ + int sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) { @@ -437,12 +674,16 @@ sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) int max_listen_fd = 0; size_t i; - /* XXX: make the number of threads configurable */ - pthread_t handler_threads[5]; + pthread_t handler_threads[loop->num_threads]; + size_t num_threads; - if ((! sock) || (! sock->listeners_num) || (! loop) || sock->chan) + if ((! sock) || (! sock->listeners_num) || sock->chan + || (! loop) || (loop->num_threads <= 0)) return -1; + if (! loop->do_loop) + return 0; + FD_ZERO(&sockets); for (i = 0; i < sock->listeners_num; ++i) { listener_t *listener = sock->listeners + i; @@ -463,13 +704,27 @@ sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) return -1; } + sdb_log(SDB_LOG_INFO, "frontend: Starting %zu connection " + "handler thread%s managing %zu listener%s", + loop->num_threads, loop->num_threads == 1 ? "" : "s", + sock->listeners_num, sock->listeners_num == 1 ? "" : "s"); + + num_threads = loop->num_threads; 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 = */ sock); + for (i = 0; i < num_threads; ++i) { + errno = 0; + if (pthread_create(&handler_threads[i], /* attr = */ NULL, + connection_handler, /* arg = */ sock)) { + char errbuf[1024]; + sdb_log(SDB_LOG_ERR, "frontend: Failed to create " + "connection handler thread: %s", + sdb_strerror(errno, errbuf, sizeof(errbuf))); + num_threads = i; + break; + } + } - while (loop->do_loop) { + while (loop->do_loop && num_threads) { struct timeval timeout = { 1, 0 }; /* one second */ sdb_llist_iter_t *iter; @@ -482,6 +737,7 @@ sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) FD_ZERO(&exceptions); ready = sockets; + FD_SET(sock->trigger[TRIGGER_READ], &ready); iter = sdb_llist_get_iter(sock->open_connections); if (! iter) { @@ -492,6 +748,13 @@ sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) while (sdb_llist_iter_has_next(iter)) { sdb_object_t *obj = sdb_llist_iter_get_next(iter); + + if (CONN(obj)->fd < 0) { + sdb_llist_iter_remove_current(iter); + sdb_object_deref(obj); + continue; + } + FD_SET(CONN(obj)->fd, &ready); FD_SET(CONN(obj)->fd, &exceptions); @@ -515,6 +778,12 @@ sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) else if (! n) continue; + if (FD_ISSET(sock->trigger[TRIGGER_READ], &ready)) { + char buf[1024]; + while (read(sock->trigger[TRIGGER_READ], buf, sizeof(buf)) > 0) + /* do nothing */; + } + /* handle new and open connections */ if (socket_handle_incoming(sock, &ready, &exceptions)) break; @@ -525,12 +794,15 @@ sdb_fe_sock_listen_and_serve(sdb_fe_socket_t *sock, sdb_fe_loop_t *loop) 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) + for (i = 0; i < num_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; + + if (! num_threads) + return -1; return 0; } /* sdb_fe_sock_listen_and_server */