Commit b087dd1d authored by Yossi Gottlieb's avatar Yossi Gottlieb
Browse files

TLS: Connections refactoring and TLS support.

* Introduce a connection abstraction layer for all socket operations and
integrate it across the code base.
* Provide an optional TLS connections implementation based on OpenSSL.
* Pull a newer version of hiredis with TLS support.
* Tests, redis-cli updates for TLS support.
parent f4d37173
...@@ -49,7 +49,7 @@ clusterNode *myself = NULL; ...@@ -49,7 +49,7 @@ clusterNode *myself = NULL;
clusterNode *createClusterNode(char *nodename, int flags); clusterNode *createClusterNode(char *nodename, int flags);
int clusterAddNode(clusterNode *node); int clusterAddNode(clusterNode *node);
void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask); void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask); void clusterReadHandler(connection *conn);
void clusterSendPing(clusterLink *link, int type); void clusterSendPing(clusterLink *link, int type);
void clusterSendFail(char *nodename); void clusterSendFail(char *nodename);
void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request); void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request);
...@@ -477,7 +477,8 @@ void clusterInit(void) { ...@@ -477,7 +477,8 @@ void clusterInit(void) {
/* 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. */
if (server.port > (65535-CLUSTER_PORT_INCR)) { int port = server.tls_cluster ? server.tls_port : server.port;
if (port > (65535-CLUSTER_PORT_INCR)) {
serverLog(LL_WARNING, "Redis port number too high. " serverLog(LL_WARNING, "Redis port number too high. "
"Cluster communication port is 10,000 port " "Cluster communication port is 10,000 port "
"numbers higher than your Redis port. " "numbers higher than your Redis port. "
...@@ -485,8 +486,7 @@ void clusterInit(void) { ...@@ -485,8 +486,7 @@ void clusterInit(void) {
"lower than 55535."); "lower than 55535.");
exit(1); exit(1);
} }
if (listenToPort(port+CLUSTER_PORT_INCR,
if (listenToPort(server.port+CLUSTER_PORT_INCR,
server.cfd,&server.cfd_count) == C_ERR) server.cfd,&server.cfd_count) == C_ERR)
{ {
exit(1); exit(1);
...@@ -508,8 +508,8 @@ void clusterInit(void) { ...@@ -508,8 +508,8 @@ void clusterInit(void) {
/* Set myself->port / cport to my listening ports, we'll just need to /* Set myself->port / cport to my listening ports, we'll just need to
* discover the IP address via MEET messages. */ * discover the IP address via MEET messages. */
myself->port = server.port; myself->port = port;
myself->cport = server.port+CLUSTER_PORT_INCR; myself->cport = port+CLUSTER_PORT_INCR;
if (server.cluster_announce_port) if (server.cluster_announce_port)
myself->port = server.cluster_announce_port; myself->port = server.cluster_announce_port;
if (server.cluster_announce_bus_port) if (server.cluster_announce_bus_port)
...@@ -593,7 +593,7 @@ clusterLink *createClusterLink(clusterNode *node) { ...@@ -593,7 +593,7 @@ clusterLink *createClusterLink(clusterNode *node) {
link->sndbuf = sdsempty(); link->sndbuf = sdsempty();
link->rcvbuf = sdsempty(); link->rcvbuf = sdsempty();
link->node = node; link->node = node;
link->fd = -1; link->conn = NULL;
return link; return link;
} }
...@@ -601,23 +601,45 @@ clusterLink *createClusterLink(clusterNode *node) { ...@@ -601,23 +601,45 @@ clusterLink *createClusterLink(clusterNode *node) {
* This function will just make sure that the original node associated * This function will just make sure that the original node associated
* with this link will have the 'link' field set to NULL. */ * with this link will have the 'link' field set to NULL. */
void freeClusterLink(clusterLink *link) { void freeClusterLink(clusterLink *link) {
if (link->fd != -1) { if (link->conn) {
aeDeleteFileEvent(server.el, link->fd, AE_READABLE|AE_WRITABLE); connClose(link->conn);
link->conn = NULL;
} }
sdsfree(link->sndbuf); sdsfree(link->sndbuf);
sdsfree(link->rcvbuf); sdsfree(link->rcvbuf);
if (link->node) if (link->node)
link->node->link = NULL; link->node->link = NULL;
close(link->fd);
zfree(link); zfree(link);
} }
static void clusterConnAcceptHandler(connection *conn) {
clusterLink *link;
if (connGetState(conn) != CONN_STATE_CONNECTED) {
serverLog(LL_VERBOSE,
"Error accepting cluster node connection: %s", connGetLastError(conn));
connClose(conn);
return;
}
/* Create a link object we use to handle the connection.
* It gets passed to the readable handler when data is available.
* Initiallly the link->node pointer is set to NULL as we don't know
* which node is, but the right node is references once we know the
* node identity. */
link = createClusterLink(NULL);
link->conn = conn;
connSetPrivateData(conn, link);
/* Register read handler */
connSetReadHandler(conn, clusterReadHandler);
}
#define MAX_CLUSTER_ACCEPTS_PER_CALL 1000 #define MAX_CLUSTER_ACCEPTS_PER_CALL 1000
void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) { void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
int cport, cfd; int cport, cfd;
int max = MAX_CLUSTER_ACCEPTS_PER_CALL; int max = MAX_CLUSTER_ACCEPTS_PER_CALL;
char cip[NET_IP_STR_LEN]; char cip[NET_IP_STR_LEN];
clusterLink *link;
UNUSED(el); UNUSED(el);
UNUSED(mask); UNUSED(mask);
UNUSED(privdata); UNUSED(privdata);
...@@ -634,19 +656,24 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -634,19 +656,24 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
"Error accepting cluster node: %s", server.neterr); "Error accepting cluster node: %s", server.neterr);
return; return;
} }
anetNonBlock(NULL,cfd);
anetEnableTcpNoDelay(NULL,cfd); connection *conn = server.tls_cluster ? connCreateAcceptedTLS(cfd,1) : connCreateAcceptedSocket(cfd);
connNonBlock(conn);
connEnableTcpNoDelay(conn);
/* Use non-blocking I/O for cluster messages. */ /* Use non-blocking I/O for cluster messages. */
serverLog(LL_VERBOSE,"Accepted cluster node %s:%d", cip, cport); serverLog(LL_VERBOSE,"Accepting cluster node connection from %s:%d", cip, cport);
/* Create a link object we use to handle the connection.
* It gets passed to the readable handler when data is available. /* Accept the connection now. connAccept() may call our handler directly
* Initiallly the link->node pointer is set to NULL as we don't know * or schedule it for later depending on connection implementation.
* which node is, but the right node is references once we know the */
* node identity. */ if (connAccept(conn, clusterConnAcceptHandler) == C_ERR) {
link = createClusterLink(NULL); serverLog(LL_VERBOSE,
link->fd = cfd; "Error accepting cluster node connection: %s",
aeCreateFileEvent(server.el,cfd,AE_READABLE,clusterReadHandler,link); connGetLastError(conn));
connClose(conn);
return;
}
} }
} }
...@@ -1447,7 +1474,7 @@ void nodeIp2String(char *buf, clusterLink *link, char *announced_ip) { ...@@ -1447,7 +1474,7 @@ void nodeIp2String(char *buf, clusterLink *link, char *announced_ip) {
memcpy(buf,announced_ip,NET_IP_STR_LEN); memcpy(buf,announced_ip,NET_IP_STR_LEN);
buf[NET_IP_STR_LEN-1] = '\0'; /* We are not sure the input is sane. */ buf[NET_IP_STR_LEN-1] = '\0'; /* We are not sure the input is sane. */
} else { } else {
anetPeerToString(link->fd, buf, NET_IP_STR_LEN, NULL); connPeerToString(link->conn, buf, NET_IP_STR_LEN, NULL);
} }
} }
...@@ -1751,7 +1778,7 @@ int clusterProcessPacket(clusterLink *link) { ...@@ -1751,7 +1778,7 @@ int clusterProcessPacket(clusterLink *link) {
{ {
char ip[NET_IP_STR_LEN]; char ip[NET_IP_STR_LEN];
if (anetSockName(link->fd,ip,sizeof(ip),NULL) != -1 && if (connSockName(link->conn,ip,sizeof(ip),NULL) != -1 &&
strcmp(ip,myself->ip)) strcmp(ip,myself->ip))
{ {
memcpy(myself->ip,ip,NET_IP_STR_LEN); memcpy(myself->ip,ip,NET_IP_STR_LEN);
...@@ -2118,35 +2145,76 @@ void handleLinkIOError(clusterLink *link) { ...@@ -2118,35 +2145,76 @@ void handleLinkIOError(clusterLink *link) {
/* Send data. This is handled using a trivial send buffer that gets /* Send data. This is handled using a trivial send buffer that gets
* consumed by write(). We don't try to optimize this for speed too much * consumed by write(). We don't try to optimize this for speed too much
* as this is a very low traffic channel. */ * as this is a very low traffic channel. */
void clusterWriteHandler(aeEventLoop *el, int fd, void *privdata, int mask) { void clusterWriteHandler(connection *conn) {
clusterLink *link = (clusterLink*) privdata; clusterLink *link = connGetPrivateData(conn);
ssize_t nwritten; ssize_t nwritten;
UNUSED(el);
UNUSED(mask);
nwritten = write(fd, link->sndbuf, sdslen(link->sndbuf)); nwritten = connWrite(conn, link->sndbuf, sdslen(link->sndbuf));
if (nwritten <= 0) { if (nwritten <= 0) {
serverLog(LL_DEBUG,"I/O error writing to node link: %s", serverLog(LL_DEBUG,"I/O error writing to node link: %s",
(nwritten == -1) ? strerror(errno) : "short write"); (nwritten == -1) ? connGetLastError(conn) : "short write");
handleLinkIOError(link); handleLinkIOError(link);
return; return;
} }
sdsrange(link->sndbuf,nwritten,-1); sdsrange(link->sndbuf,nwritten,-1);
if (sdslen(link->sndbuf) == 0) if (sdslen(link->sndbuf) == 0)
aeDeleteFileEvent(server.el, link->fd, AE_WRITABLE); connSetWriteHandler(link->conn, NULL);
}
/* A connect handler that gets called when a connection to another node
* gets established.
*/
void clusterLinkConnectHandler(connection *conn) {
clusterLink *link = connGetPrivateData(conn);
clusterNode *node = link->node;
/* Check if connection succeeded */
if (connGetState(conn) != CONN_STATE_CONNECTED) {
serverLog(LL_VERBOSE, "Connection with Node %.40s at %s:%d failed: %s",
node->name, node->ip, node->cport,
connGetLastError(conn));
freeClusterLink(link);
return;
}
/* Register a read handler from now on */
connSetReadHandler(conn, clusterReadHandler);
/* Queue a PING in the new connection ASAP: this is crucial
* to avoid false positives in failure detection.
*
* If the node is flagged as MEET, we send a MEET message instead
* of a PING one, to force the receiver to add us in its node
* table. */
mstime_t old_ping_sent = node->ping_sent;
clusterSendPing(link, node->flags & CLUSTER_NODE_MEET ?
CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING);
if (old_ping_sent) {
/* If there was an active ping before the link was
* disconnected, we want to restore the ping time, otherwise
* replaced by the clusterSendPing() call. */
node->ping_sent = old_ping_sent;
}
/* We can clear the flag after the first packet is sent.
* If we'll never receive a PONG, we'll never send new packets
* to this node. Instead after the PONG is received and we
* are no longer in meet/handshake status, we want to send
* normal PING packets. */
node->flags &= ~CLUSTER_NODE_MEET;
serverLog(LL_DEBUG,"Connecting with Node %.40s at %s:%d",
node->name, node->ip, node->cport);
} }
/* Read data. Try to read the first field of the header first to check the /* Read data. Try to read the first field of the header first to check the
* full length of the packet. When a whole packet is in memory this function * full length of the packet. When a whole packet is in memory this function
* will call the function to process the packet. And so forth. */ * will call the function to process the packet. And so forth. */
void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) { void clusterReadHandler(connection *conn) {
char buf[sizeof(clusterMsg)]; char buf[sizeof(clusterMsg)];
ssize_t nread; ssize_t nread;
clusterMsg *hdr; clusterMsg *hdr;
clusterLink *link = (clusterLink*) privdata; clusterLink *link = connGetPrivateData(conn);
unsigned int readlen, rcvbuflen; unsigned int readlen, rcvbuflen;
UNUSED(el);
UNUSED(mask);
while(1) { /* Read as long as there is data to read. */ while(1) { /* Read as long as there is data to read. */
rcvbuflen = sdslen(link->rcvbuf); rcvbuflen = sdslen(link->rcvbuf);
...@@ -2174,13 +2242,13 @@ void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -2174,13 +2242,13 @@ void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
if (readlen > sizeof(buf)) readlen = sizeof(buf); if (readlen > sizeof(buf)) readlen = sizeof(buf);
} }
nread = read(fd,buf,readlen); nread = connRead(conn,buf,readlen);
if (nread == -1 && errno == EAGAIN) return; /* No more data ready. */ if (nread == -1 && (connGetState(conn) == CONN_STATE_CONNECTED)) return; /* No more data ready. */
if (nread <= 0) { if (nread <= 0) {
/* I/O error... */ /* I/O error... */
serverLog(LL_DEBUG,"I/O error reading from node link: %s", serverLog(LL_DEBUG,"I/O error reading from node link: %s",
(nread == 0) ? "connection closed" : strerror(errno)); (nread == 0) ? "connection closed" : connGetLastError(conn));
handleLinkIOError(link); handleLinkIOError(link);
return; return;
} else { } else {
...@@ -2209,8 +2277,7 @@ void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -2209,8 +2277,7 @@ void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
* from event handlers that will do stuff with the same link later. */ * from event handlers that will do stuff with the same link later. */
void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) { void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) {
if (sdslen(link->sndbuf) == 0 && msglen != 0) if (sdslen(link->sndbuf) == 0 && msglen != 0)
aeCreateFileEvent(server.el,link->fd,AE_WRITABLE|AE_BARRIER, connSetWriteHandler(link->conn, clusterWriteHandler); /* TODO: Handle AE_BARRIER in conns */
clusterWriteHandler,link);
link->sndbuf = sdscatlen(link->sndbuf, msg, msglen); link->sndbuf = sdscatlen(link->sndbuf, msg, msglen);
...@@ -2276,11 +2343,12 @@ void clusterBuildMessageHdr(clusterMsg *hdr, int type) { ...@@ -2276,11 +2343,12 @@ void clusterBuildMessageHdr(clusterMsg *hdr, int type) {
} }
/* Handle cluster-announce-port as well. */ /* Handle cluster-announce-port as well. */
int port = server.tls_cluster ? server.tls_port : server.port;
int announced_port = server.cluster_announce_port ? int announced_port = server.cluster_announce_port ?
server.cluster_announce_port : server.port; server.cluster_announce_port : port;
int announced_cport = server.cluster_announce_bus_port ? int announced_cport = server.cluster_announce_bus_port ?
server.cluster_announce_bus_port : server.cluster_announce_bus_port :
(server.port + CLUSTER_PORT_INCR); (port + CLUSTER_PORT_INCR);
memcpy(hdr->myslots,master->slots,sizeof(hdr->myslots)); memcpy(hdr->myslots,master->slots,sizeof(hdr->myslots));
memset(hdr->slaveof,0,CLUSTER_NAMELEN); memset(hdr->slaveof,0,CLUSTER_NAMELEN);
...@@ -3383,13 +3451,11 @@ void clusterCron(void) { ...@@ -3383,13 +3451,11 @@ void clusterCron(void) {
} }
if (node->link == NULL) { if (node->link == NULL) {
int fd; clusterLink *link = createClusterLink(node);
mstime_t old_ping_sent; link->conn = server.tls_cluster ? connCreateTLS() : connCreateSocket();
clusterLink *link; connSetPrivateData(link->conn, link);
if (connConnect(link->conn, node->ip, node->cport, NET_FIRST_BIND_ADDR,
fd = anetTcpNonBlockBindConnect(server.neterr, node->ip, clusterLinkConnectHandler) == -1) {
node->cport, NET_FIRST_BIND_ADDR);
if (fd == -1) {
/* We got a synchronous error from connect before /* We got a synchronous error from connect before
* clusterSendPing() had a chance to be called. * clusterSendPing() had a chance to be called.
* If node->ping_sent is zero, failure detection can't work, * If node->ping_sent is zero, failure detection can't work,
...@@ -3399,37 +3465,11 @@ void clusterCron(void) { ...@@ -3399,37 +3465,11 @@ void clusterCron(void) {
serverLog(LL_DEBUG, "Unable to connect to " serverLog(LL_DEBUG, "Unable to connect to "
"Cluster Node [%s]:%d -> %s", node->ip, "Cluster Node [%s]:%d -> %s", node->ip,
node->cport, server.neterr); node->cport, server.neterr);
freeClusterLink(link);
continue; continue;
} }
link = createClusterLink(node);
link->fd = fd;
node->link = link; node->link = link;
aeCreateFileEvent(server.el,link->fd,AE_READABLE,
clusterReadHandler,link);
/* Queue a PING in the new connection ASAP: this is crucial
* to avoid false positives in failure detection.
*
* If the node is flagged as MEET, we send a MEET message instead
* of a PING one, to force the receiver to add us in its node
* table. */
old_ping_sent = node->ping_sent;
clusterSendPing(link, node->flags & CLUSTER_NODE_MEET ?
CLUSTERMSG_TYPE_MEET : CLUSTERMSG_TYPE_PING);
if (old_ping_sent) {
/* If there was an active ping before the link was
* disconnected, we want to restore the ping time, otherwise
* replaced by the clusterSendPing() call. */
node->ping_sent = old_ping_sent;
}
/* We can clear the flag after the first packet is sent.
* If we'll never receive a PONG, we'll never send new packets
* to this node. Instead after the PONG is received and we
* are no longer in meet/handshake status, we want to send
* normal PING packets. */
node->flags &= ~CLUSTER_NODE_MEET;
serverLog(LL_DEBUG,"Connecting with Node %.40s at %s:%d",
node->name, node->ip, node->cport);
} }
} }
dictReleaseIterator(di); dictReleaseIterator(di);
...@@ -4940,7 +4980,7 @@ void restoreCommand(client *c) { ...@@ -4940,7 +4980,7 @@ void restoreCommand(client *c) {
#define MIGRATE_SOCKET_CACHE_TTL 10 /* close cached sockets after 10 sec. */ #define MIGRATE_SOCKET_CACHE_TTL 10 /* close cached sockets after 10 sec. */
typedef struct migrateCachedSocket { typedef struct migrateCachedSocket {
int fd; connection *conn;
long last_dbid; long last_dbid;
time_t last_use_time; time_t last_use_time;
} migrateCachedSocket; } migrateCachedSocket;
...@@ -4957,7 +4997,7 @@ typedef struct migrateCachedSocket { ...@@ -4957,7 +4997,7 @@ typedef struct migrateCachedSocket {
* should be called so that the connection will be created from scratch * should be called so that the connection will be created from scratch
* the next time. */ * the next time. */
migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long timeout) { migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long timeout) {
int fd; connection *conn;
sds name = sdsempty(); sds name = sdsempty();
migrateCachedSocket *cs; migrateCachedSocket *cs;
...@@ -4977,34 +5017,27 @@ migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long ti ...@@ -4977,34 +5017,27 @@ migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long ti
/* Too many items, drop one at random. */ /* Too many items, drop one at random. */
dictEntry *de = dictGetRandomKey(server.migrate_cached_sockets); dictEntry *de = dictGetRandomKey(server.migrate_cached_sockets);
cs = dictGetVal(de); cs = dictGetVal(de);
close(cs->fd); connClose(cs->conn);
zfree(cs); zfree(cs);
dictDelete(server.migrate_cached_sockets,dictGetKey(de)); dictDelete(server.migrate_cached_sockets,dictGetKey(de));
} }
/* Create the socket */ /* Create the socket */
fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr, conn = server.tls_cluster ? connCreateTLS() : connCreateSocket();
atoi(c->argv[2]->ptr)); if (connBlockingConnect(conn, c->argv[1]->ptr, atoi(c->argv[2]->ptr), timeout)
if (fd == -1) { != C_OK) {
sdsfree(name);
addReplyErrorFormat(c,"Can't connect to target node: %s",
server.neterr);
return NULL;
}
anetEnableTcpNoDelay(server.neterr,fd);
/* Check if it connects within the specified timeout. */
if ((aeWait(fd,AE_WRITABLE,timeout) & AE_WRITABLE) == 0) {
sdsfree(name);
addReplySds(c, addReplySds(c,
sdsnew("-IOERR error or timeout connecting to the client\r\n")); sdsnew("-IOERR error or timeout connecting to the client\r\n"));
close(fd); connClose(conn);
sdsfree(name);
return NULL; return NULL;
} }
connEnableTcpNoDelay(conn);
/* Add to the cache and return it to the caller. */ /* Add to the cache and return it to the caller. */
cs = zmalloc(sizeof(*cs)); cs = zmalloc(sizeof(*cs));
cs->fd = fd; cs->conn = conn;
cs->last_dbid = -1; cs->last_dbid = -1;
cs->last_use_time = server.unixtime; cs->last_use_time = server.unixtime;
dictAdd(server.migrate_cached_sockets,name,cs); dictAdd(server.migrate_cached_sockets,name,cs);
...@@ -5025,7 +5058,7 @@ void migrateCloseSocket(robj *host, robj *port) { ...@@ -5025,7 +5058,7 @@ void migrateCloseSocket(robj *host, robj *port) {
return; return;
} }
close(cs->fd); connClose(cs->conn);
zfree(cs); zfree(cs);
dictDelete(server.migrate_cached_sockets,name); dictDelete(server.migrate_cached_sockets,name);
sdsfree(name); sdsfree(name);
...@@ -5039,7 +5072,7 @@ void migrateCloseTimedoutSockets(void) { ...@@ -5039,7 +5072,7 @@ void migrateCloseTimedoutSockets(void) {
migrateCachedSocket *cs = dictGetVal(de); migrateCachedSocket *cs = dictGetVal(de);
if ((server.unixtime - cs->last_use_time) > MIGRATE_SOCKET_CACHE_TTL) { if ((server.unixtime - cs->last_use_time) > MIGRATE_SOCKET_CACHE_TTL) {
close(cs->fd); connClose(cs->conn);
zfree(cs); zfree(cs);
dictDelete(server.migrate_cached_sockets,dictGetKey(de)); dictDelete(server.migrate_cached_sockets,dictGetKey(de));
} }
...@@ -5221,7 +5254,7 @@ try_again: ...@@ -5221,7 +5254,7 @@ try_again:
while ((towrite = sdslen(buf)-pos) > 0) { while ((towrite = sdslen(buf)-pos) > 0) {
towrite = (towrite > (64*1024) ? (64*1024) : towrite); towrite = (towrite > (64*1024) ? (64*1024) : towrite);
nwritten = syncWrite(cs->fd,buf+pos,towrite,timeout); nwritten = connSyncWrite(cs->conn,buf+pos,towrite,timeout);
if (nwritten != (signed)towrite) { if (nwritten != (signed)towrite) {
write_error = 1; write_error = 1;
goto socket_err; goto socket_err;
...@@ -5235,11 +5268,11 @@ try_again: ...@@ -5235,11 +5268,11 @@ try_again:
char buf2[1024]; /* Restore reply. */ char buf2[1024]; /* Restore reply. */
/* Read the AUTH reply if needed. */ /* Read the AUTH reply if needed. */
if (password && syncReadLine(cs->fd, buf0, sizeof(buf0), timeout) <= 0) if (password && connSyncReadLine(cs->conn, buf0, sizeof(buf0), timeout) <= 0)
goto socket_err; goto socket_err;
/* Read the SELECT reply if needed. */ /* Read the SELECT reply if needed. */
if (select && syncReadLine(cs->fd, buf1, sizeof(buf1), timeout) <= 0) if (select && connSyncReadLine(cs->conn, buf1, sizeof(buf1), timeout) <= 0)
goto socket_err; goto socket_err;
/* Read the RESTORE replies. */ /* Read the RESTORE replies. */
...@@ -5254,7 +5287,7 @@ try_again: ...@@ -5254,7 +5287,7 @@ try_again:
if (!copy) newargv = zmalloc(sizeof(robj*)*(num_keys+1)); if (!copy) newargv = zmalloc(sizeof(robj*)*(num_keys+1));
for (j = 0; j < num_keys; j++) { for (j = 0; j < num_keys; j++) {
if (syncReadLine(cs->fd, buf2, sizeof(buf2), timeout) <= 0) { if (connSyncReadLine(cs->conn, buf2, sizeof(buf2), timeout) <= 0) {
socket_error = 1; socket_error = 1;
break; break;
} }
......
...@@ -40,7 +40,7 @@ struct clusterNode; ...@@ -40,7 +40,7 @@ struct clusterNode;
/* clusterLink encapsulates everything needed to talk with a remote node. */ /* clusterLink encapsulates everything needed to talk with a remote node. */
typedef struct clusterLink { typedef struct clusterLink {
mstime_t ctime; /* Link creation time */ mstime_t ctime; /* Link creation time */
int fd; /* TCP socket file descriptor */ connection *conn; /* Connection to remote node */
sds sndbuf; /* Packet send buffer */ sds sndbuf; /* Packet send buffer */
sds rcvbuf; /* Packet reception buffer */ sds rcvbuf; /* Packet reception buffer */
struct clusterNode *node; /* Node related to this link if any, or NULL */ struct clusterNode *node; /* Node related to this link if any, or NULL */
......
...@@ -286,6 +286,15 @@ void loadServerConfigFromString(char *config) { ...@@ -286,6 +286,15 @@ void loadServerConfigFromString(char *config) {
if (server.port < 0 || server.port > 65535) { if (server.port < 0 || server.port > 65535) {
err = "Invalid port"; goto loaderr; err = "Invalid port"; goto loaderr;
} }
} else if (!strcasecmp(argv[0],"tls-port") && argc == 2) {
#ifdef USE_OPENSSL
server.tls_port = atoi(argv[1]);
if (server.port < 0 || server.port > 65535) {
err = "Invalid port"; goto loaderr;
}
#else
err = "TLS not supported"; goto loaderr;
#endif
} else if (!strcasecmp(argv[0],"tcp-backlog") && argc == 2) { } else if (!strcasecmp(argv[0],"tcp-backlog") && argc == 2) {
server.tcp_backlog = atoi(argv[1]); server.tcp_backlog = atoi(argv[1]);
if (server.tcp_backlog < 0) { if (server.tcp_backlog < 0) {
...@@ -791,6 +800,24 @@ void loadServerConfigFromString(char *config) { ...@@ -791,6 +800,24 @@ void loadServerConfigFromString(char *config) {
err = sentinelHandleConfiguration(argv+1,argc-1); err = sentinelHandleConfiguration(argv+1,argc-1);
if (err) goto loaderr; if (err) goto loaderr;
} }
} else if (!strcasecmp(argv[0],"tls-cert-file") && argc == 2) {
zfree(server.tls_cert_file);
server.tls_cert_file = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"tls-key-file") && argc == 2) {
zfree(server.tls_key_file);
server.tls_key_file = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"tls-dh-params-file") && argc == 2) {
zfree(server.tls_dh_params_file);
server.tls_dh_params_file = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"tls-ca-cert-file") && argc == 2) {
zfree(server.tls_ca_cert_file);
server.tls_ca_cert_file = zstrdup(argv[1]);
} else if (!strcasecmp(argv[0],"tls-cluster") && argc == 2) {
server.tls_cluster = yesnotoi(argv[1]);
} else if (!strcasecmp(argv[0],"tls-replication") && argc == 2) {
server.tls_replication = yesnotoi(argv[1]);
} else if (!strcasecmp(argv[0],"tls-auth-clients") && argc == 2) {
server.tls_auth_clients = yesnotoi(argv[1]);
} else { } else {
err = "Bad directive or wrong number of arguments"; goto loaderr; err = "Bad directive or wrong number of arguments"; goto loaderr;
} }
...@@ -1234,6 +1261,45 @@ void configSetCommand(client *c) { ...@@ -1234,6 +1261,45 @@ void configSetCommand(client *c) {
} config_set_enum_field( } config_set_enum_field(
"repl-diskless-load",server.repl_diskless_load,repl_diskless_load_enum) { "repl-diskless-load",server.repl_diskless_load,repl_diskless_load_enum) {
/* TLS fields. */
} config_set_special_field("tls-cert-file") {
if (tlsConfigure((char *) o->ptr, server.tls_key_file,
server.tls_dh_params_file, server.tls_ca_cert_file) == C_ERR) {
addReplyError(c,
"Unable to configure tls-cert-file. Check server logs.");
return;
}
zfree(server.tls_cert_file);
server.tls_cert_file = zstrdup(o->ptr);
} config_set_special_field("tls-key-file") {
if (tlsConfigure(server.tls_cert_file, (char *) o->ptr,
server.tls_dh_params_file, server.tls_ca_cert_file) == C_ERR) {
addReplyError(c,
"Unable to configure tls-key-file. Check server logs.");
return;
}
zfree(server.tls_key_file);
server.tls_key_file = zstrdup(o->ptr);
} config_set_special_field("tls-dh-params-file") {
if (tlsConfigure(server.tls_cert_file, server.tls_key_file,
(char *) o->ptr, server.tls_ca_cert_file) == C_ERR) {
addReplyError(c,
"Unable to configure tls-dh-params-file. Check server logs.");
return;
}
zfree(server.tls_dh_params_file);
server.tls_dh_params_file = zstrdup(o->ptr);
} config_set_special_field("tls-ca-cert-file") {
if (tlsConfigure(server.tls_cert_file, server.tls_key_file,
server.tls_dh_params_file, (char *) o->ptr) == C_ERR) {
addReplyError(c,
"Unable to configure tls-ca-cert-file. Check server logs.");
return;
}
zfree(server.tls_ca_cert_file);
server.tls_ca_cert_file = zstrdup(o->ptr);
} config_set_bool_field("tls-auth-clients", server.tls_auth_clients) {
/* Everyhing else is an error... */ /* Everyhing else is an error... */
} config_set_else { } config_set_else {
addReplyErrorFormat(c,"Unsupported CONFIG parameter: %s", addReplyErrorFormat(c,"Unsupported CONFIG parameter: %s",
...@@ -1307,6 +1373,10 @@ void configGetCommand(client *c) { ...@@ -1307,6 +1373,10 @@ void configGetCommand(client *c) {
config_get_string_field("pidfile",server.pidfile); config_get_string_field("pidfile",server.pidfile);
config_get_string_field("slave-announce-ip",server.slave_announce_ip); config_get_string_field("slave-announce-ip",server.slave_announce_ip);
config_get_string_field("replica-announce-ip",server.slave_announce_ip); config_get_string_field("replica-announce-ip",server.slave_announce_ip);
config_get_string_field("tls-cert-file",server.tls_cert_file);
config_get_string_field("tls-key-file",server.tls_key_file);
config_get_string_field("tls-dh-params-file",server.tls_dh_params_file);
config_get_string_field("tls-ca-cert-file",server.tls_ca_cert_file);
/* Numerical values */ /* Numerical values */
config_get_numerical_field("maxmemory",server.maxmemory); config_get_numerical_field("maxmemory",server.maxmemory);
...@@ -1354,6 +1424,7 @@ void configGetCommand(client *c) { ...@@ -1354,6 +1424,7 @@ void configGetCommand(client *c) {
config_get_numerical_field("slowlog-max-len", server.slowlog_max_len); config_get_numerical_field("slowlog-max-len", server.slowlog_max_len);
config_get_numerical_field("tracking-table-max-fill", server.tracking_table_max_fill); config_get_numerical_field("tracking-table-max-fill", server.tracking_table_max_fill);
config_get_numerical_field("port",server.port); config_get_numerical_field("port",server.port);
config_get_numerical_field("tls-port",server.tls_port);
config_get_numerical_field("cluster-announce-port",server.cluster_announce_port); config_get_numerical_field("cluster-announce-port",server.cluster_announce_port);
config_get_numerical_field("cluster-announce-bus-port",server.cluster_announce_bus_port); config_get_numerical_field("cluster-announce-bus-port",server.cluster_announce_bus_port);
config_get_numerical_field("tcp-backlog",server.tcp_backlog); config_get_numerical_field("tcp-backlog",server.tcp_backlog);
...@@ -1393,6 +1464,9 @@ void configGetCommand(client *c) { ...@@ -1393,6 +1464,9 @@ void configGetCommand(client *c) {
} }
config_get_bool_field("activedefrag", server.active_defrag_enabled); config_get_bool_field("activedefrag", server.active_defrag_enabled);
config_get_bool_field("tls-cluster",server.tls_cluster);
config_get_bool_field("tls-replication",server.tls_replication);
config_get_bool_field("tls-auth-clients",server.tls_auth_clients);
/* Enum values */ /* Enum values */
config_get_enum_field("maxmemory-policy", config_get_enum_field("maxmemory-policy",
...@@ -2113,10 +2187,13 @@ int rewriteConfig(char *path) { ...@@ -2113,10 +2187,13 @@ int rewriteConfig(char *path) {
} }
rewriteConfigStringOption(state,"pidfile",server.pidfile,CONFIG_DEFAULT_PID_FILE); rewriteConfigStringOption(state,"pidfile",server.pidfile,CONFIG_DEFAULT_PID_FILE);
rewriteConfigNumericalOption(state,"port",server.port,CONFIG_DEFAULT_SERVER_PORT); rewriteConfigNumericalOption(state,"tls-port",server.tls_port,CONFIG_DEFAULT_SERVER_TLS_PORT);
rewriteConfigNumericalOption(state,"cluster-announce-port",server.cluster_announce_port,CONFIG_DEFAULT_CLUSTER_ANNOUNCE_PORT); rewriteConfigNumericalOption(state,"cluster-announce-port",server.cluster_announce_port,CONFIG_DEFAULT_CLUSTER_ANNOUNCE_PORT);
rewriteConfigNumericalOption(state,"cluster-announce-bus-port",server.cluster_announce_bus_port,CONFIG_DEFAULT_CLUSTER_ANNOUNCE_BUS_PORT); rewriteConfigNumericalOption(state,"cluster-announce-bus-port",server.cluster_announce_bus_port,CONFIG_DEFAULT_CLUSTER_ANNOUNCE_BUS_PORT);
rewriteConfigNumericalOption(state,"tcp-backlog",server.tcp_backlog,CONFIG_DEFAULT_TCP_BACKLOG); rewriteConfigNumericalOption(state,"tcp-backlog",server.tcp_backlog,CONFIG_DEFAULT_TCP_BACKLOG);
rewriteConfigYesNoOption(state,"tls-cluster",server.tls_cluster,0);
rewriteConfigYesNoOption(state,"tls-replication",server.tls_replication,0);
rewriteConfigYesNoOption(state,"tls-auth-clients",server.tls_auth_clients,1);
rewriteConfigBindOption(state); rewriteConfigBindOption(state);
rewriteConfigStringOption(state,"unixsocket",server.unixsocket,NULL); rewriteConfigStringOption(state,"unixsocket",server.unixsocket,NULL);
rewriteConfigOctalOption(state,"unixsocketperm",server.unixsocketperm,CONFIG_DEFAULT_UNIX_SOCKET_PERM); rewriteConfigOctalOption(state,"unixsocketperm",server.unixsocketperm,CONFIG_DEFAULT_UNIX_SOCKET_PERM);
...@@ -2195,6 +2272,10 @@ int rewriteConfig(char *path) { ...@@ -2195,6 +2272,10 @@ int rewriteConfig(char *path) {
rewriteConfigNumericalOption(state,"hz",server.config_hz,CONFIG_DEFAULT_HZ); rewriteConfigNumericalOption(state,"hz",server.config_hz,CONFIG_DEFAULT_HZ);
rewriteConfigEnumOption(state,"supervised",server.supervised_mode,supervised_mode_enum,SUPERVISED_NONE); rewriteConfigEnumOption(state,"supervised",server.supervised_mode,supervised_mode_enum,SUPERVISED_NONE);
rewriteConfigNumericalOption(state,"rdb-key-save-delay",server.rdb_key_save_delay,CONFIG_DEFAULT_RDB_KEY_SAVE_DELAY); rewriteConfigNumericalOption(state,"rdb-key-save-delay",server.rdb_key_save_delay,CONFIG_DEFAULT_RDB_KEY_SAVE_DELAY);
rewriteConfigStringOption(state,"tls-cert-file",server.tls_cert_file,NULL);
rewriteConfigStringOption(state,"tls-key-file",server.tls_key_file,NULL);
rewriteConfigStringOption(state,"tls-dh-params-file",server.tls_dh_params_file,NULL);
rewriteConfigStringOption(state,"tls-ca-cert-file",server.tls_ca_cert_file,NULL);
/* Rewrite Sentinel config if in Sentinel mode. */ /* Rewrite Sentinel config if in Sentinel mode. */
if (server.sentinel_mode) rewriteConfigSentinelOption(state); if (server.sentinel_mode) rewriteConfigSentinelOption(state);
......
/*
* Copyright (c) 2019, Redis Labs
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#include "server.h"
#include "connhelpers.h"
/* The connections module provides a lean abstraction of network connections
* to avoid direct socket and async event management across the Redis code base.
*
* It does NOT provide advanced connection features commonly found in similar
* libraries such as complete in/out buffer management, throttling, etc. These
* functions remain in networking.c.
*
* The primary goal is to allow transparent handling of TCP and TLS based
* connections. To do so, connections have the following properties:
*
* 1. A connection may live before its corresponding socket exists. This
* allows various context and configuration setting to be handled before
* establishing the actual connection.
* 2. The caller may register/unregister logical read/write handlers to be
* called when the connection has data to read from/can accept writes.
* These logical handlers may or may not correspond to actual AE events,
* depending on the implementation (for TCP they are; for TLS they aren't).
*/
ConnectionType CT_Socket;
/* When a connection is created we must know its type already, but the
* underlying socket may or may not exist:
*
* - For accepted connections, it exists as we do not model the listen/accept
* part; So caller calls connCreateSocket() followed by connAccept().
* - For outgoing connections, the socket is created by the connection module
* itself; So caller calls connCreateSocket() followed by connConnect(),
* which registers a connect callback that fires on connected/error state
* (and after any transport level handshake was done).
*
* NOTE: An earlier version relied on connections being part of other structs
* and not independently allocated. This could lead to further optimizations
* like using container_of(), etc. However it was discontinued in favor of
* this approach for these reasons:
*
* 1. In some cases conns are created/handled outside the context of the
* containing struct, in which case it gets a bit awkward to copy them.
* 2. Future implementations may wish to allocate arbitrary data for the
* connection.
* 3. The container_of() approach is anyway risky because connections may
* be embedded in different structs, not just client.
*/
connection *connCreateSocket() {
connection *conn = zcalloc(sizeof(connection));
conn->type = &CT_Socket;
conn->fd = -1;
return conn;
}
/* Create a new socket-type connection that is already associated with
* an accepted connection.
*
* The socket is not read for I/O until connAccept() was called and
* invoked the connection-level accept handler.
*/
connection *connCreateAcceptedSocket(int fd) {
connection *conn = connCreateSocket();
conn->fd = fd;
conn->state = CONN_STATE_ACCEPTING;
return conn;
}
static int connSocketConnect(connection *conn, const char *addr, int port, const char *src_addr,
ConnectionCallbackFunc connect_handler) {
int fd = anetTcpNonBlockBestEffortBindConnect(NULL,addr,port,src_addr);
if (fd == -1) {
conn->state = CONN_STATE_ERROR;
conn->last_errno = errno;
return C_ERR;
}
conn->fd = fd;
conn->state = CONN_STATE_CONNECTING;
conn->conn_handler = connect_handler;
aeCreateFileEvent(server.el, conn->fd, AE_WRITABLE,
conn->type->ae_handler, conn);
return C_OK;
}
/* Returns true if a write handler is registered */
int connHasWriteHandler(connection *conn) {
return conn->write_handler != NULL;
}
/* Returns true if a read handler is registered */
int connHasReadHandler(connection *conn) {
return conn->read_handler != NULL;
}
/* Associate a private data pointer with the connection */
void connSetPrivateData(connection *conn, void *data) {
conn->private_data = data;
}
/* Get the associated private data pointer */
void *connGetPrivateData(connection *conn) {
return conn->private_data;
}
/* ------ Pure socket connections ------- */
/* A very incomplete list of implementation-specific calls. Much of the above shall
* move here as we implement additional connection types.
*/
static int connSocketShutdown(connection *conn, int how) {
return shutdown(conn->fd, how);
}
/* Close the connection and free resources. */
static void connSocketClose(connection *conn) {
if (conn->fd != -1) {
aeDeleteFileEvent(server.el,conn->fd,AE_READABLE);
aeDeleteFileEvent(server.el,conn->fd,AE_WRITABLE);
close(conn->fd);
conn->fd = -1;
}
/* If called from within a handler, schedule the close but
* keep the connection until the handler returns.
*/
if (conn->flags & CONN_FLAG_IN_HANDLER) {
conn->flags |= CONN_FLAG_CLOSE_SCHEDULED;
return;
}
zfree(conn);
}
static int connSocketWrite(connection *conn, const void *data, size_t data_len) {
int ret = write(conn->fd, data, data_len);
if (!ret) {
conn->state = CONN_STATE_CLOSED;
} else if (ret < 0 && errno != EAGAIN) {
conn->last_errno = errno;
conn->state = CONN_STATE_ERROR;
}
return ret;
}
static int connSocketRead(connection *conn, void *buf, size_t buf_len) {
int ret = read(conn->fd, buf, buf_len);
if (!ret) {
conn->state = CONN_STATE_CLOSED;
} else if (ret < 0 && errno != EAGAIN) {
conn->last_errno = errno;
conn->state = CONN_STATE_ERROR;
}
return ret;
}
static int connSocketAccept(connection *conn, ConnectionCallbackFunc accept_handler) {
if (conn->state != CONN_STATE_ACCEPTING) return C_ERR;
conn->state = CONN_STATE_CONNECTED;
if (!callHandler(conn, accept_handler)) return C_ERR;
return C_OK;
}
/* Register a write handler, to be called when the connection is writable.
* If NULL, the existing handler is removed.
*/
static int connSocketSetWriteHandler(connection *conn, ConnectionCallbackFunc func) {
if (func == conn->write_handler) return C_OK;
conn->write_handler = func;
if (!conn->write_handler)
aeDeleteFileEvent(server.el,conn->fd,AE_WRITABLE);
else
if (aeCreateFileEvent(server.el,conn->fd,AE_WRITABLE,
conn->type->ae_handler,conn) == AE_ERR) return C_ERR;
return C_OK;
}
/* Register a read handler, to be called when the connection is readable.
* If NULL, the existing handler is removed.
*/
static int connSocketSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
if (func == conn->read_handler) return C_OK;
conn->read_handler = func;
if (!conn->read_handler)
aeDeleteFileEvent(server.el,conn->fd,AE_READABLE);
else
if (aeCreateFileEvent(server.el,conn->fd,
AE_READABLE,conn->type->ae_handler,conn) == AE_ERR) return C_ERR;
return C_OK;
}
static const char *connSocketGetLastError(connection *conn) {
return strerror(conn->last_errno);
}
static void connSocketEventHandler(struct aeEventLoop *el, int fd, void *clientData, int mask)
{
UNUSED(el);
UNUSED(fd);
connection *conn = clientData;
if (conn->state == CONN_STATE_CONNECTING &&
(mask & AE_WRITABLE) && conn->conn_handler) {
if (connGetSocketError(conn)) {
conn->last_errno = errno;
conn->state = CONN_STATE_ERROR;
} else {
conn->state = CONN_STATE_CONNECTED;
}
if (!conn->write_handler) aeDeleteFileEvent(server.el,conn->fd,AE_WRITABLE);
if (!callHandler(conn, conn->conn_handler)) return;
conn->conn_handler = NULL;
}
/* Handle normal I/O flows */
if ((mask & AE_READABLE) && conn->read_handler) {
if (!callHandler(conn, conn->read_handler)) return;
}
if ((mask & AE_WRITABLE) && conn->write_handler) {
if (!callHandler(conn, conn->write_handler)) return;
}
}
static int connSocketBlockingConnect(connection *conn, const char *addr, int port, long long timeout) {
int fd = anetTcpNonBlockConnect(NULL,addr,port);
if (fd == -1) {
conn->state = CONN_STATE_ERROR;
conn->last_errno = errno;
return C_ERR;
}
if ((aeWait(fd, AE_WRITABLE, timeout) & AE_WRITABLE) == 0) {
conn->state = CONN_STATE_ERROR;
conn->last_errno = ETIMEDOUT;
}
conn->fd = fd;
conn->state = CONN_STATE_CONNECTED;
return C_OK;
}
/* Connection-based versions of syncio.c functions.
* NOTE: This should ideally be refactored out in favor of pure async work.
*/
static ssize_t connSocketSyncWrite(connection *conn, char *ptr, ssize_t size, long long timeout) {
return syncWrite(conn->fd, ptr, size, timeout);
}
static ssize_t connSocketSyncRead(connection *conn, char *ptr, ssize_t size, long long timeout) {
return syncRead(conn->fd, ptr, size, timeout);
}
static ssize_t connSocketSyncReadLine(connection *conn, char *ptr, ssize_t size, long long timeout) {
return syncReadLine(conn->fd, ptr, size, timeout);
}
ConnectionType CT_Socket = {
.ae_handler = connSocketEventHandler,
.close = connSocketClose,
.shutdown = connSocketShutdown,
.write = connSocketWrite,
.read = connSocketRead,
.accept = connSocketAccept,
.connect = connSocketConnect,
.set_write_handler = connSocketSetWriteHandler,
.set_read_handler = connSocketSetReadHandler,
.get_last_error = connSocketGetLastError,
.blocking_connect = connSocketBlockingConnect,
.sync_write = connSocketSyncWrite,
.sync_read = connSocketSyncRead,
.sync_readline = connSocketSyncReadLine
};
int connGetSocketError(connection *conn) {
int sockerr = 0;
socklen_t errlen = sizeof(sockerr);
if (getsockopt(conn->fd, SOL_SOCKET, SO_ERROR, &sockerr, &errlen) == -1)
sockerr = errno;
return sockerr;
}
int connPeerToString(connection *conn, char *ip, size_t ip_len, int *port) {
return anetPeerToString(conn ? conn->fd : -1, ip, ip_len, port);
}
int connFormatPeer(connection *conn, char *buf, size_t buf_len) {
return anetFormatPeer(conn ? conn->fd : -1, buf, buf_len);
}
int connSockName(connection *conn, char *ip, size_t ip_len, int *port) {
return anetSockName(conn->fd, ip, ip_len, port);
}
int connBlock(connection *conn) {
if (conn->fd == -1) return C_ERR;
return anetBlock(NULL, conn->fd);
}
int connNonBlock(connection *conn) {
if (conn->fd == -1) return C_ERR;
return anetNonBlock(NULL, conn->fd);
}
int connEnableTcpNoDelay(connection *conn) {
if (conn->fd == -1) return C_ERR;
return anetEnableTcpNoDelay(NULL, conn->fd);
}
int connDisableTcpNoDelay(connection *conn) {
if (conn->fd == -1) return C_ERR;
return anetDisableTcpNoDelay(NULL, conn->fd);
}
int connKeepAlive(connection *conn, int interval) {
if (conn->fd == -1) return C_ERR;
return anetKeepAlive(NULL, conn->fd, interval);
}
int connSendTimeout(connection *conn, long long ms) {
return anetSendTimeout(NULL, conn->fd, ms);
}
int connRecvTimeout(connection *conn, long long ms) {
return anetRecvTimeout(NULL, conn->fd, ms);
}
int connGetState(connection *conn) {
return conn->state;
}
/* Return a text that describes the connection, suitable for inclusion
* in CLIENT LIST and similar outputs.
*
* For sockets, we always return "fd=<fdnum>" to maintain compatibility.
*/
const char *connGetInfo(connection *conn, char *buf, size_t buf_len) {
snprintf(buf, buf_len-1, "fd=%i", conn->fd);
return buf;
}
/*
* Copyright (c) 2019, Redis Labs
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef __REDIS_CONNECTION_H
#define __REDIS_CONNECTION_H
#define CONN_INFO_LEN 32
struct aeEventLoop;
typedef struct connection connection;
typedef enum {
CONN_STATE_NONE = 0,
CONN_STATE_CONNECTING,
CONN_STATE_ACCEPTING,
CONN_STATE_CONNECTED,
CONN_STATE_CLOSED,
CONN_STATE_ERROR
} ConnectionState;
#define CONN_FLAG_IN_HANDLER (1<<0) /* A handler execution is in progress */
#define CONN_FLAG_CLOSE_SCHEDULED (1<<1) /* Closed scheduled by a handler */
typedef void (*ConnectionCallbackFunc)(struct connection *conn);
typedef struct ConnectionType {
void (*ae_handler)(struct aeEventLoop *el, int fd, void *clientData, int mask);
int (*connect)(struct connection *conn, const char *addr, int port, const char *source_addr, ConnectionCallbackFunc connect_handler);
int (*write)(struct connection *conn, const void *data, size_t data_len);
int (*read)(struct connection *conn, void *buf, size_t buf_len);
int (*shutdown)(struct connection *conn, int how);
void (*close)(struct connection *conn);
int (*accept)(struct connection *conn, ConnectionCallbackFunc accept_handler);
int (*set_write_handler)(struct connection *conn, ConnectionCallbackFunc handler);
int (*set_read_handler)(struct connection *conn, ConnectionCallbackFunc handler);
const char *(*get_last_error)(struct connection *conn);
int (*blocking_connect)(struct connection *conn, const char *addr, int port, long long timeout);
ssize_t (*sync_write)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
ssize_t (*sync_read)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
ssize_t (*sync_readline)(struct connection *conn, char *ptr, ssize_t size, long long timeout);
} ConnectionType;
struct connection {
ConnectionType *type;
ConnectionState state;
int flags;
int last_errno;
void *private_data;
ConnectionCallbackFunc conn_handler;
ConnectionCallbackFunc write_handler;
ConnectionCallbackFunc read_handler;
int fd;
};
/* The connection module does not deal with listening and accepting sockets,
* so we assume we have a socket when an incoming connection is created.
*
* The fd supplied should therefore be associated with an already accept()ed
* socket.
*
* connAccept() may directly call accept_handler(), or return and call it
* at a later time. This behavior is a bit awkward but aims to reduce the need
* to wait for the next event loop, if no additional handshake is required.
*/
static inline int connAccept(connection *conn, ConnectionCallbackFunc accept_handler) {
return conn->type->accept(conn, accept_handler);
}
/* Establish a connection. The connect_handler will be called when the connection
* is established, or if an error has occured.
*
* The connection handler will be responsible to set up any read/write handlers
* as needed.
*
* If C_ERR is returned, the operation failed and the connection handler shall
* not be expected.
*/
static inline int connConnect(connection *conn, const char *addr, int port, const char *src_addr,
ConnectionCallbackFunc connect_handler) {
return conn->type->connect(conn, addr, port, src_addr, connect_handler);
}
/* Blocking connect.
*
* NOTE: This is implemented in order to simplify the transition to the abstract
* connections, but should probably be refactored out of cluster.c and replication.c,
* in favor of a pure async implementation.
*/
static inline int connBlockingConnect(connection *conn, const char *addr, int port, long long timeout) {
return conn->type->blocking_connect(conn, addr, port, timeout);
}
/* Write to connection, behaves the same as write(2).
*
* Like write(2), a short write is possible. A -1 return indicates an error.
*
* The caller should NOT rely on errno. Testing for an EAGAIN-like condition, use
* connGetState() to see if the connection state is still CONN_STATE_CONNECTED.
*/
static inline int connWrite(connection *conn, const void *data, size_t data_len) {
return conn->type->write(conn, data, data_len);
}
/* Read from the connection, behaves the same as read(2).
*
* Like read(2), a short read is possible. A return value of 0 will indicate the
* connection was closed, and -1 will indicate an error.
*
* The caller should NOT rely on errno. Testing for an EAGAIN-like condition, use
* connGetState() to see if the connection state is still CONN_STATE_CONNECTED.
*/
static inline int connRead(connection *conn, void *buf, size_t buf_len) {
return conn->type->read(conn, buf, buf_len);
}
/* Register a write handler, to be called when the connection is writable.
* If NULL, the existing handler is removed.
*/
static inline int connSetWriteHandler(connection *conn, ConnectionCallbackFunc func) {
return conn->type->set_write_handler(conn, func);
}
/* Register a read handler, to be called when the connection is readable.
* If NULL, the existing handler is removed.
*/
static inline int connSetReadHandler(connection *conn, ConnectionCallbackFunc func) {
return conn->type->set_read_handler(conn, func);
}
static inline void connClose(connection *conn) {
conn->type->close(conn);
}
static inline int connShutdown(connection *conn, int how) {
return conn->type->shutdown(conn, how);
}
/* Returns the last error encountered by the connection, as a string. If no error,
* a NULL is returned.
*/
static inline const char *connGetLastError(connection *conn) {
return conn->type->get_last_error(conn);
}
static inline ssize_t connSyncWrite(connection *conn, char *ptr, ssize_t size, long long timeout) {
return conn->type->sync_write(conn, ptr, size, timeout);
}
static inline ssize_t connSyncRead(connection *conn, char *ptr, ssize_t size, long long timeout) {
return conn->type->sync_read(conn, ptr, size, timeout);
}
static inline ssize_t connSyncReadLine(connection *conn, char *ptr, ssize_t size, long long timeout) {
return conn->type->sync_readline(conn, ptr, size, timeout);
}
connection *connCreateSocket();
connection *connCreateAcceptedSocket(int fd);
connection *connCreateTLS();
connection *connCreateAcceptedTLS(int fd, int require_auth);
void connSetPrivateData(connection *conn, void *data);
void *connGetPrivateData(connection *conn);
int connGetState(connection *conn);
int connHasWriteHandler(connection *conn);
int connHasReadHandler(connection *conn);
int connGetSocketError(connection *conn);
/* anet-style wrappers to conns */
int connBlock(connection *conn);
int connNonBlock(connection *conn);
int connEnableTcpNoDelay(connection *conn);
int connDisableTcpNoDelay(connection *conn);
int connKeepAlive(connection *conn, int interval);
int connSendTimeout(connection *conn, long long ms);
int connRecvTimeout(connection *conn, long long ms);
int connPeerToString(connection *conn, char *ip, size_t ip_len, int *port);
int connFormatPeer(connection *conn, char *buf, size_t buf_len);
int connSockName(connection *conn, char *ip, size_t ip_len, int *port);
const char *connGetInfo(connection *conn, char *buf, size_t buf_len);
#endif /* __REDIS_CONNECTION_H */
/*
* Copyright (c) 2019, Redis Labs
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef __REDIS_CONNHELPERS_H
#define __REDIS_CONNHELPERS_H
#include "connection.h"
static inline void enterHandler(connection *conn) {
conn->flags |= CONN_FLAG_IN_HANDLER;
}
static inline int exitHandler(connection *conn) {
conn->flags &= ~CONN_FLAG_IN_HANDLER;
if (conn->flags & CONN_FLAG_CLOSE_SCHEDULED) {
connClose(conn);
return 0;
}
return 1;
}
static inline int callHandler(connection *conn, ConnectionCallbackFunc handler) {
conn->flags |= CONN_FLAG_IN_HANDLER;
if (handler) handler(conn);
conn->flags &= ~CONN_FLAG_IN_HANDLER;
if (conn->flags & CONN_FLAG_CLOSE_SCHEDULED) {
connClose(conn);
return 0;
}
return 1;
}
#endif /* __REDIS_CONNHELPERS_H */
...@@ -699,11 +699,12 @@ void _serverAssert(const char *estr, const char *file, int line) { ...@@ -699,11 +699,12 @@ void _serverAssert(const char *estr, const char *file, int line) {
void _serverAssertPrintClientInfo(const client *c) { void _serverAssertPrintClientInfo(const client *c) {
int j; int j;
char conninfo[CONN_INFO_LEN];
bugReportStart(); bugReportStart();
serverLog(LL_WARNING,"=== ASSERTION FAILED CLIENT CONTEXT ==="); serverLog(LL_WARNING,"=== ASSERTION FAILED CLIENT CONTEXT ===");
serverLog(LL_WARNING,"client->flags = %llu", (unsigned long long)c->flags); serverLog(LL_WARNING,"client->flags = %llu", (unsigned long long) c->flags);
serverLog(LL_WARNING,"client->fd = %d", c->fd); serverLog(LL_WARNING,"client->conn = %s", connGetInfo(c->conn, conninfo, sizeof(conninfo)));
serverLog(LL_WARNING,"client->argc = %d", c->argc); serverLog(LL_WARNING,"client->argc = %d", c->argc);
for (j=0; j < c->argc; j++) { for (j=0; j < c->argc; j++) {
char buf[128]; char buf[128];
......
...@@ -2764,7 +2764,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch ...@@ -2764,7 +2764,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
/* Create the client and dispatch the command. */ /* Create the client and dispatch the command. */
va_start(ap, fmt); va_start(ap, fmt);
c = createClient(-1); c = createClient(NULL);
c->user = NULL; /* Root user. */ c->user = NULL; /* Root user. */
argv = moduleCreateArgvFromUserFormat(cmdname,fmt,&argc,&flags,ap); argv = moduleCreateArgvFromUserFormat(cmdname,fmt,&argc,&flags,ap);
replicate = flags & REDISMODULE_ARGV_REPLICATE; replicate = flags & REDISMODULE_ARGV_REPLICATE;
...@@ -3681,7 +3681,7 @@ RedisModuleBlockedClient *RM_BlockClient(RedisModuleCtx *ctx, RedisModuleCmdFunc ...@@ -3681,7 +3681,7 @@ RedisModuleBlockedClient *RM_BlockClient(RedisModuleCtx *ctx, RedisModuleCmdFunc
bc->disconnect_callback = NULL; /* Set by RM_SetDisconnectCallback() */ bc->disconnect_callback = NULL; /* Set by RM_SetDisconnectCallback() */
bc->free_privdata = free_privdata; bc->free_privdata = free_privdata;
bc->privdata = NULL; bc->privdata = NULL;
bc->reply_client = createClient(-1); bc->reply_client = createClient(NULL);
bc->reply_client->flags |= CLIENT_MODULE; bc->reply_client->flags |= CLIENT_MODULE;
bc->dbid = c->db->id; bc->dbid = c->db->id;
c->bpop.timeout = timeout_ms ? (mstime()+timeout_ms) : 0; c->bpop.timeout = timeout_ms ? (mstime()+timeout_ms) : 0;
...@@ -3922,7 +3922,7 @@ RedisModuleCtx *RM_GetThreadSafeContext(RedisModuleBlockedClient *bc) { ...@@ -3922,7 +3922,7 @@ RedisModuleCtx *RM_GetThreadSafeContext(RedisModuleBlockedClient *bc) {
* access it safely from another thread, so we create a fake client here * access it safely from another thread, so we create a fake client here
* in order to keep things like the currently selected database and similar * in order to keep things like the currently selected database and similar
* things. */ * things. */
ctx->client = createClient(-1); ctx->client = createClient(NULL);
if (bc) { if (bc) {
selectDb(ctx->client,bc->dbid); selectDb(ctx->client,bc->dbid);
ctx->client->id = bc->client->id; ctx->client->id = bc->client->id;
...@@ -5113,7 +5113,7 @@ void moduleInitModulesSystem(void) { ...@@ -5113,7 +5113,7 @@ void moduleInitModulesSystem(void) {
/* Set up the keyspace notification susbscriber list and static client */ /* Set up the keyspace notification susbscriber list and static client */
moduleKeyspaceSubscribers = listCreate(); moduleKeyspaceSubscribers = listCreate();
moduleFreeContextReusedClient = createClient(-1); moduleFreeContextReusedClient = createClient(NULL);
moduleFreeContextReusedClient->flags |= CLIENT_MODULE; moduleFreeContextReusedClient->flags |= CLIENT_MODULE;
moduleFreeContextReusedClient->user = NULL; /* root user. */ moduleFreeContextReusedClient->user = NULL; /* root user. */
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -35,6 +35,7 @@ ...@@ -35,6 +35,7 @@
#include <stdio.h> #include <stdio.h>
#include <stdint.h> #include <stdint.h>
#include "sds.h" #include "sds.h"
#include "connection.h"
#define RIO_FLAG_READ_ERROR (1<<0) #define RIO_FLAG_READ_ERROR (1<<0)
#define RIO_FLAG_WRITE_ERROR (1<<1) #define RIO_FLAG_WRITE_ERROR (1<<1)
...@@ -76,22 +77,22 @@ struct _rio { ...@@ -76,22 +77,22 @@ struct _rio {
off_t buffered; /* Bytes written since last fsync. */ off_t buffered; /* Bytes written since last fsync. */
off_t autosync; /* fsync after 'autosync' bytes written. */ off_t autosync; /* fsync after 'autosync' bytes written. */
} file; } file;
/* file descriptor */ /* Connection object */
struct { struct {
int fd; /* File descriptor. */ connection *conn; /* Connection */
off_t pos; /* pos in buf that was returned */ off_t pos; /* pos in buf that was returned */
sds buf; /* buffered data */ sds buf; /* buffered data */
size_t read_limit; /* don't allow to buffer/read more than that */ size_t read_limit; /* don't allow to buffer/read more than that */
size_t read_so_far; /* amount of data read from the rio (not buffered) */ size_t read_so_far; /* amount of data read from the rio (not buffered) */
} fd; } conn;
/* Multiple FDs target (used to write to N sockets). */ /* Multiple FDs target (used to write to N sockets). */
struct { struct {
int *fds; /* File descriptors. */ connection **conns; /* Connections */
int *state; /* Error state of each fd. 0 (if ok) or errno. */ int *state; /* Error state of each fd. 0 (if ok) or errno. */
int numfds; int numconns;
off_t pos; off_t pos;
sds buf; sds buf;
} fdset; } connset;
} io; } io;
}; };
...@@ -159,11 +160,11 @@ static inline void rioClearErrors(rio *r) { ...@@ -159,11 +160,11 @@ static inline void rioClearErrors(rio *r) {
void rioInitWithFile(rio *r, FILE *fp); void rioInitWithFile(rio *r, FILE *fp);
void rioInitWithBuffer(rio *r, sds s); void rioInitWithBuffer(rio *r, sds s);
void rioInitWithFd(rio *r, int fd, size_t read_limit); void rioInitWithConn(rio *r, connection *conn, size_t read_limit);
void rioInitWithFdset(rio *r, int *fds, int numfds); void rioInitWithConnset(rio *r, connection **conns, int numconns);
void rioFreeFdset(rio *r); void rioFreeConnset(rio *r);
void rioFreeFd(rio *r, sds* out_remainingBufferedData); void rioFreeConn(rio *r, sds* out_remainingBufferedData);
size_t rioWriteBulkCount(rio *r, char prefix, long count); size_t rioWriteBulkCount(rio *r, char prefix, long count);
size_t rioWriteBulkString(rio *r, const char *buf, size_t len); size_t rioWriteBulkString(rio *r, const char *buf, size_t len);
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -8,6 +8,7 @@ source ../instances.tcl ...@@ -8,6 +8,7 @@ source ../instances.tcl
source ../../support/cluster.tcl ; # Redis Cluster client. source ../../support/cluster.tcl ; # Redis Cluster client.
set ::instances_count 20 ; # How many instances we use at max. set ::instances_count 20 ; # How many instances we use at max.
set ::tlsdir "../../tls"
proc main {} { proc main {} {
parse_options parse_options
......
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