Unverified Commit 2968d8e3 authored by Salvatore Sanfilippo's avatar Salvatore Sanfilippo Committed by GitHub
Browse files

Merge branch 'unstable' into ModuleSecurity

parents ff682d79 c6fb9d09
...@@ -77,6 +77,15 @@ FINAL_LDFLAGS=$(LDFLAGS) $(REDIS_LDFLAGS) $(DEBUG) ...@@ -77,6 +77,15 @@ FINAL_LDFLAGS=$(LDFLAGS) $(REDIS_LDFLAGS) $(DEBUG)
FINAL_LIBS=-lm FINAL_LIBS=-lm
DEBUG=-g -ggdb DEBUG=-g -ggdb
# Linux ARM needs -latomic at linking time
ifneq (,$(filter aarch64 armv,$(uname_M)))
FINAL_LIBS+=-latomic
else
ifneq (,$(findstring armv,$(uname_M)))
FINAL_LIBS+=-latomic
endif
endif
ifeq ($(uname_S),SunOS) ifeq ($(uname_S),SunOS)
# SunOS # SunOS
ifneq ($(@@),32bit) ifneq ($(@@),32bit)
...@@ -93,6 +102,8 @@ else ...@@ -93,6 +102,8 @@ else
ifeq ($(uname_S),Darwin) ifeq ($(uname_S),Darwin)
# Darwin # Darwin
FINAL_LIBS+= -ldl FINAL_LIBS+= -ldl
OPENSSL_CFLAGS=-I/usr/local/opt/openssl/include
OPENSSL_LDFLAGS=-L/usr/local/opt/openssl/lib
else else
ifeq ($(uname_S),AIX) ifeq ($(uname_S),AIX)
# AIX # AIX
...@@ -145,6 +156,12 @@ ifeq ($(MALLOC),jemalloc) ...@@ -145,6 +156,12 @@ ifeq ($(MALLOC),jemalloc)
FINAL_LIBS := ../deps/jemalloc/lib/libjemalloc.a $(FINAL_LIBS) FINAL_LIBS := ../deps/jemalloc/lib/libjemalloc.a $(FINAL_LIBS)
endif endif
ifeq ($(BUILD_TLS),yes)
FINAL_CFLAGS+=-DUSE_OPENSSL $(OPENSSL_CFLAGS)
FINAL_LDFLAGS+=$(OPENSSL_LDFLAGS)
FINAL_LIBS += ../deps/hiredis/libhiredis_ssl.a -lssl -lcrypto
endif
REDIS_CC=$(QUIET_CC)$(CC) $(FINAL_CFLAGS) REDIS_CC=$(QUIET_CC)$(CC) $(FINAL_CFLAGS)
REDIS_LD=$(QUIET_LINK)$(CC) $(FINAL_LDFLAGS) REDIS_LD=$(QUIET_LINK)$(CC) $(FINAL_LDFLAGS)
REDIS_INSTALL=$(QUIET_INSTALL)$(INSTALL) REDIS_INSTALL=$(QUIET_INSTALL)$(INSTALL)
...@@ -164,7 +181,7 @@ endif ...@@ -164,7 +181,7 @@ endif
REDIS_SERVER_NAME=redis-server REDIS_SERVER_NAME=redis-server
REDIS_SENTINEL_NAME=redis-sentinel REDIS_SENTINEL_NAME=redis-sentinel
REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o acl.o gopher.o tracking.o REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o gopher.o tracking.o connection.o tls.o sha256.o
REDIS_CLI_NAME=redis-cli REDIS_CLI_NAME=redis-cli
REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o anet.o ae.o crc64.o siphash.o crc16.o REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o anet.o ae.o crc64.o siphash.o crc16.o
REDIS_BENCHMARK_NAME=redis-benchmark REDIS_BENCHMARK_NAME=redis-benchmark
......
...@@ -28,6 +28,7 @@ ...@@ -28,6 +28,7 @@
*/ */
#include "server.h" #include "server.h"
#include "sha256.h"
#include <fcntl.h> #include <fcntl.h>
/* ============================================================================= /* =============================================================================
...@@ -93,6 +94,9 @@ void ACLResetSubcommandsForCommand(user *u, unsigned long id); ...@@ -93,6 +94,9 @@ void ACLResetSubcommandsForCommand(user *u, unsigned long id);
void ACLResetSubcommands(user *u); void ACLResetSubcommands(user *u);
void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub); void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub);
/* The length of the string representation of a hashed password. */
#define HASH_PASSWORD_LEN SHA256_BLOCK_SIZE*2
/* ============================================================================= /* =============================================================================
* Helper functions for the rest of the ACL implementation * Helper functions for the rest of the ACL implementation
* ==========================================================================*/ * ==========================================================================*/
...@@ -139,6 +143,25 @@ int time_independent_strcmp(char *a, char *b) { ...@@ -139,6 +143,25 @@ int time_independent_strcmp(char *a, char *b) {
return diff; /* If zero strings are the same. */ return diff; /* If zero strings are the same. */
} }
/* Given an SDS string, returns the SHA256 hex representation as a
* new SDS string. */
sds ACLHashPassword(unsigned char *cleartext, size_t len) {
SHA256_CTX ctx;
unsigned char hash[SHA256_BLOCK_SIZE];
char hex[HASH_PASSWORD_LEN];
char *cset = "0123456789abcdef";
sha256_init(&ctx);
sha256_update(&ctx,(unsigned char*)cleartext,len);
sha256_final(&ctx,hash);
for (int j = 0; j < SHA256_BLOCK_SIZE; j++) {
hex[j*2] = cset[((hash[j]&0xF0)>>4)];
hex[j*2+1] = cset[(hash[j]&0xF)];
}
return sdsnewlen(hex,HASH_PASSWORD_LEN);
}
/* ============================================================================= /* =============================================================================
* Low level ACL API * Low level ACL API
* ==========================================================================*/ * ==========================================================================*/
...@@ -502,7 +525,7 @@ sds ACLDescribeUser(user *u) { ...@@ -502,7 +525,7 @@ sds ACLDescribeUser(user *u) {
listRewind(u->passwords,&li); listRewind(u->passwords,&li);
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
sds thispass = listNodeValue(ln); sds thispass = listNodeValue(ln);
res = sdscatlen(res,">",1); res = sdscatlen(res,"#",1);
res = sdscatsds(res,thispass); res = sdscatsds(res,thispass);
res = sdscatlen(res," ",1); res = sdscatlen(res," ",1);
} }
...@@ -629,7 +652,14 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) { ...@@ -629,7 +652,14 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
* ><password> Add this password to the list of valid password for the user. * ><password> Add this password to the list of valid password for the user.
* For example >mypass will add "mypass" to the list. * For example >mypass will add "mypass" to the list.
* This directive clears the "nopass" flag (see later). * This directive clears the "nopass" flag (see later).
* #<hash> Add this password hash to the list of valid hashes for
* the user. This is useful if you have previously computed
* the hash, and don't want to store it in plaintext.
* This directive clears the "nopass" flag (see later).
* <<password> Remove this password from the list of valid passwords. * <<password> Remove this password from the list of valid passwords.
* !<hash> Remove this hashed password from the list of valid passwords.
* This is useful when you want to remove a password just by
* hash without knowing its plaintext version at all.
* nopass All the set passwords of the user are removed, and the user * nopass All the set passwords of the user are removed, and the user
* is flagged as requiring no password: it means that every * is flagged as requiring no password: it means that every
* password will work against this user. If this directive is * password will work against this user. If this directive is
...@@ -665,6 +695,7 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) { ...@@ -665,6 +695,7 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
* EEXIST: You are adding a key pattern after "*" was already added. This is * EEXIST: You are adding a key pattern after "*" was already added. This is
* almost surely an error on the user side. * almost surely an error on the user side.
* ENODEV: The password you are trying to remove from the user does not exist. * ENODEV: The password you are trying to remove from the user does not exist.
* EBADMSG: The hash you are trying to add is not a valid hash.
*/ */
int ACLSetUser(user *u, const char *op, ssize_t oplen) { int ACLSetUser(user *u, const char *op, ssize_t oplen) {
if (oplen == -1) oplen = strlen(op); if (oplen == -1) oplen = strlen(op);
...@@ -700,14 +731,48 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) { ...@@ -700,14 +731,48 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
} else if (!strcasecmp(op,"resetpass")) { } else if (!strcasecmp(op,"resetpass")) {
u->flags &= ~USER_FLAG_NOPASS; u->flags &= ~USER_FLAG_NOPASS;
listEmpty(u->passwords); listEmpty(u->passwords);
} else if (op[0] == '>') { } else if (op[0] == '>' || op[0] == '#') {
sds newpass = sdsnewlen(op+1,oplen-1); sds newpass;
if (op[0] == '>') {
newpass = ACLHashPassword((unsigned char*)op+1,oplen-1);
} else {
if (oplen != HASH_PASSWORD_LEN + 1) {
errno = EBADMSG;
return C_ERR;
}
/* Password hashes can only be characters that represent
* hexadecimal values, which are numbers and lowercase
* characters 'a' through 'f'.
*/
for(int i = 1; i < HASH_PASSWORD_LEN + 1; i++) {
char c = op[i];
if ((c < 'a' || c > 'f') && (c < '0' || c > '9')) {
errno = EBADMSG;
return C_ERR;
}
}
newpass = sdsnewlen(op+1,oplen-1);
}
listNode *ln = listSearchKey(u->passwords,newpass); listNode *ln = listSearchKey(u->passwords,newpass);
/* Avoid re-adding the same password multiple times. */ /* Avoid re-adding the same password multiple times. */
if (ln == NULL) listAddNodeTail(u->passwords,newpass); if (ln == NULL)
listAddNodeTail(u->passwords,newpass);
else
sdsfree(newpass);
u->flags &= ~USER_FLAG_NOPASS; u->flags &= ~USER_FLAG_NOPASS;
} else if (op[0] == '<') { } else if (op[0] == '<' || op[0] == '!') {
sds delpass = sdsnewlen(op+1,oplen-1); sds delpass;
if (op[0] == '<') {
delpass = ACLHashPassword((unsigned char*)op+1,oplen-1);
} else {
if (oplen != HASH_PASSWORD_LEN + 1) {
errno = EBADMSG;
return C_ERR;
}
delpass = sdsnewlen(op+1,oplen-1);
}
listNode *ln = listSearchKey(u->passwords,delpass); listNode *ln = listSearchKey(u->passwords,delpass);
sdsfree(delpass); sdsfree(delpass);
if (ln) { if (ln) {
...@@ -724,7 +789,10 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) { ...@@ -724,7 +789,10 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
sds newpat = sdsnewlen(op+1,oplen-1); sds newpat = sdsnewlen(op+1,oplen-1);
listNode *ln = listSearchKey(u->patterns,newpat); listNode *ln = listSearchKey(u->patterns,newpat);
/* Avoid re-adding the same pattern multiple times. */ /* Avoid re-adding the same pattern multiple times. */
if (ln == NULL) listAddNodeTail(u->patterns,newpat); if (ln == NULL)
listAddNodeTail(u->patterns,newpat);
else
sdsfree(newpat);
u->flags &= ~USER_FLAG_ALLKEYS; u->flags &= ~USER_FLAG_ALLKEYS;
} else if (op[0] == '+' && op[1] != '@') { } else if (op[0] == '+' && op[1] != '@') {
if (strchr(op,'|') == NULL) { if (strchr(op,'|') == NULL) {
...@@ -822,6 +890,9 @@ char *ACLSetUserStringError(void) { ...@@ -822,6 +890,9 @@ char *ACLSetUserStringError(void) {
else if (errno == ENODEV) else if (errno == ENODEV)
errmsg = "The password you are trying to remove from the user does " errmsg = "The password you are trying to remove from the user does "
"not exist"; "not exist";
else if (errno == EBADMSG)
errmsg = "The password hash must be exactly 64 characters and contain "
"only lowercase hexadecimal characters";
return errmsg; return errmsg;
} }
...@@ -879,11 +950,15 @@ int ACLCheckUserCredentials(robj *username, robj *password) { ...@@ -879,11 +950,15 @@ int ACLCheckUserCredentials(robj *username, robj *password) {
listIter li; listIter li;
listNode *ln; listNode *ln;
listRewind(u->passwords,&li); listRewind(u->passwords,&li);
sds hashed = ACLHashPassword(password->ptr,sdslen(password->ptr));
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
sds thispass = listNodeValue(ln); sds thispass = listNodeValue(ln);
if (!time_independent_strcmp(password->ptr, thispass)) if (!time_independent_strcmp(hashed, thispass)) {
sdsfree(hashed);
return C_OK; return C_OK;
}
} }
sdsfree(hashed);
/* If we reached this point, no password matched. */ /* If we reached this point, no password matched. */
errno = EINVAL; errno = EINVAL;
......
...@@ -66,7 +66,7 @@ typedef struct list { ...@@ -66,7 +66,7 @@ typedef struct list {
#define listSetMatchMethod(l,m) ((l)->match = (m)) #define listSetMatchMethod(l,m) ((l)->match = (m))
#define listGetDupMethod(l) ((l)->dup) #define listGetDupMethod(l) ((l)->dup)
#define listGetFree(l) ((l)->free) #define listGetFreeMethod(l) ((l)->free)
#define listGetMatchMethod(l) ((l)->match) #define listGetMatchMethod(l) ((l)->match)
/* Prototypes */ /* Prototypes */
......
...@@ -76,6 +76,7 @@ aeEventLoop *aeCreateEventLoop(int setsize) { ...@@ -76,6 +76,7 @@ aeEventLoop *aeCreateEventLoop(int setsize) {
eventLoop->maxfd = -1; eventLoop->maxfd = -1;
eventLoop->beforesleep = NULL; eventLoop->beforesleep = NULL;
eventLoop->aftersleep = NULL; eventLoop->aftersleep = NULL;
eventLoop->flags = 0;
if (aeApiCreate(eventLoop) == -1) goto err; if (aeApiCreate(eventLoop) == -1) goto err;
/* Events with mask == AE_NONE are not set. So let's initialize the /* Events with mask == AE_NONE are not set. So let's initialize the
* vector with it. */ * vector with it. */
...@@ -97,6 +98,14 @@ int aeGetSetSize(aeEventLoop *eventLoop) { ...@@ -97,6 +98,14 @@ int aeGetSetSize(aeEventLoop *eventLoop) {
return eventLoop->setsize; return eventLoop->setsize;
} }
/* Tells the next iteration/s of the event processing to set timeout of 0. */
void aeSetDontWait(aeEventLoop *eventLoop, int noWait) {
if (noWait)
eventLoop->flags |= AE_DONT_WAIT;
else
eventLoop->flags &= ~AE_DONT_WAIT;
}
/* Resize the maximum set size of the event loop. /* Resize the maximum set size of the event loop.
* If the requested set size is smaller than the current set size, but * If the requested set size is smaller than the current set size, but
* there is already a file descriptor in use that is >= the requested * there is already a file descriptor in use that is >= the requested
...@@ -406,6 +415,11 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags) ...@@ -406,6 +415,11 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
} }
} }
if (eventLoop->flags & AE_DONT_WAIT) {
tv.tv_sec = tv.tv_usec = 0;
tvp = &tv;
}
/* Call the multiplexing API, will return only on timeout or when /* Call the multiplexing API, will return only on timeout or when
* some event fires. */ * some event fires. */
numevents = aeApiPoll(eventLoop, tvp); numevents = aeApiPoll(eventLoop, tvp);
......
...@@ -106,6 +106,7 @@ typedef struct aeEventLoop { ...@@ -106,6 +106,7 @@ typedef struct aeEventLoop {
void *apidata; /* This is used for polling API specific data */ void *apidata; /* This is used for polling API specific data */
aeBeforeSleepProc *beforesleep; aeBeforeSleepProc *beforesleep;
aeBeforeSleepProc *aftersleep; aeBeforeSleepProc *aftersleep;
int flags;
} aeEventLoop; } aeEventLoop;
/* Prototypes */ /* Prototypes */
...@@ -128,5 +129,6 @@ void aeSetBeforeSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *beforesleep ...@@ -128,5 +129,6 @@ void aeSetBeforeSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *beforesleep
void aeSetAfterSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *aftersleep); void aeSetAfterSleepProc(aeEventLoop *eventLoop, aeBeforeSleepProc *aftersleep);
int aeGetSetSize(aeEventLoop *eventLoop); int aeGetSetSize(aeEventLoop *eventLoop);
int aeResizeSetSize(aeEventLoop *eventLoop, int setsize); int aeResizeSetSize(aeEventLoop *eventLoop, int setsize);
void aeSetDontWait(aeEventLoop *eventLoop, int noWait);
#endif #endif
...@@ -121,8 +121,8 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) { ...@@ -121,8 +121,8 @@ static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) {
if (e->events & EPOLLIN) mask |= AE_READABLE; if (e->events & EPOLLIN) mask |= AE_READABLE;
if (e->events & EPOLLOUT) mask |= AE_WRITABLE; if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
if (e->events & EPOLLERR) mask |= AE_WRITABLE; if (e->events & EPOLLERR) mask |= AE_WRITABLE|AE_READABLE;
if (e->events & EPOLLHUP) mask |= AE_WRITABLE; if (e->events & EPOLLHUP) mask |= AE_WRITABLE|AE_READABLE;
eventLoop->fired[j].fd = e->data.fd; eventLoop->fired[j].fd = e->data.fd;
eventLoop->fired[j].mask = mask; eventLoop->fired[j].mask = mask;
} }
......
...@@ -279,8 +279,8 @@ static int anetCreateSocket(char *err, int domain) { ...@@ -279,8 +279,8 @@ static int anetCreateSocket(char *err, int domain) {
#define ANET_CONNECT_NONE 0 #define ANET_CONNECT_NONE 0
#define ANET_CONNECT_NONBLOCK 1 #define ANET_CONNECT_NONBLOCK 1
#define ANET_CONNECT_BE_BINDING 2 /* Best effort binding. */ #define ANET_CONNECT_BE_BINDING 2 /* Best effort binding. */
static int anetTcpGenericConnect(char *err, char *addr, int port, static int anetTcpGenericConnect(char *err, const char *addr, int port,
char *source_addr, int flags) const char *source_addr, int flags)
{ {
int s = ANET_ERR, rv; int s = ANET_ERR, rv;
char portstr[6]; /* strlen("65535") + 1; */ char portstr[6]; /* strlen("65535") + 1; */
...@@ -359,31 +359,31 @@ end: ...@@ -359,31 +359,31 @@ end:
} }
} }
int anetTcpConnect(char *err, char *addr, int port) int anetTcpConnect(char *err, const char *addr, int port)
{ {
return anetTcpGenericConnect(err,addr,port,NULL,ANET_CONNECT_NONE); return anetTcpGenericConnect(err,addr,port,NULL,ANET_CONNECT_NONE);
} }
int anetTcpNonBlockConnect(char *err, char *addr, int port) int anetTcpNonBlockConnect(char *err, const char *addr, int port)
{ {
return anetTcpGenericConnect(err,addr,port,NULL,ANET_CONNECT_NONBLOCK); return anetTcpGenericConnect(err,addr,port,NULL,ANET_CONNECT_NONBLOCK);
} }
int anetTcpNonBlockBindConnect(char *err, char *addr, int port, int anetTcpNonBlockBindConnect(char *err, const char *addr, int port,
char *source_addr) const char *source_addr)
{ {
return anetTcpGenericConnect(err,addr,port,source_addr, return anetTcpGenericConnect(err,addr,port,source_addr,
ANET_CONNECT_NONBLOCK); ANET_CONNECT_NONBLOCK);
} }
int anetTcpNonBlockBestEffortBindConnect(char *err, char *addr, int port, int anetTcpNonBlockBestEffortBindConnect(char *err, const char *addr, int port,
char *source_addr) const char *source_addr)
{ {
return anetTcpGenericConnect(err,addr,port,source_addr, return anetTcpGenericConnect(err,addr,port,source_addr,
ANET_CONNECT_NONBLOCK|ANET_CONNECT_BE_BINDING); ANET_CONNECT_NONBLOCK|ANET_CONNECT_BE_BINDING);
} }
int anetUnixGenericConnect(char *err, char *path, int flags) int anetUnixGenericConnect(char *err, const char *path, int flags)
{ {
int s; int s;
struct sockaddr_un sa; struct sockaddr_un sa;
...@@ -411,12 +411,12 @@ int anetUnixGenericConnect(char *err, char *path, int flags) ...@@ -411,12 +411,12 @@ int anetUnixGenericConnect(char *err, char *path, int flags)
return s; return s;
} }
int anetUnixConnect(char *err, char *path) int anetUnixConnect(char *err, const char *path)
{ {
return anetUnixGenericConnect(err,path,ANET_CONNECT_NONE); return anetUnixGenericConnect(err,path,ANET_CONNECT_NONE);
} }
int anetUnixNonBlockConnect(char *err, char *path) int anetUnixNonBlockConnect(char *err, const char *path)
{ {
return anetUnixGenericConnect(err,path,ANET_CONNECT_NONBLOCK); return anetUnixGenericConnect(err,path,ANET_CONNECT_NONBLOCK);
} }
......
...@@ -49,12 +49,12 @@ ...@@ -49,12 +49,12 @@
#undef ip_len #undef ip_len
#endif #endif
int anetTcpConnect(char *err, char *addr, int port); int anetTcpConnect(char *err, const char *addr, int port);
int anetTcpNonBlockConnect(char *err, char *addr, int port); int anetTcpNonBlockConnect(char *err, const char *addr, int port);
int anetTcpNonBlockBindConnect(char *err, char *addr, int port, char *source_addr); int anetTcpNonBlockBindConnect(char *err, const char *addr, int port, const char *source_addr);
int anetTcpNonBlockBestEffortBindConnect(char *err, char *addr, int port, char *source_addr); int anetTcpNonBlockBestEffortBindConnect(char *err, const char *addr, int port, const char *source_addr);
int anetUnixConnect(char *err, char *path); int anetUnixConnect(char *err, const char *path);
int anetUnixNonBlockConnect(char *err, char *path); int anetUnixNonBlockConnect(char *err, const char *path);
int anetRead(int fd, char *buf, int count); int anetRead(int fd, char *buf, int count);
int anetResolve(char *err, char *host, char *ipbuf, size_t ipbuf_len); int anetResolve(char *err, char *host, char *ipbuf, size_t ipbuf_len);
int anetResolveIP(char *err, char *host, char *ipbuf, size_t ipbuf_len); int anetResolveIP(char *err, char *host, char *ipbuf, size_t ipbuf_len);
......
...@@ -264,9 +264,9 @@ int startAppendOnly(void) { ...@@ -264,9 +264,9 @@ int startAppendOnly(void) {
strerror(errno)); strerror(errno));
return C_ERR; return C_ERR;
} }
if (server.rdb_child_pid != -1) { if (hasActiveChildProcess() && server.aof_child_pid == -1) {
server.aof_rewrite_scheduled = 1; server.aof_rewrite_scheduled = 1;
serverLog(LL_WARNING,"AOF was enabled but there is already a child process saving an RDB file on disk. An AOF background was scheduled to start when possible."); serverLog(LL_WARNING,"AOF was enabled but there is already another background operation. An AOF background was scheduled to start when possible.");
} else { } else {
/* If there is a pending AOF rewrite, we need to switch it off and /* If there is a pending AOF rewrite, we need to switch it off and
* start a new one: the old one cannot be reused because it is not * start a new one: the old one cannot be reused because it is not
...@@ -385,6 +385,10 @@ void flushAppendOnlyFile(int force) { ...@@ -385,6 +385,10 @@ void flushAppendOnlyFile(int force) {
* there is much to do about the whole server stopping for power problems * there is much to do about the whole server stopping for power problems
* or alike */ * or alike */
if (server.aof_flush_sleep && sdslen(server.aof_buf)) {
usleep(server.aof_flush_sleep);
}
latencyStartMonitor(latency); latencyStartMonitor(latency);
nwritten = aofWrite(server.aof_fd,server.aof_buf,sdslen(server.aof_buf)); nwritten = aofWrite(server.aof_fd,server.aof_buf,sdslen(server.aof_buf));
latencyEndMonitor(latency); latencyEndMonitor(latency);
...@@ -395,7 +399,7 @@ void flushAppendOnlyFile(int force) { ...@@ -395,7 +399,7 @@ void flushAppendOnlyFile(int force) {
* useful for graphing / monitoring purposes. */ * useful for graphing / monitoring purposes. */
if (sync_in_progress) { if (sync_in_progress) {
latencyAddSampleIfNeeded("aof-write-pending-fsync",latency); latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
} else if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) { } else if (hasActiveChildProcess()) {
latencyAddSampleIfNeeded("aof-write-active-child",latency); latencyAddSampleIfNeeded("aof-write-active-child",latency);
} else { } else {
latencyAddSampleIfNeeded("aof-write-alone",latency); latencyAddSampleIfNeeded("aof-write-alone",latency);
...@@ -491,9 +495,8 @@ void flushAppendOnlyFile(int force) { ...@@ -491,9 +495,8 @@ void flushAppendOnlyFile(int force) {
try_fsync: try_fsync:
/* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
* children doing I/O in the background. */ * children doing I/O in the background. */
if (server.aof_no_fsync_on_rewrite && if (server.aof_no_fsync_on_rewrite && hasActiveChildProcess())
(server.aof_child_pid != -1 || server.rdb_child_pid != -1)) return;
return;
/* Perform the fsync if needed. */ /* Perform the fsync if needed. */
if (server.aof_fsync == AOF_FSYNC_ALWAYS) { if (server.aof_fsync == AOF_FSYNC_ALWAYS) {
...@@ -649,11 +652,12 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a ...@@ -649,11 +652,12 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
/* In Redis commands are always executed in the context of a client, so in /* In Redis commands are always executed in the context of a client, so in
* order to load the append only file we need to create a fake client. */ * order to load the append only file we need to create a fake client. */
struct client *createFakeClient(void) { struct client *createAOFClient(void) {
struct client *c = zmalloc(sizeof(*c)); struct client *c = zmalloc(sizeof(*c));
selectDb(c,0); selectDb(c,0);
c->fd = -1; c->id = CLIENT_ID_AOF; /* So modules can identify it's the AOF client. */
c->conn = NULL;
c->name = NULL; c->name = NULL;
c->querybuf = sdsempty(); c->querybuf = sdsempty();
c->querybuf_peak = 0; c->querybuf_peak = 0;
...@@ -726,8 +730,8 @@ int loadAppendOnlyFile(char *filename) { ...@@ -726,8 +730,8 @@ int loadAppendOnlyFile(char *filename) {
* to the same file we're about to read. */ * to the same file we're about to read. */
server.aof_state = AOF_OFF; server.aof_state = AOF_OFF;
fakeClient = createFakeClient(); fakeClient = createAOFClient();
startLoadingFile(fp, filename); startLoadingFile(fp, filename, RDBFLAGS_AOF_PREAMBLE);
/* Check if this AOF file has an RDB preamble. In that case we need to /* Check if this AOF file has an RDB preamble. In that case we need to
* load the RDB file and later continue loading the AOF tail. */ * load the RDB file and later continue loading the AOF tail. */
...@@ -742,7 +746,7 @@ int loadAppendOnlyFile(char *filename) { ...@@ -742,7 +746,7 @@ int loadAppendOnlyFile(char *filename) {
serverLog(LL_NOTICE,"Reading RDB preamble from AOF file..."); serverLog(LL_NOTICE,"Reading RDB preamble from AOF file...");
if (fseek(fp,0,SEEK_SET) == -1) goto readerr; if (fseek(fp,0,SEEK_SET) == -1) goto readerr;
rioInitWithFile(&rdb,fp); rioInitWithFile(&rdb,fp);
if (rdbLoadRio(&rdb,NULL,1) != C_OK) { if (rdbLoadRio(&rdb,RDBFLAGS_AOF_PREAMBLE,NULL) != C_OK) {
serverLog(LL_WARNING,"Error reading the RDB preamble of the AOF file, AOF loading aborted"); serverLog(LL_WARNING,"Error reading the RDB preamble of the AOF file, AOF loading aborted");
goto readerr; goto readerr;
} else { } else {
...@@ -763,6 +767,7 @@ int loadAppendOnlyFile(char *filename) { ...@@ -763,6 +767,7 @@ int loadAppendOnlyFile(char *filename) {
if (!(loops++ % 1000)) { if (!(loops++ % 1000)) {
loadingProgress(ftello(fp)); loadingProgress(ftello(fp));
processEventsWhileBlocked(); processEventsWhileBlocked();
processModuleLoadingProgressEvent(1);
} }
if (fgets(buf,sizeof(buf),fp) == NULL) { if (fgets(buf,sizeof(buf),fp) == NULL) {
...@@ -836,6 +841,8 @@ int loadAppendOnlyFile(char *filename) { ...@@ -836,6 +841,8 @@ int loadAppendOnlyFile(char *filename) {
freeFakeClientArgv(fakeClient); freeFakeClientArgv(fakeClient);
fakeClient->cmd = NULL; fakeClient->cmd = NULL;
if (server.aof_load_truncated) valid_up_to = ftello(fp); if (server.aof_load_truncated) valid_up_to = ftello(fp);
if (server.key_load_delay)
usleep(server.key_load_delay);
} }
/* This point can only be reached when EOF is reached without errors. /* This point can only be reached when EOF is reached without errors.
...@@ -853,7 +860,7 @@ loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */ ...@@ -853,7 +860,7 @@ loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */
fclose(fp); fclose(fp);
freeFakeClient(fakeClient); freeFakeClient(fakeClient);
server.aof_state = old_aof_state; server.aof_state = old_aof_state;
stopLoading(); stopLoading(1);
aofUpdateCurrentSize(); aofUpdateCurrentSize();
server.aof_rewrite_base_size = server.aof_current_size; server.aof_rewrite_base_size = server.aof_current_size;
server.aof_fsync_offset = server.aof_current_size; server.aof_fsync_offset = server.aof_current_size;
...@@ -862,6 +869,7 @@ loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */ ...@@ -862,6 +869,7 @@ loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */
readerr: /* Read error. If feof(fp) is true, fall through to unexpected EOF. */ readerr: /* Read error. If feof(fp) is true, fall through to unexpected EOF. */
if (!feof(fp)) { if (!feof(fp)) {
if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */ if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */
fclose(fp);
serverLog(LL_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno)); serverLog(LL_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno));
exit(1); exit(1);
} }
...@@ -892,11 +900,13 @@ uxeof: /* Unexpected AOF end of file. */ ...@@ -892,11 +900,13 @@ uxeof: /* Unexpected AOF end of file. */
} }
} }
if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */ if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */
fclose(fp);
serverLog(LL_WARNING,"Unexpected end of file reading the append only file. You can: 1) Make a backup of your AOF file, then use ./redis-check-aof --fix <filename>. 2) Alternatively you can set the 'aof-load-truncated' configuration option to yes and restart the server."); serverLog(LL_WARNING,"Unexpected end of file reading the append only file. You can: 1) Make a backup of your AOF file, then use ./redis-check-aof --fix <filename>. 2) Alternatively you can set the 'aof-load-truncated' configuration option to yes and restart the server.");
exit(1); exit(1);
fmterr: /* Format error. */ fmterr: /* Format error. */
if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */ if (fakeClient) freeFakeClient(fakeClient); /* avoid valgrind warning */
fclose(fp);
serverLog(LL_WARNING,"Bad file format reading the append only file: make a backup of your AOF file, then use ./redis-check-aof --fix <filename>"); serverLog(LL_WARNING,"Bad file format reading the append only file: make a backup of your AOF file, then use ./redis-check-aof --fix <filename>");
exit(1); exit(1);
} }
...@@ -1391,9 +1401,11 @@ int rewriteAppendOnlyFile(char *filename) { ...@@ -1391,9 +1401,11 @@ int rewriteAppendOnlyFile(char *filename) {
if (server.aof_rewrite_incremental_fsync) if (server.aof_rewrite_incremental_fsync)
rioSetAutoSync(&aof,REDIS_AUTOSYNC_BYTES); rioSetAutoSync(&aof,REDIS_AUTOSYNC_BYTES);
startSaving(RDBFLAGS_AOF_PREAMBLE);
if (server.aof_use_rdb_preamble) { if (server.aof_use_rdb_preamble) {
int error; int error;
if (rdbSaveRio(&aof,&error,RDB_SAVE_AOF_PREAMBLE,NULL) == C_ERR) { if (rdbSaveRio(&aof,&error,RDBFLAGS_AOF_PREAMBLE,NULL) == C_ERR) {
errno = error; errno = error;
goto werr; goto werr;
} }
...@@ -1456,15 +1468,18 @@ int rewriteAppendOnlyFile(char *filename) { ...@@ -1456,15 +1468,18 @@ int rewriteAppendOnlyFile(char *filename) {
if (rename(tmpfile,filename) == -1) { if (rename(tmpfile,filename) == -1) {
serverLog(LL_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno)); serverLog(LL_WARNING,"Error moving temp append only file on the final destination: %s", strerror(errno));
unlink(tmpfile); unlink(tmpfile);
stopSaving(0);
return C_ERR; return C_ERR;
} }
serverLog(LL_NOTICE,"SYNC append only file rewrite performed"); serverLog(LL_NOTICE,"SYNC append only file rewrite performed");
stopSaving(1);
return C_OK; return C_OK;
werr: werr:
serverLog(LL_WARNING,"Write error writing append only file on disk: %s", strerror(errno)); serverLog(LL_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
fclose(fp); fclose(fp);
unlink(tmpfile); unlink(tmpfile);
stopSaving(0);
return C_ERR; return C_ERR;
} }
...@@ -1560,39 +1575,24 @@ void aofClosePipes(void) { ...@@ -1560,39 +1575,24 @@ void aofClosePipes(void) {
*/ */
int rewriteAppendOnlyFileBackground(void) { int rewriteAppendOnlyFileBackground(void) {
pid_t childpid; pid_t childpid;
long long start;
if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR; if (hasActiveChildProcess()) return C_ERR;
if (aofCreatePipes() != C_OK) return C_ERR; if (aofCreatePipes() != C_OK) return C_ERR;
openChildInfoPipe(); openChildInfoPipe();
start = ustime(); if ((childpid = redisFork()) == 0) {
if ((childpid = fork()) == 0) {
char tmpfile[256]; char tmpfile[256];
/* Child */ /* Child */
closeListeningSockets(0);
redisSetProcTitle("redis-aof-rewrite"); redisSetProcTitle("redis-aof-rewrite");
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid()); snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
if (rewriteAppendOnlyFile(tmpfile) == C_OK) { if (rewriteAppendOnlyFile(tmpfile) == C_OK) {
size_t private_dirty = zmalloc_get_private_dirty(-1); sendChildCOWInfo(CHILD_INFO_TYPE_AOF, "AOF rewrite");
if (private_dirty) {
serverLog(LL_NOTICE,
"AOF rewrite: %zu MB of memory used by copy-on-write",
private_dirty/(1024*1024));
}
server.child_info_data.cow_size = private_dirty;
sendChildInfo(CHILD_INFO_TYPE_AOF);
exitFromChild(0); exitFromChild(0);
} else { } else {
exitFromChild(1); exitFromChild(1);
} }
} else { } else {
/* Parent */ /* Parent */
server.stat_fork_time = ustime()-start;
server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
if (childpid == -1) { if (childpid == -1) {
closeChildInfoPipe(); closeChildInfoPipe();
serverLog(LL_WARNING, serverLog(LL_WARNING,
...@@ -1606,7 +1606,6 @@ int rewriteAppendOnlyFileBackground(void) { ...@@ -1606,7 +1606,6 @@ int rewriteAppendOnlyFileBackground(void) {
server.aof_rewrite_scheduled = 0; server.aof_rewrite_scheduled = 0;
server.aof_rewrite_time_start = time(NULL); server.aof_rewrite_time_start = time(NULL);
server.aof_child_pid = childpid; server.aof_child_pid = childpid;
updateDictResizePolicy();
/* We set appendseldb to -1 in order to force the next call to the /* We set appendseldb to -1 in order to force the next call to the
* feedAppendOnlyFile() to issue a SELECT command, so the differences * feedAppendOnlyFile() to issue a SELECT command, so the differences
* accumulated by the parent into server.aof_rewrite_buf will start * accumulated by the parent into server.aof_rewrite_buf will start
...@@ -1621,13 +1620,14 @@ int rewriteAppendOnlyFileBackground(void) { ...@@ -1621,13 +1620,14 @@ int rewriteAppendOnlyFileBackground(void) {
void bgrewriteaofCommand(client *c) { void bgrewriteaofCommand(client *c) {
if (server.aof_child_pid != -1) { if (server.aof_child_pid != -1) {
addReplyError(c,"Background append only file rewriting already in progress"); addReplyError(c,"Background append only file rewriting already in progress");
} else if (server.rdb_child_pid != -1) { } else if (hasActiveChildProcess()) {
server.aof_rewrite_scheduled = 1; server.aof_rewrite_scheduled = 1;
addReplyStatus(c,"Background append only file rewriting scheduled"); addReplyStatus(c,"Background append only file rewriting scheduled");
} else if (rewriteAppendOnlyFileBackground() == C_OK) { } else if (rewriteAppendOnlyFileBackground() == C_OK) {
addReplyStatus(c,"Background append only file rewriting started"); addReplyStatus(c,"Background append only file rewriting started");
} else { } else {
addReply(c,shared.err); addReplyError(c,"Can't execute an AOF background rewriting. "
"Please check the server logs for more information.");
} }
} }
...@@ -1766,7 +1766,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) { ...@@ -1766,7 +1766,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
server.aof_selected_db = -1; /* Make sure SELECT is re-issued */ server.aof_selected_db = -1; /* Make sure SELECT is re-issued */
aofUpdateCurrentSize(); aofUpdateCurrentSize();
server.aof_rewrite_base_size = server.aof_current_size; server.aof_rewrite_base_size = server.aof_current_size;
server.aof_current_size = server.aof_current_size; server.aof_fsync_offset = server.aof_current_size;
/* Clear regular AOF buffer since its contents was just written to /* Clear regular AOF buffer since its contents was just written to
* the new AOF from the background rewrite buffer. */ * the new AOF from the background rewrite buffer. */
......
...@@ -27,6 +27,9 @@ ...@@ -27,6 +27,9 @@
* POSSIBILITY OF SUCH DAMAGE. * POSSIBILITY OF SUCH DAMAGE.
*/ */
#ifndef __BIO_H
#define __BIO_H
/* Exported API */ /* Exported API */
void bioInit(void); void bioInit(void);
void bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3); void bioCreateBackgroundJob(int type, void *arg1, void *arg2, void *arg3);
...@@ -40,3 +43,5 @@ void bioKillThreads(void); ...@@ -40,3 +43,5 @@ void bioKillThreads(void);
#define BIO_AOF_FSYNC 1 /* Deferred AOF fsync. */ #define BIO_AOF_FSYNC 1 /* Deferred AOF fsync. */
#define BIO_LAZY_FREE 2 /* Deferred objects freeing. */ #define BIO_LAZY_FREE 2 /* Deferred objects freeing. */
#define BIO_NUM_OPS 3 #define BIO_NUM_OPS 3
#endif
This diff is collapsed.
...@@ -80,6 +80,8 @@ void receiveChildInfo(void) { ...@@ -80,6 +80,8 @@ void receiveChildInfo(void) {
server.stat_rdb_cow_bytes = server.child_info_data.cow_size; server.stat_rdb_cow_bytes = server.child_info_data.cow_size;
} else if (server.child_info_data.process_type == CHILD_INFO_TYPE_AOF) { } else if (server.child_info_data.process_type == CHILD_INFO_TYPE_AOF) {
server.stat_aof_cow_bytes = server.child_info_data.cow_size; server.stat_aof_cow_bytes = server.child_info_data.cow_size;
} else if (server.child_info_data.process_type == CHILD_INFO_TYPE_MODULE) {
server.stat_module_cow_bytes = server.child_info_data.cow_size;
} }
} }
} }
This diff is collapsed.
...@@ -13,15 +13,10 @@ ...@@ -13,15 +13,10 @@
/* The following defines are amount of time, sometimes expressed as /* The following defines are amount of time, sometimes expressed as
* multiplicators of the node timeout value (when ending with MULT). */ * multiplicators of the node timeout value (when ending with MULT). */
#define CLUSTER_DEFAULT_NODE_TIMEOUT 15000
#define CLUSTER_DEFAULT_SLAVE_VALIDITY 10 /* Slave max data age factor. */
#define CLUSTER_DEFAULT_REQUIRE_FULL_COVERAGE 1
#define CLUSTER_DEFAULT_SLAVE_NO_FAILOVER 0 /* Failover by default. */
#define CLUSTER_FAIL_REPORT_VALIDITY_MULT 2 /* Fail report validity. */ #define CLUSTER_FAIL_REPORT_VALIDITY_MULT 2 /* Fail report validity. */
#define CLUSTER_FAIL_UNDO_TIME_MULT 2 /* Undo fail if master is back. */ #define CLUSTER_FAIL_UNDO_TIME_MULT 2 /* Undo fail if master is back. */
#define CLUSTER_FAIL_UNDO_TIME_ADD 10 /* Some additional time. */ #define CLUSTER_FAIL_UNDO_TIME_ADD 10 /* Some additional time. */
#define CLUSTER_FAILOVER_DELAY 5 /* Seconds */ #define CLUSTER_FAILOVER_DELAY 5 /* Seconds */
#define CLUSTER_DEFAULT_MIGRATION_BARRIER 1
#define CLUSTER_MF_TIMEOUT 5000 /* Milliseconds to do a manual failover. */ #define CLUSTER_MF_TIMEOUT 5000 /* Milliseconds to do a manual failover. */
#define CLUSTER_MF_PAUSE_MULT 2 /* Master pause manual failover mult. */ #define CLUSTER_MF_PAUSE_MULT 2 /* Master pause manual failover mult. */
#define CLUSTER_SLAVE_MIGRATION_DELAY 5000 /* Delay for slave migration. */ #define CLUSTER_SLAVE_MIGRATION_DELAY 5000 /* Delay for slave migration. */
...@@ -40,7 +35,7 @@ struct clusterNode; ...@@ -40,7 +35,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 */
......
This diff is collapsed.
/*
* 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.
*/
/* 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 < 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.
*
* The barrier flag indicates a write barrier is requested, resulting with
* CONN_FLAG_WRITE_BARRIER set. This will ensure that the write handler is
* always called before and not after the read handler in a single event
* loop.
*/
static int connSocketSetWriteHandler(connection *conn, ConnectionCallbackFunc func, int barrier) {
if (func == conn->write_handler) return C_OK;
conn->write_handler = func;
if (barrier)
conn->flags |= CONN_FLAG_WRITE_BARRIER;
else
conn->flags &= ~CONN_FLAG_WRITE_BARRIER;
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;
}
/* Normally we execute the readable event first, and the writable
* event later. This is useful as sometimes we may be able
* to serve the reply of a query immediately after processing the
* query.
*
* However if WRITE_BARRIER is set in the mask, our application is
* asking us to do the reverse: never fire the writable event
* after the readable. In such a case, we invert the calls.
* This is useful when, for instance, we want to do things
* in the beforeSleep() hook, like fsync'ing a file to disk,
* before replying to a client. */
int invert = conn->flags & CONN_FLAG_WRITE_BARRIER;
int call_write = (mask & AE_WRITABLE) && conn->write_handler;
int call_read = (mask & AE_READABLE) && conn->read_handler;
/* Handle normal I/O flows */
if (!invert && call_read) {
if (!callHandler(conn, conn->read_handler)) return;
}
/* Fire the writable event. */
if (call_write) {
if (!callHandler(conn, conn->write_handler)) return;
}
/* If we have to invert the call, fire the readable event now
* after the writable one. */
if (invert && call_read) {
if (!callHandler(conn, conn->read_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,
.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;
}
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
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