Commit 0b27cfe3 authored by zhenwei pi's avatar zhenwei pi
Browse files

Introduce .listen into connection type



Introduce listen method into connection type, this allows no hard code
of listen logic. Originally, we initialize server during startup like
this:
    if (server.port)
        listenToPort(server.port,&server.ipfd);
    if (server.tls_port)
        listenToPort(server.port,&server.tlsfd);
    if (server.unixsocket)
        anetUnixServer(...server.unixsocket...);

    ...
    if (createSocketAcceptHandler(&server.ipfd, acceptTcpHandler) != C_OK)
    if (createSocketAcceptHandler(&server.tlsfd, acceptTcpHandler) != C_OK)
    if (createSocketAcceptHandler(&server.sofd, acceptTcpHandler) != C_OK)
    ...

If a new connection type gets supported, we have to add more hard code
to setup listener.

Introduce .listen and refactor listener, and Unix socket supports this.
this allows to setup listener arguments and create listener in a loop.

What's more, '.listen' is defined in connection.h, so we should include
server.h to import 'struct socketFds', but server.h has already include
'connection.h'. To avoid including loop(also to make code reasonable),
define 'struct connListener' in connection.h instead of 'struct socketFds'
in server.h. This leads this commit to get more changes.

There are more fields in 'struct connListener', hence it's possible to
simplify changeBindAddr & applyTLSPort() & updatePort() into a single
logic: update the listener config from the server.xxx, and re-create
the listener.

Because of the new field 'priv' in struct connListener, we expect to pass
this to the accept handler(even it's not used currently), this may be used
in the future.
Signed-off-by: default avatarzhenwei pi <pizhenwei@bytedance.com>
parent 45617385
...@@ -678,9 +678,6 @@ void clusterInit(void) { ...@@ -678,9 +678,6 @@ void clusterInit(void) {
} }
if (saveconf) clusterSaveConfigOrDie(1); if (saveconf) clusterSaveConfigOrDie(1);
/* We need a listening TCP port for our cluster messaging needs. */
server.cfd.count = 0;
/* Port sanity check II /* Port sanity check II
* The other handshake port check is triggered too late to stop * The other handshake port check is triggered too late to stop
* us from trying to use a too-high cluster port number. */ * us from trying to use a too-high cluster port number. */
...@@ -696,14 +693,25 @@ void clusterInit(void) { ...@@ -696,14 +693,25 @@ void clusterInit(void) {
serverLog(LL_WARNING, "No bind address is configured, but it is required for the Cluster bus."); serverLog(LL_WARNING, "No bind address is configured, but it is required for the Cluster bus.");
exit(1); exit(1);
} }
int cport = server.cluster_port ? server.cluster_port : port + CLUSTER_PORT_INCR;
if (listenToPort(cport, &server.cfd) == C_ERR ) { if (connectionIndexByType(connTypeOfCluster()->get_type(NULL)) < 0) {
serverLog(LL_WARNING, "Missing connection type %s, but it is required for the Cluster bus.", connTypeOfCluster()->get_type(NULL));
exit(1);
}
connListener *listener = &server.clistener;
listener->count = 0;
listener->bindaddr = server.bindaddr;
listener->bindaddr_count = server.bindaddr_count;
listener->port = server.cluster_port ? server.cluster_port : port + CLUSTER_PORT_INCR;
listener->ct = connTypeOfCluster();
if (connListen(listener) == C_ERR ) {
/* Note: the following log text is matched by the test suite. */ /* Note: the following log text is matched by the test suite. */
serverLog(LL_WARNING, "Failed listening on port %u (cluster), aborting.", cport); serverLog(LL_WARNING, "Failed listening on port %u (cluster), aborting.", listener->port);
exit(1); exit(1);
} }
if (createSocketAcceptHandler(&server.cfd, clusterAcceptHandler) != C_OK) { if (createSocketAcceptHandler(&server.clistener, clusterAcceptHandler) != C_OK) {
serverPanic("Unrecoverable error creating Redis Cluster socket accept handler."); serverPanic("Unrecoverable error creating Redis Cluster socket accept handler.");
} }
......
...@@ -2430,7 +2430,14 @@ static int updateHZ(const char **err) { ...@@ -2430,7 +2430,14 @@ static int updateHZ(const char **err) {
} }
static int updatePort(const char **err) { static int updatePort(const char **err) {
if (changeListenPort(server.port, &server.ipfd, connAcceptHandler(connectionTypeTcp())) == C_ERR) { connListener *listener = listenerByType(CONN_TYPE_SOCKET);
serverAssert(listener != NULL);
listener->bindaddr = server.bindaddr;
listener->bindaddr_count = server.bindaddr_count;
listener->port = server.port;
listener->ct = connectionByType(CONN_TYPE_SOCKET);
if (changeListener(listener) == C_ERR) {
*err = "Unable to listen on this port. Check server logs."; *err = "Unable to listen on this port. Check server logs.";
return 0; return 0;
} }
...@@ -2545,12 +2552,36 @@ int updateRequirePass(const char **err) { ...@@ -2545,12 +2552,36 @@ int updateRequirePass(const char **err) {
return 1; return 1;
} }
/* applyBind affects both TCP and TLS (if enabled) together */
static int applyBind(const char **err) { static int applyBind(const char **err) {
if (changeBindAddr() == C_ERR) { connListener *tcp_listener = listenerByType(CONN_TYPE_SOCKET);
connListener *tls_listener = listenerByType(CONN_TYPE_TLS);
serverAssert(tcp_listener != NULL);
tcp_listener->bindaddr = server.bindaddr;
tcp_listener->bindaddr_count = server.bindaddr_count;
tcp_listener->port = server.port;
tcp_listener->ct = connectionByType(CONN_TYPE_SOCKET);
if (changeListener(tcp_listener) == C_ERR) {
*err = "Failed to bind to specified addresses."; *err = "Failed to bind to specified addresses.";
if (tls_listener)
closeListener(tls_listener); /* failed with TLS together */
return 0; return 0;
} }
if (server.tls_port != 0) {
serverAssert(tls_listener != NULL);
tls_listener->bindaddr = server.bindaddr;
tls_listener->bindaddr_count = server.bindaddr_count;
tls_listener->port = server.tls_port;
tls_listener->ct = connectionByType(CONN_TYPE_TLS);
if (changeListener(tls_listener) == C_ERR) {
*err = "Failed to bind to specified addresses.";
closeListener(tcp_listener); /* failed with TCP together */
return 0;
}
}
return 1; return 1;
} }
...@@ -2591,7 +2622,13 @@ static int applyTLSPort(const char **err) { ...@@ -2591,7 +2622,13 @@ static int applyTLSPort(const char **err) {
return 0; return 0;
} }
if (changeListenPort(server.tls_port, &server.tlsfd, connAcceptHandler(connectionTypeTls())) == C_ERR) { connListener *listener = listenerByType(CONN_TYPE_TLS);
serverAssert(listener != NULL);
listener->bindaddr = server.bindaddr;
listener->bindaddr_count = server.bindaddr_count;
listener->port = server.tls_port;
listener->ct = connectionByType(CONN_TYPE_TLS);
if (changeListener(listener) == C_ERR) {
*err = "Unable to listen on this port. Check server logs."; *err = "Unable to listen on this port. Check server logs.";
return 0; return 0;
} }
......
...@@ -44,6 +44,7 @@ ...@@ -44,6 +44,7 @@
struct aeEventLoop; struct aeEventLoop;
typedef struct connection connection; typedef struct connection connection;
typedef struct connListener connListener;
typedef enum { typedef enum {
CONN_STATE_NONE = 0, CONN_STATE_NONE = 0,
...@@ -77,6 +78,7 @@ typedef struct ConnectionType { ...@@ -77,6 +78,7 @@ typedef struct ConnectionType {
void (*ae_handler)(struct aeEventLoop *el, int fd, void *clientData, int mask); void (*ae_handler)(struct aeEventLoop *el, int fd, void *clientData, int mask);
aeFileProc *accept_handler; aeFileProc *accept_handler;
int (*addr)(connection *conn, char *ip, size_t ip_len, int *port, int remote); int (*addr)(connection *conn, char *ip, size_t ip_len, int *port, int remote);
int (*listen)(connListener *listener);
/* create/close connection */ /* create/close connection */
connection* (*conn_create)(void); connection* (*conn_create)(void);
...@@ -122,6 +124,19 @@ struct connection { ...@@ -122,6 +124,19 @@ struct connection {
int fd; int fd;
}; };
#define CONFIG_BINDADDR_MAX 16
/* Setup a listener by a connection type */
struct connListener {
int fd[CONFIG_BINDADDR_MAX];
int count;
char **bindaddr;
int bindaddr_count;
int port;
ConnectionType *ct;
void *priv; /* used by connection type specified data */
};
/* The connection module does not deal with listening and accepting sockets, /* The connection module does not deal with listening and accepting sockets,
* so we assume we have a socket when an incoming connection is created. * so we assume we have a socket when an incoming connection is created.
* *
...@@ -406,6 +421,11 @@ int connTypeHasPendingData(void); ...@@ -406,6 +421,11 @@ int connTypeHasPendingData(void);
/* walk all the connection types and process pending data for each connection type */ /* walk all the connection types and process pending data for each connection type */
int connTypeProcessPendingData(void); int connTypeProcessPendingData(void);
/* Listen on an initialized listener */
static inline int connListen(connListener *listener) {
return listener->ct->listen(listener);
}
/* Get accept_handler of a connection type */ /* Get accept_handler of a connection type */
static inline aeFileProc *connAcceptHandler(ConnectionType *ct) { static inline aeFileProc *connAcceptHandler(ConnectionType *ct) {
if (ct) if (ct)
......
...@@ -1862,9 +1862,7 @@ void initServerConfig(void) { ...@@ -1862,9 +1862,7 @@ void initServerConfig(void) {
server.bindaddr_count = CONFIG_DEFAULT_BINDADDR_COUNT; server.bindaddr_count = CONFIG_DEFAULT_BINDADDR_COUNT;
for (j = 0; j < CONFIG_DEFAULT_BINDADDR_COUNT; j++) for (j = 0; j < CONFIG_DEFAULT_BINDADDR_COUNT; j++)
server.bindaddr[j] = zstrdup(default_bindaddr[j]); server.bindaddr[j] = zstrdup(default_bindaddr[j]);
server.ipfd.count = 0; memset(server.listeners, 0x00, sizeof(server.listeners));
server.tlsfd.count = 0;
server.sofd.count = 0;
server.active_expire_enabled = 1; server.active_expire_enabled = 1;
server.skip_checksum_validation = 0; server.skip_checksum_validation = 0;
server.loading = 0; server.loading = 0;
...@@ -2232,7 +2230,7 @@ void checkTcpBacklogSettings(void) { ...@@ -2232,7 +2230,7 @@ void checkTcpBacklogSettings(void) {
#endif #endif
} }
void closeSocketListeners(socketFds *sfd) { void closeListener(connListener *sfd) {
int j; int j;
for (j = 0; j < sfd->count; j++) { for (j = 0; j < sfd->count; j++) {
...@@ -2247,11 +2245,11 @@ void closeSocketListeners(socketFds *sfd) { ...@@ -2247,11 +2245,11 @@ void closeSocketListeners(socketFds *sfd) {
/* Create an event handler for accepting new connections in TCP or TLS domain sockets. /* Create an event handler for accepting new connections in TCP or TLS domain sockets.
* This works atomically for all socket fds */ * This works atomically for all socket fds */
int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) { int createSocketAcceptHandler(connListener *sfd, aeFileProc *accept_handler) {
int j; int j;
for (j = 0; j < sfd->count; j++) { for (j = 0; j < sfd->count; j++) {
if (aeCreateFileEvent(server.el, sfd->fd[j], AE_READABLE, accept_handler,NULL) == AE_ERR) { if (aeCreateFileEvent(server.el, sfd->fd[j], AE_READABLE, accept_handler,sfd) == AE_ERR) {
/* Rollback */ /* Rollback */
for (j = j-1; j >= 0; j--) aeDeleteFileEvent(server.el, sfd->fd[j], AE_READABLE); for (j = j-1; j >= 0; j--) aeDeleteFileEvent(server.el, sfd->fd[j], AE_READABLE);
return C_ERR; return C_ERR;
...@@ -2264,7 +2262,8 @@ int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) { ...@@ -2264,7 +2262,8 @@ int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) {
* binding the addresses specified in the Redis server configuration. * binding the addresses specified in the Redis server configuration.
* *
* The listening file descriptors are stored in the integer array 'fds' * The listening file descriptors are stored in the integer array 'fds'
* and their number is set in '*count'. * and their number is set in '*count'. Actually @sfd should be 'listener',
* for the historical reasons, let's keep 'sfd' here.
* *
* The addresses to bind are specified in the global server.bindaddr array * The addresses to bind are specified in the global server.bindaddr array
* and their number is server.bindaddr_count. If the server configuration * and their number is server.bindaddr_count. If the server configuration
...@@ -2278,14 +2277,15 @@ int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) { ...@@ -2278,14 +2277,15 @@ int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler) {
* impossible to bind, or no bind addresses were specified in the server * impossible to bind, or no bind addresses were specified in the server
* configuration but the function is not able to bind * for at least * configuration but the function is not able to bind * for at least
* one of the IPv4 or IPv6 protocols. */ * one of the IPv4 or IPv6 protocols. */
int listenToPort(int port, socketFds *sfd) { int listenToPort(connListener *sfd) {
int j; int j;
char **bindaddr = server.bindaddr; int port = sfd->port;
char **bindaddr = sfd->bindaddr;
/* If we have no bind address, we don't listen on a TCP socket */ /* If we have no bind address, we don't listen on a TCP socket */
if (server.bindaddr_count == 0) return C_OK; if (sfd->bindaddr_count == 0) return C_OK;
for (j = 0; j < server.bindaddr_count; j++) { for (j = 0; j < sfd->bindaddr_count; j++) {
char* addr = bindaddr[j]; char* addr = bindaddr[j];
int optional = *addr == '-'; int optional = *addr == '-';
if (optional) addr++; if (optional) addr++;
...@@ -2309,7 +2309,7 @@ int listenToPort(int port, socketFds *sfd) { ...@@ -2309,7 +2309,7 @@ int listenToPort(int port, socketFds *sfd) {
continue; continue;
/* Rollback successful listens before exiting */ /* Rollback successful listens before exiting */
closeSocketListeners(sfd); closeListener(sfd);
return C_ERR; return C_ERR;
} }
if (server.socket_mark_id > 0) anetSetSockMarkId(NULL, sfd->fd[sfd->count], server.socket_mark_id); if (server.socket_mark_id > 0) anetSetSockMarkId(NULL, sfd->fd[sfd->count], server.socket_mark_id);
...@@ -2470,38 +2470,38 @@ void initServer(void) { ...@@ -2470,38 +2470,38 @@ void initServer(void) {
} }
server.db = zmalloc(sizeof(redisDb)*server.dbnum); server.db = zmalloc(sizeof(redisDb)*server.dbnum);
/* Open the TCP listening socket for the user commands. */ /* Setup listeners from server config for TCP/TLS/Unix */
if (server.port != 0 && int conn_index;
listenToPort(server.port,&server.ipfd) == C_ERR) { connListener *listener;
/* Note: the following log text is matched by the test suite. */ if (server.port != 0) {
serverLog(LL_WARNING, "Failed listening on port %u (TCP), aborting.", server.port); conn_index = connectionIndexByType(CONN_TYPE_SOCKET);
exit(1); if (conn_index < 0)
} serverPanic("Failed finding connection listener of %s", CONN_TYPE_SOCKET);
if (server.tls_port != 0 && listener = &server.listeners[conn_index];
listenToPort(server.tls_port,&server.tlsfd) == C_ERR) { listener->bindaddr = server.bindaddr;
/* Note: the following log text is matched by the test suite. */ listener->bindaddr_count = server.bindaddr_count;
serverLog(LL_WARNING, "Failed listening on port %u (TLS), aborting.", server.tls_port); listener->port = server.port;
exit(1); listener->ct = connectionByType(CONN_TYPE_SOCKET);
}
if (server.tls_port != 0) {
conn_index = connectionIndexByType(CONN_TYPE_TLS);
if (conn_index < 0)
serverPanic("Failed finding connection listener of %s", CONN_TYPE_TLS);
listener = &server.listeners[conn_index];
listener->bindaddr = server.bindaddr;
listener->bindaddr_count = server.bindaddr_count;
listener->port = server.tls_port;
listener->ct = connectionByType(CONN_TYPE_TLS);
} }
/* Open the listening Unix domain socket. */
if (server.unixsocket != NULL) { if (server.unixsocket != NULL) {
unlink(server.unixsocket); /* don't care if this fails */ conn_index = connectionIndexByType(CONN_TYPE_UNIX);
server.sofd.fd[0] = anetUnixServer(server.neterr,server.unixsocket, if (conn_index < 0)
(mode_t)server.unixsocketperm, server.tcp_backlog); serverPanic("Failed finding connection listener of %s", CONN_TYPE_UNIX);
if (server.sofd.fd[0] == ANET_ERR) { listener = &server.listeners[conn_index];
serverLog(LL_WARNING, "Failed opening Unix socket: %s", server.neterr); listener->bindaddr = &server.unixsocket;
exit(1); listener->bindaddr_count = 1;
} listener->ct = connectionByType(CONN_TYPE_UNIX);
anetNonBlock(NULL,server.sofd.fd[0]); listener->priv = &server.unixsocketperm; /* Unix socket specified */
anetCloexec(server.sofd.fd[0]);
server.sofd.count = 1;
}
/* Abort if there are no listening sockets at all. */
if (server.ipfd.count == 0 && server.tlsfd.count == 0 && server.sofd.count == 0) {
serverLog(LL_WARNING, "Configured to not listen anywhere, exiting.");
exit(1);
} }
/* Create the Redis databases, and initialize other internal state. */ /* Create the Redis databases, and initialize other internal state. */
...@@ -2584,18 +2584,28 @@ void initServer(void) { ...@@ -2584,18 +2584,28 @@ void initServer(void) {
exit(1); exit(1);
} }
/* Create an event handler for accepting new connections in TCP and Unix /* create all the configured listener, and add handler to start to accept */
* domain sockets. */ int listen_fds = 0;
if (createSocketAcceptHandler(&server.ipfd, connAcceptHandler(connectionTypeTcp())) != C_OK) { for (j = 0; j < CONN_TYPE_MAX; j++) {
serverPanic("Unrecoverable error creating TCP socket accept handler."); listener = &server.listeners[j];
} if (listener->ct == NULL)
if (createSocketAcceptHandler(&server.tlsfd, connAcceptHandler(connectionTypeTls())) != C_OK) { continue;
serverPanic("Unrecoverable error creating TLS socket accept handler.");
} if (connListen(listener) == C_ERR) {
if (createSocketAcceptHandler(&server.sofd, connAcceptHandler(connectionTypeUnix())) != C_OK) { serverLog(LL_WARNING, "Failed listening on port %u (%s), aborting.", listener->port, listener->ct->get_type(NULL));
serverPanic("Unrecoverable error creating server.sofd file event."); exit(1);
}
if (createSocketAcceptHandler(listener, connAcceptHandler(listener->ct)) != C_OK)
serverPanic("Unrecoverable error creating %s listener accept handler.", listener->ct->get_type(NULL));
listen_fds += listener->count;
} }
if (listen_fds == 0) {
serverLog(LL_WARNING, "Configured to not listen anywhere, exiting.");
exit(1);
}
/* Register a readable event for the pipe used to awake the event loop /* Register a readable event for the pipe used to awake the event loop
* from module threads. */ * from module threads. */
...@@ -4067,11 +4077,16 @@ void incrementErrorCount(const char *fullerr, size_t namelen) { ...@@ -4067,11 +4077,16 @@ void incrementErrorCount(const char *fullerr, size_t namelen) {
void closeListeningSockets(int unlink_unix_socket) { void closeListeningSockets(int unlink_unix_socket) {
int j; int j;
for (j = 0; j < server.ipfd.count; j++) close(server.ipfd.fd[j]); for (int i = 0; i < CONN_TYPE_MAX; i++) {
for (j = 0; j < server.tlsfd.count; j++) close(server.tlsfd.fd[j]); connListener *listener = &server.listeners[i];
for (j = 0; j < server.sofd.count; j++) close(server.sofd.fd[j]); if (listener->ct == NULL)
continue;
for (j = 0; j < listener->count; j++) close(listener->fd[j]);
}
if (server.cluster_enabled) if (server.cluster_enabled)
for (j = 0; j < server.cfd.count; j++) close(server.cfd.fd[j]); for (j = 0; j < server.clistener.count; j++) close(server.clistener.fd[j]);
if (unlink_unix_socket && server.unixsocket) { if (unlink_unix_socket && server.unixsocket) {
serverLog(LL_NOTICE,"Removing the unix socket file."); serverLog(LL_NOTICE,"Removing the unix socket file.");
if (unlink(server.unixsocket) != 0) if (unlink(server.unixsocket) != 0)
...@@ -6266,61 +6281,38 @@ void redisAsciiArt(void) { ...@@ -6266,61 +6281,38 @@ void redisAsciiArt(void) {
zfree(buf); zfree(buf);
} }
int changeBindAddr(void) { /* Get the server listener by type name */
/* Close old TCP and TLS servers */ connListener *listenerByType(const char *typename) {
closeSocketListeners(&server.ipfd); int conn_index;
closeSocketListeners(&server.tlsfd);
/* Bind to the new port */ conn_index = connectionIndexByType(typename);
if ((server.port != 0 && listenToPort(server.port, &server.ipfd) != C_OK) || if (conn_index < 0)
(server.tls_port != 0 && listenToPort(server.tls_port, &server.tlsfd) != C_OK)) { return NULL;
serverLog(LL_WARNING, "Failed to bind");
closeSocketListeners(&server.ipfd);
closeSocketListeners(&server.tlsfd);
return C_ERR;
}
/* Create TCP and TLS event handlers */
if (createSocketAcceptHandler(&server.ipfd, connAcceptHandler(connectionTypeTcp())) != C_OK) {
serverPanic("Unrecoverable error creating TCP socket accept handler.");
}
if (createSocketAcceptHandler(&server.tlsfd, connAcceptHandler(connectionTypeTls())) != C_OK) {
serverPanic("Unrecoverable error creating TLS socket accept handler.");
}
if (server.set_proc_title) redisSetProcTitle(NULL);
return C_OK; return &server.listeners[conn_index];
} }
int changeListenPort(int port, socketFds *sfd, aeFileProc *accept_handler) { /* Close original listener, re-create a new listener from the updated bind address & port */
socketFds new_sfd = {{0}}; int changeListener(connListener *listener) {
/* Close old servers */ /* Close old servers */
closeSocketListeners(sfd); closeListener(listener);
/* Just close the server if port disabled */ /* Just close the server if port disabled */
if (port == 0) { if (listener->port == 0) {
if (server.set_proc_title) redisSetProcTitle(NULL); if (server.set_proc_title) redisSetProcTitle(NULL);
return C_OK; return C_OK;
} }
/* Bind to the new port */ /* Re-create listener */
if (listenToPort(port, &new_sfd) != C_OK) { if (connListen(listener) != C_OK) {
return C_ERR; return C_ERR;
} }
/* Create event handlers */ /* Create event handlers */
if (createSocketAcceptHandler(&new_sfd, accept_handler) != C_OK) { if (createSocketAcceptHandler(listener, listener->ct->accept_handler) != C_OK) {
closeSocketListeners(&new_sfd); serverPanic("Unrecoverable error creating %s accept handler.", listener->ct->get_type(NULL));
return C_ERR;
} }
/* Copy new descriptors */
sfd->count = new_sfd.count;
memcpy(sfd->fd, new_sfd.fd, sizeof(new_sfd.fd));
if (server.set_proc_title) redisSetProcTitle(NULL); if (server.set_proc_title) redisSetProcTitle(NULL);
return C_OK; return C_OK;
...@@ -7135,10 +7127,15 @@ int main(int argc, char **argv) { ...@@ -7135,10 +7127,15 @@ int main(int argc, char **argv) {
exit(1); exit(1);
} }
} }
if (server.ipfd.count > 0 || server.tlsfd.count > 0)
serverLog(LL_NOTICE,"Ready to accept connections"); for (j = 0; j < CONN_TYPE_MAX; j++) {
if (server.sofd.count > 0) connListener *listener = &server.listeners[j];
serverLog(LL_NOTICE,"The server is now ready to accept connections at %s", server.unixsocket); if (listener->ct == NULL)
continue;
serverLog(LL_NOTICE,"Ready to accept connections %s", listener->ct->get_type(NULL));
}
if (server.supervised_mode == SUPERVISED_SYSTEMD) { if (server.supervised_mode == SUPERVISED_SYSTEMD) {
if (!server.masterhost) { if (!server.masterhost) {
redisCommunicateSystemd("STATUS=Ready to accept connections\n"); redisCommunicateSystemd("STATUS=Ready to accept connections\n");
......
...@@ -1376,11 +1376,6 @@ struct malloc_stats { ...@@ -1376,11 +1376,6 @@ struct malloc_stats {
size_t allocator_resident; size_t allocator_resident;
}; };
typedef struct socketFds {
int fd[CONFIG_BINDADDR_MAX];
int count;
} socketFds;
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
* TLS Context Configuration * TLS Context Configuration
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
...@@ -1514,11 +1509,9 @@ struct redisServer { ...@@ -1514,11 +1509,9 @@ struct redisServer {
char *bind_source_addr; /* Source address to bind on for outgoing connections */ char *bind_source_addr; /* Source address to bind on for outgoing connections */
char *unixsocket; /* UNIX socket path */ char *unixsocket; /* UNIX socket path */
unsigned int unixsocketperm; /* UNIX socket permission (see mode_t) */ unsigned int unixsocketperm; /* UNIX socket permission (see mode_t) */
socketFds ipfd; /* TCP socket file descriptors */ connListener listeners[CONN_TYPE_MAX]; /* TCP/Unix/TLS even more types */
socketFds tlsfd; /* TLS socket file descriptors */
socketFds sofd; /* Unix socket file descriptor */
uint32_t socket_mark_id; /* ID for listen socket marking */ uint32_t socket_mark_id; /* ID for listen socket marking */
socketFds cfd; /* Cluster bus listening socket */ connListener clistener; /* Cluster bus listener */
list *clients; /* List of active clients */ list *clients; /* List of active clients */
list *clients_to_close; /* Clients to close asynchronously */ list *clients_to_close; /* Clients to close asynchronously */
list *clients_pending_write; /* There is to write or install handler. */ list *clients_pending_write; /* There is to write or install handler. */
...@@ -2525,7 +2518,7 @@ char *getClientTypeName(int class); ...@@ -2525,7 +2518,7 @@ char *getClientTypeName(int class);
void flushSlavesOutputBuffers(void); void flushSlavesOutputBuffers(void);
void disconnectSlaves(void); void disconnectSlaves(void);
void evictClients(void); void evictClients(void);
int listenToPort(int port, socketFds *fds); int listenToPort(connListener *fds);
void pauseClients(pause_purpose purpose, mstime_t end, pause_type type); void pauseClients(pause_purpose purpose, mstime_t end, pause_type type);
void unpauseClients(pause_purpose purpose); void unpauseClients(pause_purpose purpose);
int areClientsPaused(void); int areClientsPaused(void);
...@@ -2891,9 +2884,10 @@ int processCommand(client *c); ...@@ -2891,9 +2884,10 @@ int processCommand(client *c);
int processPendingCommandAndInputBuffer(client *c); int processPendingCommandAndInputBuffer(client *c);
void setupSignalHandlers(void); void setupSignalHandlers(void);
void removeSignalHandlers(void); void removeSignalHandlers(void);
int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler); int createSocketAcceptHandler(connListener *sfd, aeFileProc *accept_handler);
int changeListenPort(int port, socketFds *sfd, aeFileProc *accept_handler); connListener *listenerByType(const char *typename);
int changeBindAddr(void); int changeListener(connListener *listener);
void closeListener(connListener *listener);
struct redisCommand *lookupSubcommand(struct redisCommand *container, sds sub_name); struct redisCommand *lookupSubcommand(struct redisCommand *container, sds sub_name);
struct redisCommand *lookupCommand(robj **argv, int argc); struct redisCommand *lookupCommand(robj **argv, int argc);
struct redisCommand *lookupCommandBySdsLogic(dict *commands, sds s); struct redisCommand *lookupCommandBySdsLogic(dict *commands, sds s);
......
...@@ -329,6 +329,10 @@ static int connSocketAddr(connection *conn, char *ip, size_t ip_len, int *port, ...@@ -329,6 +329,10 @@ static int connSocketAddr(connection *conn, char *ip, size_t ip_len, int *port,
return C_ERR; return C_ERR;
} }
static int connSocketListen(connListener *listener) {
return listenToPort(listener);
}
static int connSocketBlockingConnect(connection *conn, const char *addr, int port, long long timeout) { static int connSocketBlockingConnect(connection *conn, const char *addr, int port, long long timeout) {
int fd = anetTcpNonBlockConnect(NULL,addr,port); int fd = anetTcpNonBlockConnect(NULL,addr,port);
if (fd == -1) { if (fd == -1) {
...@@ -382,6 +386,7 @@ static ConnectionType CT_Socket = { ...@@ -382,6 +386,7 @@ static ConnectionType CT_Socket = {
.ae_handler = connSocketEventHandler, .ae_handler = connSocketEventHandler,
.accept_handler = connSocketAcceptHandler, .accept_handler = connSocketAcceptHandler,
.addr = connSocketAddr, .addr = connSocketAddr,
.listen = connSocketListen,
/* create/close connection */ /* create/close connection */
.conn_create = connCreateSocket, .conn_create = connCreateSocket,
......
...@@ -743,6 +743,10 @@ static int connTLSAddr(connection *conn, char *ip, size_t ip_len, int *port, int ...@@ -743,6 +743,10 @@ static int connTLSAddr(connection *conn, char *ip, size_t ip_len, int *port, int
return anetFdToString(conn->fd, ip, ip_len, port, remote); return anetFdToString(conn->fd, ip, ip_len, port, remote);
} }
static int connTLSListen(connListener *listener) {
return listenToPort(listener);
}
static void connTLSClose(connection *conn_) { static void connTLSClose(connection *conn_) {
tls_connection *conn = (tls_connection *) conn_; tls_connection *conn = (tls_connection *) conn_;
...@@ -1104,6 +1108,7 @@ static ConnectionType CT_TLS = { ...@@ -1104,6 +1108,7 @@ static ConnectionType CT_TLS = {
.ae_handler = tlsEventHandler, .ae_handler = tlsEventHandler,
.accept_handler = tlsAcceptHandler, .accept_handler = tlsAcceptHandler,
.addr = connTLSAddr, .addr = connTLSAddr,
.listen = connTLSListen,
/* create/close connection */ /* create/close connection */
.conn_create = connCreateTLS, .conn_create = connCreateTLS,
......
...@@ -43,6 +43,31 @@ static int connUnixAddr(connection *conn, char *ip, size_t ip_len, int *port, in ...@@ -43,6 +43,31 @@ static int connUnixAddr(connection *conn, char *ip, size_t ip_len, int *port, in
return connectionTypeTcp()->addr(conn, ip, ip_len, port, remote); return connectionTypeTcp()->addr(conn, ip, ip_len, port, remote);
} }
static int connUnixListen(connListener *listener) {
int fd;
mode_t *perm = (mode_t *)listener->priv;
if (listener->bindaddr_count == 0)
return C_OK;
/* currently listener->bindaddr_count is always 1, we still use a loop here in case Redis supports multi Unix socket in the future */
for (int j = 0; j < listener->bindaddr_count; j++) {
char *addr = listener->bindaddr[j];
unlink(addr); /* don't care if this fails */
fd = anetUnixServer(server.neterr, addr, *perm, server.tcp_backlog);
if (fd == ANET_ERR) {
serverLog(LL_WARNING, "Failed opening Unix socket: %s", server.neterr);
exit(1);
}
anetNonBlock(NULL, fd);
anetCloexec(fd);
listener->fd[listener->count++] = fd;
}
return C_OK;
}
static connection *connCreateUnix(void) { static connection *connCreateUnix(void) {
connection *conn = zcalloc(sizeof(connection)); connection *conn = zcalloc(sizeof(connection));
conn->type = &CT_Unix; conn->type = &CT_Unix;
...@@ -135,6 +160,7 @@ static ConnectionType CT_Unix = { ...@@ -135,6 +160,7 @@ static ConnectionType CT_Unix = {
.ae_handler = connUnixEventHandler, .ae_handler = connUnixEventHandler,
.accept_handler = connUnixAcceptHandler, .accept_handler = connUnixAcceptHandler,
.addr = connUnixAddr, .addr = connUnixAddr,
.listen = connUnixListen,
/* create/close connection */ /* create/close connection */
.conn_create = connCreateUnix, .conn_create = connCreateUnix,
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment