diff options
| -rw-r--r-- | src/Makefile | 2 | ||||
| -rw-r--r-- | src/connection.c | 3 | ||||
| -rw-r--r-- | src/connection.h | 6 | ||||
| -rw-r--r-- | src/networking.c | 19 | ||||
| -rw-r--r-- | src/server.c | 2 | ||||
| -rw-r--r-- | src/server.h | 1 | ||||
| -rw-r--r-- | src/unix.c | 168 |
7 files changed, 177 insertions, 24 deletions
diff --git a/src/Makefile b/src/Makefile index f5736514e..814f3c0aa 100644 --- a/src/Makefile +++ b/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) diff --git a/src/connection.c b/src/connection.c index 72db82212..33e0c03c1 100644 --- a/src/connection.c +++ b/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(); diff --git a/src/connection.h b/src/connection.h index feb084629..d54ed296d 100644 --- a/src/connection.h +++ b/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 */ diff --git a/src/networking.c b/src/networking.c index d71e30b71..d6494fc09 100644 --- a/src/networking.c +++ b/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; diff --git a/src/server.c b/src/server.c index 795010473..5f6810c51 100644 --- a/src/server.c +++ b/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."); } diff --git a/src/server.h b/src/server.h index ad91d5218..e6088a8b4 100644 --- a/src/server.h +++ b/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); diff --git a/src/unix.c b/src/unix.c new file mode 100644 index 000000000..924a9ddd2 --- /dev/null +++ b/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); +} |