frontend: Make connection I/O handling more flexible.

[sysdb.git] / src / frontend / sock.c

diff --git a/src/frontend/sock.c b/src/frontend/sock.c
index 90aeab1406a6a6389f8ce083dbab72a624b4a936..642f03f92b0f87d25e3783bb501ecbce793ff6c3 100644 (file)
--- a/src/frontend/sock.c
+++ b/src/frontend/sock.c
@@ -37,6 +37,7 @@
 #include "utils/channel.h"
 #include "utils/error.h"
 #include "utils/llist.h"
+#include "utils/os.h"
 #include "utils/strbuf.h"
 
 #include <assert.h>
@@ -54,6 +55,8 @@
 #include <sys/socket.h>
 #include <sys/un.h>
 
+#include <libgen.h>
+
 #include <pthread.h>
 
 /*
@@ -65,14 +68,15 @@ typedef struct {
        int   type;
 
        int sock_fd;
+       int (*accept)(sdb_conn_t *);
 } listener_t;
 
 typedef struct {
        int type;
        const char *prefix;
 
-       int (*opener)(listener_t *);
-       void (*closer)(listener_t *);
+       int (*open)(listener_t *);
+       void (*close)(listener_t *);
 } fe_listener_impl_t;
 
 struct sdb_fe_socket {
@@ -93,6 +97,9 @@ struct sdb_fe_socket {
 static int
 open_unix_sock(listener_t *listener)
 {
+       const char *addr;
+       char *addr_copy;
+       char *base_dir;
        struct sockaddr_un sa;
        int status;
 
@@ -104,12 +111,35 @@ open_unix_sock(listener_t *listener)
                return -1;
        }
 
+       if (*listener->address == '/')
+               addr = listener->address;
+       else
+               addr = listener->address + strlen("unix:");
+
        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, addr, sizeof(sa.sun_path));
 
-       if (unlink(listener->address + strlen("unix:")) && (errno != ENOENT)) {
+       addr_copy = strdup(addr);
+       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(addr) && (errno != ENOENT)) {
                char errbuf[1024];
                sdb_log(SDB_LOG_WARNING, "frontend: Failed to remove stale UNIX "
                                "socket %s: %s", listener->address + strlen("unix:"),
@@ -129,15 +159,22 @@ open_unix_sock(listener_t *listener)
 static void
 close_unix_sock(listener_t *listener)
 {
+       const char *addr;
        assert(listener);
+
        if (! listener->address)
                return;
 
+       if (*listener->address == '/')
+               addr = listener->address;
+       else
+               addr = listener->address + strlen("unix:");
+
        if (listener->sock_fd >= 0)
                close(listener->sock_fd);
        listener->sock_fd = -1;
 
-       unlink(listener->address + strlen("unix:"));
+       unlink(addr);
 } /* close_unix_sock */
 
 /*
@@ -147,7 +184,7 @@ close_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_UNIXSOCK = 0, /* this is the default */
 };
 static fe_listener_impl_t listener_impls[] = {
        { LISTENER_UNIXSOCK, "unix", open_unix_sock, close_unix_sock },
@@ -164,7 +201,7 @@ listener_listen(listener_t *listener)
 
        /* try to reopen */
        if (listener->sock_fd < 0)
-               if (listener_impls[listener->type].opener(listener))
+               if (listener_impls[listener->type].open(listener))
                        return -1;
        assert(listener->sock_fd >= 0);
 
@@ -182,8 +219,8 @@ listener_close(listener_t *listener)
 {
        assert(listener);
 
-       if (listener_impls[listener->type].closer)
-               listener_impls[listener->type].closer(listener);
+       if (listener_impls[listener->type].close)
+               listener_impls[listener->type].close(listener);
 
        if (listener->sock_fd >= 0)
                close(listener->sock_fd);
@@ -199,7 +236,7 @@ get_type(const char *address)
 
        sep = strchr(address, (int)':');
        if (! sep)
-               return -1;
+               return listener_impls[0].type;
 
        assert(sep > address);
        len = (size_t)(sep - address);
@@ -263,8 +300,9 @@ listener_create(sdb_fe_socket_t *sock, const char *address)
                return NULL;
        }
        listener->type = type;
+       listener->accept = NULL;
 
-       if (listener_impls[type].opener(listener)) {
+       if (listener_impls[type].open(listener)) {
                /* prints error */
                listener_destroy(listener);
                return NULL;
@@ -330,7 +368,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));
@@ -360,6 +398,12 @@ connection_accept(sdb_fe_socket_t *sock, listener_t *listener)
        if (! obj)
                return -1;
 
+       if (listener->accept && listener->accept(CONN(obj))) {
+               /* accept() is expected to log an error */
+               sdb_object_deref(obj);
+               return -1;
+       }
+
        status = sdb_llist_append(sock->open_connections, obj);
        if (status)
                sdb_log(SDB_LOG_ERR, "frontend: Failed to append "
@@ -507,8 +551,8 @@ 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 %d connection "
-                       "handler thread%s managing %d listener%s",
+       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");