Unverified Commit 3f14bfd8 authored by Loris Cro's avatar Loris Cro Committed by GitHub
Browse files

Merge pull request #1 from antirez/unstable

update to latest unstable
parents 8ea4bdd9 720d1fd3
...@@ -94,6 +94,9 @@ void ACLResetSubcommandsForCommand(user *u, unsigned long id); ...@@ -94,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
* ==========================================================================*/ * ==========================================================================*/
...@@ -145,7 +148,7 @@ int time_independent_strcmp(char *a, char *b) { ...@@ -145,7 +148,7 @@ int time_independent_strcmp(char *a, char *b) {
sds ACLHashPassword(unsigned char *cleartext, size_t len) { sds ACLHashPassword(unsigned char *cleartext, size_t len) {
SHA256_CTX ctx; SHA256_CTX ctx;
unsigned char hash[SHA256_BLOCK_SIZE]; unsigned char hash[SHA256_BLOCK_SIZE];
char hex[SHA256_BLOCK_SIZE*2]; char hex[HASH_PASSWORD_LEN];
char *cset = "0123456789abcdef"; char *cset = "0123456789abcdef";
sha256_init(&ctx); sha256_init(&ctx);
...@@ -156,7 +159,7 @@ sds ACLHashPassword(unsigned char *cleartext, size_t len) { ...@@ -156,7 +159,7 @@ sds ACLHashPassword(unsigned char *cleartext, size_t len) {
hex[j*2] = cset[((hash[j]&0xF0)>>4)]; hex[j*2] = cset[((hash[j]&0xF0)>>4)];
hex[j*2+1] = cset[(hash[j]&0xF)]; hex[j*2+1] = cset[(hash[j]&0xF)];
} }
return sdsnewlen(hex,SHA256_BLOCK_SIZE*2); return sdsnewlen(hex,HASH_PASSWORD_LEN);
} }
/* ============================================================================= /* =============================================================================
...@@ -522,7 +525,7 @@ sds ACLDescribeUser(user *u) { ...@@ -522,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);
} }
...@@ -649,7 +652,14 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) { ...@@ -649,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
...@@ -685,6 +695,7 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) { ...@@ -685,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);
...@@ -720,8 +731,30 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) { ...@@ -720,8 +731,30 @@ 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 = ACLHashPassword((unsigned char*)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) if (ln == NULL)
...@@ -729,8 +762,17 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) { ...@@ -729,8 +762,17 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
else else
sdsfree(newpass); sdsfree(newpass);
u->flags &= ~USER_FLAG_NOPASS; u->flags &= ~USER_FLAG_NOPASS;
} else if (op[0] == '<') { } else if (op[0] == '<' || op[0] == '!') {
sds delpass = ACLHashPassword((unsigned char*)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) {
...@@ -848,6 +890,9 @@ char *ACLSetUserStringError(void) { ...@@ -848,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;
} }
......
...@@ -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;
...@@ -1394,9 +1401,11 @@ int rewriteAppendOnlyFile(char *filename) { ...@@ -1394,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;
} }
...@@ -1459,15 +1468,18 @@ int rewriteAppendOnlyFile(char *filename) { ...@@ -1459,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;
} }
...@@ -1563,39 +1575,24 @@ void aofClosePipes(void) { ...@@ -1563,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,
...@@ -1609,7 +1606,6 @@ int rewriteAppendOnlyFileBackground(void) { ...@@ -1609,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
...@@ -1624,13 +1620,14 @@ int rewriteAppendOnlyFileBackground(void) { ...@@ -1624,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.");
} }
} }
......
...@@ -174,6 +174,7 @@ void unblockClient(client *c) { ...@@ -174,6 +174,7 @@ void unblockClient(client *c) {
} else if (c->btype == BLOCKED_WAIT) { } else if (c->btype == BLOCKED_WAIT) {
unblockClientWaitingReplicas(c); unblockClientWaitingReplicas(c);
} else if (c->btype == BLOCKED_MODULE) { } else if (c->btype == BLOCKED_MODULE) {
if (moduleClientIsBlockedOnKeys(c)) unblockClientWaitingData(c);
unblockClientFromModule(c); unblockClientFromModule(c);
} else { } else {
serverPanic("Unknown btype in unblockClient()."); serverPanic("Unknown btype in unblockClient().");
...@@ -430,6 +431,49 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) { ...@@ -430,6 +431,49 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
} }
} }
/* Helper function for handleClientsBlockedOnKeys(). This function is called
* in order to check if we can serve clients blocked by modules using
* RM_BlockClientOnKeys(), when the corresponding key was signaled as ready:
* our goal here is to call the RedisModuleBlockedClient reply() callback to
* see if the key is really able to serve the client, and in that case,
* unblock it. */
void serveClientsBlockedOnKeyByModule(readyList *rl) {
dictEntry *de;
/* We serve clients in the same order they blocked for
* this key, from the first blocked to the last. */
de = dictFind(rl->db->blocking_keys,rl->key);
if (de) {
list *clients = dictGetVal(de);
int numclients = listLength(clients);
while(numclients--) {
listNode *clientnode = listFirst(clients);
client *receiver = clientnode->value;
/* Put at the tail, so that at the next call
* we'll not run into it again: clients here may not be
* ready to be served, so they'll remain in the list
* sometimes. We want also be able to skip clients that are
* not blocked for the MODULE type safely. */
listDelNode(clients,clientnode);
listAddNodeTail(clients,receiver);
if (receiver->btype != BLOCKED_MODULE) continue;
/* Note that if *this* client cannot be served by this key,
* it does not mean that another client that is next into the
* list cannot be served as well: they may be blocked by
* different modules with different triggers to consider if a key
* is ready or not. This means we can't exit the loop but need
* to continue after the first failure. */
if (!moduleTryServeClientBlockedOnKey(receiver, rl->key)) continue;
moduleUnblockClient(receiver);
}
}
}
/* This function should be called by Redis every time a single command, /* This function should be called by Redis every time a single command,
* a MULTI/EXEC block, or a Lua script, terminated its execution after * a MULTI/EXEC block, or a Lua script, terminated its execution after
* being called by a client. It handles serving clients blocked in * being called by a client. It handles serving clients blocked in
...@@ -480,6 +524,10 @@ void handleClientsBlockedOnKeys(void) { ...@@ -480,6 +524,10 @@ void handleClientsBlockedOnKeys(void) {
serveClientsBlockedOnSortedSetKey(o,rl); serveClientsBlockedOnSortedSetKey(o,rl);
else if (o->type == OBJ_STREAM) else if (o->type == OBJ_STREAM)
serveClientsBlockedOnStreamKey(o,rl); serveClientsBlockedOnStreamKey(o,rl);
/* We want to serve clients blocked on module keys
* regardless of the object type: we don't know what the
* module is trying to accomplish right now. */
serveClientsBlockedOnKeyByModule(rl);
} }
/* Free this item. */ /* Free this item. */
......
...@@ -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.
...@@ -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 */
......
This diff is collapsed.
This diff is collapsed.
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.
*/
#ifndef __REDIS_CONNHELPERS_H
#define __REDIS_CONNHELPERS_H
#include "connection.h"
/* These are helper functions that are common to different connection
* implementations (currently sockets in connection.c and TLS in tls.c).
*
* Currently helpers implement the mechanisms for invoking connection
* handlers, tracking in-handler states and dealing with deferred
* destruction (if invoked by a handler).
*/
/* Called whenever a handler is invoked on a connection and sets the
* CONN_FLAG_IN_HANDLER flag to indicate we're in a handler context.
*
* An attempt to close a connection while CONN_FLAG_IN_HANDLER is
* set will result with deferred close, i.e. setting the CONN_FLAG_CLOSE_SCHEDULED
* instead of destructing it.
*/
static inline void enterHandler(connection *conn) {
conn->flags |= CONN_FLAG_IN_HANDLER;
}
/* Called whenever a handler returns. This unsets the CONN_FLAG_IN_HANDLER
* flag and performs actual close/destruction if a deferred close was
* scheduled by the 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;
}
/* Helper for connection implementations to call handlers:
* 1. Mark the handler in use.
* 2. Execute the handler (if set).
* 3. Mark the handler as NOT in use and perform deferred close if was
* requested by the handler at any time.
*/
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 */
...@@ -60,10 +60,7 @@ robj *lookupKey(redisDb *db, robj *key, int flags) { ...@@ -60,10 +60,7 @@ robj *lookupKey(redisDb *db, robj *key, int flags) {
/* Update the access time for the ageing algorithm. /* Update the access time for the ageing algorithm.
* Don't do it if we have a saving child, as this will trigger * Don't do it if we have a saving child, as this will trigger
* a copy on write madness. */ * a copy on write madness. */
if (server.rdb_child_pid == -1 && if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH)){
server.aof_child_pid == -1 &&
!(flags & LOOKUP_NOTOUCH))
{
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) { if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
updateLFU(val); updateLFU(val);
} else { } else {
...@@ -154,9 +151,13 @@ robj *lookupKeyRead(redisDb *db, robj *key) { ...@@ -154,9 +151,13 @@ robj *lookupKeyRead(redisDb *db, robj *key) {
* *
* Returns the linked value object if the key exists or NULL if the key * Returns the linked value object if the key exists or NULL if the key
* does not exist in the specified DB. */ * does not exist in the specified DB. */
robj *lookupKeyWrite(redisDb *db, robj *key) { robj *lookupKeyWriteWithFlags(redisDb *db, robj *key, int flags) {
expireIfNeeded(db,key); expireIfNeeded(db,key);
return lookupKey(db,key,LOOKUP_NONE); return lookupKey(db,key,flags);
}
robj *lookupKeyWrite(redisDb *db, robj *key) {
return lookupKeyWriteWithFlags(db, key, LOOKUP_NONE);
} }
robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply) { robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply) {
...@@ -353,6 +354,12 @@ long long emptyDbGeneric(redisDb *dbarray, int dbnum, int flags, void(callback)( ...@@ -353,6 +354,12 @@ long long emptyDbGeneric(redisDb *dbarray, int dbnum, int flags, void(callback)(
return -1; return -1;
} }
/* Fire the flushdb modules event. */
RedisModuleFlushInfoV1 fi = {REDISMODULE_FLUSHINFO_VERSION,!async,dbnum};
moduleFireServerEvent(REDISMODULE_EVENT_FLUSHDB,
REDISMODULE_SUBEVENT_FLUSHDB_START,
&fi);
/* Make sure the WATCHed keys are affected by the FLUSH* commands. /* Make sure the WATCHed keys are affected by the FLUSH* commands.
* Note that we need to call the function while the keys are still * Note that we need to call the function while the keys are still
* there. */ * there. */
...@@ -383,6 +390,13 @@ long long emptyDbGeneric(redisDb *dbarray, int dbnum, int flags, void(callback)( ...@@ -383,6 +390,13 @@ long long emptyDbGeneric(redisDb *dbarray, int dbnum, int flags, void(callback)(
} }
} }
if (dbnum == -1) flushSlaveKeysWithExpireList(); if (dbnum == -1) flushSlaveKeysWithExpireList();
/* Also fire the end event. Note that this event will fire almost
* immediately after the start event if the flush is asynchronous. */
moduleFireServerEvent(REDISMODULE_EVENT_FLUSHDB,
REDISMODULE_SUBEVENT_FLUSHDB_END,
&fi);
return removed; return removed;
} }
...@@ -451,6 +465,29 @@ int getFlushCommandFlags(client *c, int *flags) { ...@@ -451,6 +465,29 @@ int getFlushCommandFlags(client *c, int *flags) {
return C_OK; return C_OK;
} }
/* Flushes the whole server data set. */
void flushAllDataAndResetRDB(int flags) {
server.dirty += emptyDb(-1,flags,NULL);
if (server.rdb_child_pid != -1) killRDBChild();
if (server.saveparamslen > 0) {
/* Normally rdbSave() will reset dirty, but we don't want this here
* as otherwise FLUSHALL will not be replicated nor put into the AOF. */
int saved_dirty = server.dirty;
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
rdbSave(server.rdb_filename,rsiptr);
server.dirty = saved_dirty;
}
server.dirty++;
#if defined(USE_JEMALLOC)
/* jemalloc 5 doesn't release pages back to the OS when there's no traffic.
* for large databases, flushdb blocks for long anyway, so a bit more won't
* harm and this way the flush and purge will be synchroneus. */
if (!(flags & EMPTYDB_ASYNC))
jemalloc_purge();
#endif
}
/* FLUSHDB [ASYNC] /* FLUSHDB [ASYNC]
* *
* Flushes the currently SELECTed Redis DB. */ * Flushes the currently SELECTed Redis DB. */
...@@ -460,6 +497,13 @@ void flushdbCommand(client *c) { ...@@ -460,6 +497,13 @@ void flushdbCommand(client *c) {
if (getFlushCommandFlags(c,&flags) == C_ERR) return; if (getFlushCommandFlags(c,&flags) == C_ERR) return;
server.dirty += emptyDb(c->db->id,flags,NULL); server.dirty += emptyDb(c->db->id,flags,NULL);
addReply(c,shared.ok); addReply(c,shared.ok);
#if defined(USE_JEMALLOC)
/* jemalloc 5 doesn't release pages back to the OS when there's no traffic.
* for large databases, flushdb blocks for long anyway, so a bit more won't
* harm and this way the flush and purge will be synchroneus. */
if (!(flags & EMPTYDB_ASYNC))
jemalloc_purge();
#endif
} }
/* FLUSHALL [ASYNC] /* FLUSHALL [ASYNC]
...@@ -467,21 +511,9 @@ void flushdbCommand(client *c) { ...@@ -467,21 +511,9 @@ void flushdbCommand(client *c) {
* Flushes the whole server data set. */ * Flushes the whole server data set. */
void flushallCommand(client *c) { void flushallCommand(client *c) {
int flags; int flags;
if (getFlushCommandFlags(c,&flags) == C_ERR) return; if (getFlushCommandFlags(c,&flags) == C_ERR) return;
server.dirty += emptyDb(-1,flags,NULL); flushAllDataAndResetRDB(flags);
addReply(c,shared.ok); addReply(c,shared.ok);
if (server.rdb_child_pid != -1) killRDBChild();
if (server.saveparamslen > 0) {
/* Normally rdbSave() will reset dirty, but we don't want this here
* as otherwise FLUSHALL will not be replicated nor put into the AOF. */
int saved_dirty = server.dirty;
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
rdbSave(server.rdb_filename,rsiptr);
server.dirty = saved_dirty;
}
server.dirty++;
} }
/* This command implements DEL and LAZYDEL. */ /* This command implements DEL and LAZYDEL. */
......
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