Introduce unix socket connection type

Unix socket uses different accept handler/create listener from TCP,
to hide these difference to avoid hard code, use a new unix socket
connection type. Also move 'acceptUnixHandler' into unix.c.

Currently, the connection framework becomes like following:

                   uplayer
                      |
               connection layer
                 /    |     \
               TCP   Unix   TLS

It's possible to build Unix socket support as a shared library, and
load it dynamically. Because TCP and Unix socket don't require any
heavy dependencies or overheads, we build them into Redis statically.

Signed-off-by: zhenwei pi <pizhenwei@bytedance.com>

src/Makefile

@@ -316,7 +316,7 @@ endif
 
 REDIS_SERVER_NAME=redis-server$(PROG_SUFFIX)
 REDIS_SENTINEL_NAME=redis-sentinel$(PROG_SUFFIX)
-REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script_lua.o script.o functions.o function_lua.o commands.o strl.o connection.o
+REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o eval.o bio.o rio.o rand.o memtest.o syscheck.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o socket.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script_lua.o script.o functions.o function_lua.o commands.o strl.o connection.o unix.o
 REDIS_CLI_NAME=redis-cli$(PROG_SUFFIX)
 REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o ae.o redisassert.o crcspeed.o crc64.o siphash.o crc16.o monotonic.o cli_common.o mt19937-64.o strl.o
 REDIS_BENCHMARK_NAME=redis-benchmark$(PROG_SUFFIX)

src/connection.c

@@ -56,6 +56,9 @@ int connTypeInitialize() {
     /* currently socket connection type is necessary  */
     serverAssert(RedisRegisterConnectionTypeSocket() == C_OK);
 
+    /* currently unix socket connection type is necessary  */
+    serverAssert(RedisRegisterConnectionTypeUnix() == C_OK);
+
     /* may fail if without BUILD_TLS=yes */
     RedisRegisterConnectionTypeTLS();
 

src/connection.h

@@ -58,8 +58,9 @@ typedef enum {
 #define CONN_FLAG_WRITE_BARRIER     (1<<1)      /* Write barrier requested */
 
 #define CONN_TYPE_SOCKET            0
-#define CONN_TYPE_TLS               1
-#define CONN_TYPE_MAX               2
+#define CONN_TYPE_UNIX              1
+#define CONN_TYPE_TLS               2
+#define CONN_TYPE_MAX               3
 
 typedef void (*ConnectionCallbackFunc)(struct connection *conn);
 
@@ -394,6 +395,7 @@ static inline aeFileProc *connAcceptHandler(int type) {
 }
 
 int RedisRegisterConnectionTypeSocket();
+int RedisRegisterConnectionTypeUnix();
 int RedisRegisterConnectionTypeTLS();
 
 #endif  /* __REDIS_CONNECTION_H */

src/networking.c

@@ -1327,25 +1327,6 @@ void acceptCommonHandler(connection *conn, int flags, char *ip) {
     }
 }
 
-void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
-    int cfd, max = MAX_ACCEPTS_PER_CALL;
-    UNUSED(el);
-    UNUSED(mask);
-    UNUSED(privdata);
-
-    while(max--) {
-        cfd = anetUnixAccept(server.neterr, fd);
-        if (cfd == ANET_ERR) {
-            if (errno != EWOULDBLOCK)
-                serverLog(LL_WARNING,
-                    "Accepting client connection: %s", server.neterr);
-            return;
-        }
-        serverLog(LL_VERBOSE,"Accepted connection to %s", server.unixsocket);
-        acceptCommonHandler(connCreateAccepted(CONN_TYPE_SOCKET, cfd, NULL),CLIENT_UNIX_SOCKET,NULL);
-    }
-}
-
 void freeClientOriginalArgv(client *c) {
     /* We didn't rewrite this client */
     if (!c->original_argv) return;

src/server.c

@@ -2592,7 +2592,7 @@ void initServer(void) {
     if (createSocketAcceptHandler(&server.tlsfd, connAcceptHandler(CONN_TYPE_TLS)) != C_OK) {
         serverPanic("Unrecoverable error creating TLS socket accept handler.");
     }
-    if (createSocketAcceptHandler(&server.sofd, acceptUnixHandler) != C_OK) {
+    if (createSocketAcceptHandler(&server.sofd, connAcceptHandler(CONN_TYPE_UNIX)) != C_OK) {
         serverPanic("Unrecoverable error creating server.sofd file event.");
     }
 

src/server.h

@@ -2461,7 +2461,6 @@ void setDeferredAttributeLen(client *c, void *node, long length);
 void setDeferredPushLen(client *c, void *node, long length);
 int processInputBuffer(client *c);
 void acceptCommonHandler(connection *conn, int flags, char *ip);
-void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask);
 void readQueryFromClient(connection *conn);
 int prepareClientToWrite(client *c);
 void addReplyNull(client *c);

src/unix.c

@@ -0,0 +1,168 @@
+/* ==========================================================================
+ * unix.c - unix socket connection implementation
+ * --------------------------------------------------------------------------
+ * Copyright (C) 2022  zhenwei pi
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a
+ * copy of this software and associated documentation files (the
+ * "Software"), to deal in the Software without restriction, including
+ * without limitation the rights to use, copy, modify, merge, publish,
+ * distribute, sublicense, and/or sell copies of the Software, and to permit
+ * persons to whom the Software is furnished to do so, subject to the
+ * following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included
+ * in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
+ * OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN
+ * NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
+ * DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+ * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE
+ * USE OR OTHER DEALINGS IN THE SOFTWARE.
+ * ==========================================================================
+ */
+
+#include "server.h"
+#include "connection.h"
+
+static ConnectionType CT_Unix;
+
+static int connUnixGetType(connection *conn) {
+    (void) conn;
+
+    return CONN_TYPE_UNIX;
+}
+
+static void connUnixEventHandler(struct aeEventLoop *el, int fd, void *clientData, int mask) {
+    connectionByType(CONN_TYPE_SOCKET)->ae_handler(el, fd, clientData, mask);
+}
+
+static int connUnixAddr(connection *conn, char *ip, size_t ip_len, int *port, int remote) {
+    return connectionByType(CONN_TYPE_SOCKET)->addr(conn, ip, ip_len, port, remote);
+}
+
+static connection *connCreateUnix(void) {
+    connection *conn = zcalloc(sizeof(connection));
+    conn->type = &CT_Unix;
+    conn->fd = -1;
+
+    return conn;
+}
+
+static connection *connCreateAcceptedUnix(int fd, void *priv) {
+    UNUSED(priv);
+    connection *conn = connCreateUnix();
+    conn->fd = fd;
+    conn->state = CONN_STATE_ACCEPTING;
+    return conn;
+}
+
+static void connUnixAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
+    int cfd, max = MAX_ACCEPTS_PER_CALL;
+    UNUSED(el);
+    UNUSED(mask);
+    UNUSED(privdata);
+
+    while(max--) {
+        cfd = anetUnixAccept(server.neterr, fd);
+        if (cfd == ANET_ERR) {
+            if (errno != EWOULDBLOCK)
+                serverLog(LL_WARNING,
+                    "Accepting client connection: %s", server.neterr);
+            return;
+        }
+        serverLog(LL_VERBOSE,"Accepted connection to %s", server.unixsocket);
+        acceptCommonHandler(connCreateAcceptedUnix(cfd, NULL),CLIENT_UNIX_SOCKET,NULL);
+    }
+}
+
+static void connUnixClose(connection *conn) {
+    connectionByType(CONN_TYPE_SOCKET)->close(conn);
+}
+
+static int connUnixAccept(connection *conn, ConnectionCallbackFunc accept_handler) {
+    return connectionByType(CONN_TYPE_SOCKET)->accept(conn, accept_handler);
+}
+
+static int connUnixWrite(connection *conn, const void *data, size_t data_len) {
+    return connectionByType(CONN_TYPE_SOCKET)->write(conn, data, data_len);
+}
+
+static int connUnixWritev(connection *conn, const struct iovec *iov, int iovcnt) {
+    return connectionByType(CONN_TYPE_SOCKET)->writev(conn, iov, iovcnt);
+}
+
+static int connUnixRead(connection *conn, void *buf, size_t buf_len) {
+    return connectionByType(CONN_TYPE_SOCKET)->read(conn, buf, buf_len);
+}
+
+static int connUnixSetWriteHandler(connection *conn, ConnectionCallbackFunc func, int barrier) {
+    return connectionByType(CONN_TYPE_SOCKET)->set_write_handler(conn, func, barrier);
+}
+
+static int connUnixSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
+    return connectionByType(CONN_TYPE_SOCKET)->set_read_handler(conn, func);
+}
+
+static const char *connUnixGetLastError(connection *conn) {
+    return strerror(conn->last_errno);
+}
+
+static ssize_t connUnixSyncWrite(connection *conn, char *ptr, ssize_t size, long long timeout) {
+    return syncWrite(conn->fd, ptr, size, timeout);
+}
+
+static ssize_t connUnixSyncRead(connection *conn, char *ptr, ssize_t size, long long timeout) {
+    return syncRead(conn->fd, ptr, size, timeout);
+}
+
+static ssize_t connUnixSyncReadLine(connection *conn, char *ptr, ssize_t size, long long timeout) {
+    return syncReadLine(conn->fd, ptr, size, timeout);
+}
+
+static ConnectionType CT_Unix = {
+    /* connection type */
+    .get_type = connUnixGetType,
+
+    /* connection type initialize & finalize & configure */
+    .init = NULL,
+    .cleanup = NULL,
+    .configure = NULL,
+
+    /* ae & accept & listen & error & address handler */
+    .ae_handler = connUnixEventHandler,
+    .accept_handler = connUnixAcceptHandler,
+    .addr = connUnixAddr,
+
+    /* create/close connection */
+    .conn_create = connCreateUnix,
+    .conn_create_accepted = connCreateAcceptedUnix,
+    .close = connUnixClose,
+
+    /* connect & accept */
+    .connect = NULL,
+    .blocking_connect = NULL,
+    .accept = connUnixAccept,
+
+    /* IO */
+    .write = connUnixWrite,
+    .writev = connUnixWritev,
+    .read = connUnixRead,
+    .set_write_handler = connUnixSetWriteHandler,
+    .set_read_handler = connUnixSetReadHandler,
+    .get_last_error = connUnixGetLastError,
+    .sync_write = connUnixSyncWrite,
+    .sync_read = connUnixSyncRead,
+    .sync_readline = connUnixSyncReadLine,
+
+    /* pending data */
+    .has_pending_data = NULL,
+    .process_pending_data = NULL,
+};
+
+int RedisRegisterConnectionTypeUnix()
+{
+    return connTypeRegister(&CT_Unix);
+}