fluencelabs/redis
1
0
Fork 0
mirror of https://github.com/fluencelabs/redis synced 2025-03-30 22:31:03 +00:00
Code Issues Projects Releases Wiki Activity

Merge remote branch 'origin/unstable' into unstable

Browse Source
This commit is contained in:
antirez 2011-06-05 20:51:49 +02:00
commit cf1eefa420
12 changed files with 333 additions and 213 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/aof.c
View File

@ -565,16 +565,18 @@ werr:
*/ */
int rewriteAppendOnlyFileBackground(void) { int rewriteAppendOnlyFileBackground(void) {
pid_t childpid; pid_t childpid;
long long start;
if (server.bgrewritechildpid != -1) return REDIS_ERR; if (server.bgrewritechildpid != -1) return REDIS_ERR;
if (server.ds_enabled != 0) { if (server.ds_enabled != 0) {
redisLog(REDIS_WARNING,"BGREWRITEAOF called with diskstore enabled: AOF is not supported when diskstore is enabled. Operation not performed."); redisLog(REDIS_WARNING,"BGREWRITEAOF called with diskstore enabled: AOF is not supported when diskstore is enabled. Operation not performed.");
return REDIS_ERR; return REDIS_ERR;
} }
start = ustime();
if ((childpid = fork()) == 0) { if ((childpid = fork()) == 0) {
/* Child */
char tmpfile[256]; char tmpfile[256];
/* Child */
if (server.ipfd > 0) close(server.ipfd); if (server.ipfd > 0) close(server.ipfd);
if (server.sofd > 0) close(server.sofd); if (server.sofd > 0) close(server.sofd);
snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid()); snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
@ -585,6 +587,7 @@ int rewriteAppendOnlyFileBackground(void) {
} }
} else { } else {
/* Parent */ /* Parent */
server.stat_fork_time = ustime()-start;
if (childpid == -1) { if (childpid == -1) {
redisLog(REDIS_WARNING, redisLog(REDIS_WARNING,
"Can't rewrite append only file in background: fork: %s", "Can't rewrite append only file in background: fork: %s",

5
src/object.c
View File

@ -180,7 +180,7 @@ void decrRefCount(void *obj) {
robj *o = obj; robj *o = obj;
if (o->refcount <= 0) redisPanic("decrRefCount against refcount <= 0"); if (o->refcount <= 0) redisPanic("decrRefCount against refcount <= 0");
if (--(o->refcount) == 0) { if (o->refcount == 1) {
switch(o->type) { switch(o->type) {
case REDIS_STRING: freeStringObject(o); break; case REDIS_STRING: freeStringObject(o); break;
case REDIS_LIST: freeListObject(o); break; case REDIS_LIST: freeListObject(o); break;
@ -189,8 +189,9 @@ void decrRefCount(void *obj) {
case REDIS_HASH: freeHashObject(o); break; case REDIS_HASH: freeHashObject(o); break;
default: redisPanic("Unknown object type"); break; default: redisPanic("Unknown object type"); break;
} }
o->ptr = NULL; /* defensive programming. We'll see NULL in traces. */
zfree(o); zfree(o);
} else {
o->refcount--;
} }
} }

3
src/rdb.c
View File

@ -482,6 +482,7 @@ werr:
int rdbSaveBackground(char *filename) { int rdbSaveBackground(char *filename) {
pid_t childpid; pid_t childpid;
long long start;
if (server.bgsavechildpid != -1 || if (server.bgsavechildpid != -1 ||
server.bgsavethread != (pthread_t) -1) return REDIS_ERR; server.bgsavethread != (pthread_t) -1) return REDIS_ERR;
@ -493,6 +494,7 @@ int rdbSaveBackground(char *filename) {
return dsRdbSaveBackground(filename); return dsRdbSaveBackground(filename);
} }
start = ustime();
if ((childpid = fork()) == 0) { if ((childpid = fork()) == 0) {
int retval; int retval;
@ -503,6 +505,7 @@ int rdbSaveBackground(char *filename) {
_exit((retval == REDIS_OK) ? 0 : 1); _exit((retval == REDIS_OK) ? 0 : 1);
} else { } else {
/* Parent */ /* Parent */
server.stat_fork_time = ustime()-start;
if (childpid == -1) { if (childpid == -1) {
redisLog(REDIS_WARNING,"Can't save in background: fork: %s", redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
strerror(errno)); strerror(errno));

61
src/redis-cli.c
View File

@ -55,6 +55,7 @@ static struct config {
int hostport; int hostport;
char *hostsocket; char *hostsocket;
long repeat; long repeat;
long interval;
int dbnum; int dbnum;
int interactive; int interactive;
int shutdown; int shutdown;
@ -87,9 +88,11 @@ static long long mstime(void) {
static void cliRefreshPrompt(void) { static void cliRefreshPrompt(void) {
if (config.dbnum == 0) if (config.dbnum == 0)
snprintf(config.prompt,sizeof(config.prompt),"redis> "); snprintf(config.prompt,sizeof(config.prompt),"redis %s:%d> ",
config.hostip, config.hostport);
else else
snprintf(config.prompt,sizeof(config.prompt),"redis:%d> ",config.dbnum); snprintf(config.prompt,sizeof(config.prompt),"redis %s:%d[%d]> ",
config.hostip, config.hostport, config.dbnum);
} }
/*------------------------------------------------------------------------------ /*------------------------------------------------------------------------------
@ -314,10 +317,9 @@ static int cliConnect(int force) {
return REDIS_OK; return REDIS_OK;
} }
static void cliPrintContextErrorAndExit() { static void cliPrintContextError() {
if (context == NULL) return; if (context == NULL) return;
fprintf(stderr,"Error: %s\n",context->errstr); fprintf(stderr,"Error: %s\n",context->errstr);
exit(1);
} }
static sds cliFormatReplyTTY(redisReply *r, char *prefix) { static sds cliFormatReplyTTY(redisReply *r, char *prefix) {
@ -436,7 +438,8 @@ static int cliReadReply(int output_raw_strings) {
if (context->err == REDIS_ERR_EOF) if (context->err == REDIS_ERR_EOF)
return REDIS_ERR; return REDIS_ERR;
} }
cliPrintContextErrorAndExit(); cliPrintContextError();
exit(1);
return REDIS_ERR; /* avoid compiler warning */ return REDIS_ERR; /* avoid compiler warning */
} }
@ -462,10 +465,7 @@ static int cliSendCommand(int argc, char **argv, int repeat) {
size_t *argvlen; size_t *argvlen;
int j, output_raw; int j, output_raw;
if (context == NULL) { if (context == NULL) return REDIS_ERR;
printf("Not connected, please use: connect <host> <port>\n");
return REDIS_OK;
}
output_raw = 0; output_raw = 0;
if (!strcasecmp(command,"info") || if (!strcasecmp(command,"info") ||
@ -518,6 +518,8 @@ static int cliSendCommand(int argc, char **argv, int repeat) {
cliRefreshPrompt(); cliRefreshPrompt();
} }
} }
if (config.interval) usleep(config.interval);
fflush(stdout); /* Make it grep friendly */
} }
free(argvlen); free(argvlen);
@ -553,6 +555,10 @@ static int parseOptions(int argc, char **argv) {
} else if (!strcmp(argv[i],"-r") && !lastarg) { } else if (!strcmp(argv[i],"-r") && !lastarg) {
config.repeat = strtoll(argv[i+1],NULL,10); config.repeat = strtoll(argv[i+1],NULL,10);
i++; i++;
} else if (!strcmp(argv[i],"-i") && !lastarg) {
double seconds = atof(argv[i+1]);
config.interval = seconds*1000000;
i++;
} else if (!strcmp(argv[i],"-n") && !lastarg) { } else if (!strcmp(argv[i],"-n") && !lastarg) {
config.dbnum = atoi(argv[i+1]); config.dbnum = atoi(argv[i+1]);
i++; i++;
@ -605,6 +611,8 @@ static void usage() {
" -s <socket> Server socket (overrides hostname and port)\n" " -s <socket> Server socket (overrides hostname and port)\n"
" -a <password> Password to use when connecting to the server\n" " -a <password> Password to use when connecting to the server\n"
" -r <repeat> Execute specified command N times\n" " -r <repeat> Execute specified command N times\n"
" -i <interval> When -r is used, waits <interval> seconds per command.\n"
" It is possible to specify sub-second times like -i 0.1.\n"
" -n <db> Database number\n" " -n <db> Database number\n"
" -x Read last argument from STDIN\n" " -x Read last argument from STDIN\n"
" -d <delimiter> Multi-bulk delimiter in for raw formatting (default: \\n)\n" " -d <delimiter> Multi-bulk delimiter in for raw formatting (default: \\n)\n"
@ -616,6 +624,7 @@ static void usage() {
" cat /etc/passwd | redis-cli -x set mypasswd\n" " cat /etc/passwd | redis-cli -x set mypasswd\n"
" redis-cli get mypasswd\n" " redis-cli get mypasswd\n"
" redis-cli -r 100 lpush mylist x\n" " redis-cli -r 100 lpush mylist x\n"
" redis-cli -r 100 -i 1 info | grep used_memory_human:\n"
"\n" "\n"
"When no command is given, redis-cli starts in interactive mode.\n" "When no command is given, redis-cli starts in interactive mode.\n"
"Type \"help\" in interactive mode for information on available commands.\n" "Type \"help\" in interactive mode for information on available commands.\n"
@ -681,14 +690,25 @@ static void repl() {
linenoiseClearScreen(); linenoiseClearScreen();
} else { } else {
long long start_time = mstime(), elapsed; long long start_time = mstime(), elapsed;
int repeat, skipargs = 0;
if (cliSendCommand(argc,argv,1) != REDIS_OK) { repeat = atoi(argv[0]);
if (repeat) {
skipargs = 1;
} else {
repeat = 1;
}
if (cliSendCommand(argc-skipargs,argv+skipargs,repeat)
!= REDIS_OK)
{
cliConnect(1); cliConnect(1);
/* If we still cannot send the command, /* If we still cannot send the command print error.
* print error and abort. */ * We'll try to reconnect the next time. */
if (cliSendCommand(argc,argv,1) != REDIS_OK) if (cliSendCommand(argc-skipargs,argv+skipargs,repeat)
cliPrintContextErrorAndExit(); != REDIS_OK)
cliPrintContextError();
} }
elapsed = mstime()-start_time; elapsed = mstime()-start_time;
if (elapsed >= 500) { if (elapsed >= 500) {
@ -726,6 +746,7 @@ int main(int argc, char **argv) {
config.hostport = 6379; config.hostport = 6379;
config.hostsocket = NULL; config.hostsocket = NULL;
config.repeat = 1; config.repeat = 1;
config.interval = 0;
config.dbnum = 0; config.dbnum = 0;
config.interactive = 0; config.interactive = 0;
config.shutdown = 0; config.shutdown = 0;
@ -741,11 +762,15 @@ int main(int argc, char **argv) {
argc -= firstarg; argc -= firstarg;
argv += firstarg; argv += firstarg;
/* Try to connect */
if (cliConnect(0) != REDIS_OK) exit(1);
/* Start interactive mode when no command is provided */ /* Start interactive mode when no command is provided */
if (argc == 0) repl(); if (argc == 0) {
/* Note that in repl mode we don't abort on connection error.
* A new attempt will be performed for every command send. */
cliConnect(0);
repl();
}
/* Otherwise, we have some arguments to execute */ /* Otherwise, we have some arguments to execute */
if (cliConnect(0) != REDIS_OK) exit(1);
return noninteractive(argc,convertToSds(argc,argv)); return noninteractive(argc,convertToSds(argc,argv));
} }

12
src/redis.c
View File

@ -116,9 +116,9 @@ struct redisCommand redisCommandTable[] = {
{"sdiff",sdiffCommand,-2,REDIS_CMD_DENYOOM,NULL,1,-1,1,0,0}, {"sdiff",sdiffCommand,-2,REDIS_CMD_DENYOOM,NULL,1,-1,1,0,0},
{"sdiffstore",sdiffstoreCommand,-3,REDIS_CMD_DENYOOM,NULL,2,-1,1,0,0}, {"sdiffstore",sdiffstoreCommand,-3,REDIS_CMD_DENYOOM,NULL,2,-1,1,0,0},
{"smembers",sinterCommand,2,0,NULL,1,1,1,0,0}, {"smembers",sinterCommand,2,0,NULL,1,1,1,0,0},
{"zadd",zaddCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1,0,0}, {"zadd",zaddCommand,-4,REDIS_CMD_DENYOOM,NULL,1,1,1,0,0},
{"zincrby",zincrbyCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1,0,0}, {"zincrby",zincrbyCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1,0,0},
{"zrem",zremCommand,3,0,NULL,1,1,1,0,0}, {"zrem",zremCommand,-3,0,NULL,1,1,1,0,0},
{"zremrangebyscore",zremrangebyscoreCommand,4,0,NULL,1,1,1,0,0}, {"zremrangebyscore",zremrangebyscoreCommand,4,0,NULL,1,1,1,0,0},
{"zremrangebyrank",zremrangebyrankCommand,4,0,NULL,1,1,1,0,0}, {"zremrangebyrank",zremrangebyrankCommand,4,0,NULL,1,1,1,0,0},
{"zunionstore",zunionstoreCommand,-4,REDIS_CMD_DENYOOM,zunionInterGetKeys,0,0,0,0,0}, {"zunionstore",zunionstoreCommand,-4,REDIS_CMD_DENYOOM,zunionInterGetKeys,0,0,0,0,0},
@ -871,6 +871,7 @@ void initServerConfig() {
server.masterport = 6379; server.masterport = 6379;
server.master = NULL; server.master = NULL;
server.replstate = REDIS_REPL_NONE; server.replstate = REDIS_REPL_NONE;
server.repl_syncio_timeout = REDIS_REPL_SYNCIO_TIMEOUT;
server.repl_serve_stale_data = 1; server.repl_serve_stale_data = 1;
/* Double constants initialization */ /* Double constants initialization */
@ -963,6 +964,7 @@ void initServer() {
server.stat_keyspace_misses = 0; server.stat_keyspace_misses = 0;
server.stat_keyspace_hits = 0; server.stat_keyspace_hits = 0;
server.stat_peak_memory = 0; server.stat_peak_memory = 0;
server.stat_fork_time = 0;
server.unixtime = time(NULL); server.unixtime = time(NULL);
aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL); aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
if (server.ipfd > 0 && aeCreateFileEvent(server.el,server.ipfd,AE_READABLE, if (server.ipfd > 0 && aeCreateFileEvent(server.el,server.ipfd,AE_READABLE,
@ -1438,7 +1440,8 @@ sds genRedisInfoString(char *section) {
"keyspace_hits:%lld\r\n" "keyspace_hits:%lld\r\n"
"keyspace_misses:%lld\r\n" "keyspace_misses:%lld\r\n"
"pubsub_channels:%ld\r\n" "pubsub_channels:%ld\r\n"
"pubsub_patterns:%u\r\n", "pubsub_patterns:%u\r\n"
"latest_fork_usec:%lld\r\n",
server.stat_numconnections, server.stat_numconnections,
server.stat_numcommands, server.stat_numcommands,
server.stat_expiredkeys, server.stat_expiredkeys,
@ -1446,7 +1449,8 @@ sds genRedisInfoString(char *section) {
server.stat_keyspace_hits, server.stat_keyspace_hits,
server.stat_keyspace_misses, server.stat_keyspace_misses,
dictSize(server.pubsub_channels), dictSize(server.pubsub_channels),
listLength(server.pubsub_patterns)); listLength(server.pubsub_patterns),
server.stat_fork_time);
} }
/* Replication */ /* Replication */

16
src/redis.h
View File

@ -41,7 +41,6 @@
#define REDIS_MAXIDLETIME (60*5) /* default client timeout */ #define REDIS_MAXIDLETIME (60*5) /* default client timeout */
#define REDIS_IOBUF_LEN 1024 #define REDIS_IOBUF_LEN 1024
#define REDIS_LOADBUF_LEN 1024 #define REDIS_LOADBUF_LEN 1024
#define REDIS_STATIC_ARGS 8
#define REDIS_DEFAULT_DBNUM 16 #define REDIS_DEFAULT_DBNUM 16
#define REDIS_CONFIGLINE_MAX 1024 #define REDIS_CONFIGLINE_MAX 1024
#define REDIS_MAX_SYNC_TIME 60 /* Slave can't take more to sync */ #define REDIS_MAX_SYNC_TIME 60 /* Slave can't take more to sync */
@ -153,10 +152,14 @@
#define REDIS_REQ_MULTIBULK 2 #define REDIS_REQ_MULTIBULK 2
/* Slave replication state - slave side */ /* Slave replication state - slave side */
#define REDIS_REPL_NONE 0 /* No active replication */ #define REDIS_REPL_NONE 0 /* No active replication */
#define REDIS_REPL_CONNECT 1 /* Must connect to master */ #define REDIS_REPL_CONNECT 1 /* Must connect to master */
#define REDIS_REPL_TRANSFER 2 /* Receiving .rdb from master */ #define REDIS_REPL_CONNECTING 2 /* Connecting to master */
#define REDIS_REPL_CONNECTED 3 /* Connected to master */ #define REDIS_REPL_TRANSFER 3 /* Receiving .rdb from master */
#define REDIS_REPL_CONNECTED 4 /* Connected to master */
/* Synchronous read timeout - slave side */
#define REDIS_REPL_SYNCIO_TIMEOUT 5
/* Slave replication state - from the point of view of master /* Slave replication state - from the point of view of master
* Note that in SEND_BULK and ONLINE state the slave receives new updates * Note that in SEND_BULK and ONLINE state the slave receives new updates
@ -543,6 +546,7 @@ struct redisServer {
long long stat_keyspace_hits; /* number of successful lookups of keys */ long long stat_keyspace_hits; /* number of successful lookups of keys */
long long stat_keyspace_misses; /* number of failed lookups of keys */ long long stat_keyspace_misses; /* number of failed lookups of keys */
size_t stat_peak_memory; /* max used memory record */ size_t stat_peak_memory; /* max used memory record */
long long stat_fork_time; /* time needed to perform latets fork() */
/* Configuration */ /* Configuration */
int verbosity; int verbosity;
int maxidletime; int maxidletime;
@ -585,6 +589,7 @@ struct redisServer {
char *masterhost; char *masterhost;
int masterport; int masterport;
redisClient *master; /* client that is master for this slave */ redisClient *master; /* client that is master for this slave */
int repl_syncio_timeout; /* timeout for synchronous I/O calls */
int replstate; /* replication status if the instance is a slave */ int replstate; /* replication status if the instance is a slave */
off_t repl_transfer_left; /* bytes left reading .rdb */ off_t repl_transfer_left; /* bytes left reading .rdb */
int repl_transfer_s; /* slave -> master SYNC socket */ int repl_transfer_s; /* slave -> master SYNC socket */
@ -888,7 +893,6 @@ int fwriteBulkCount(FILE *fp, char prefix, int count);
/* Replication */ /* Replication */
void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc); void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc);
void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc); void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc);
int syncWithMaster(void);
void updateSlavesWaitingBgsave(int bgsaveerr); void updateSlavesWaitingBgsave(int bgsaveerr);
void replicationCron(void); void replicationCron(void);

150
src/replication.c
View File

@ -10,38 +10,8 @@
void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) { void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
listNode *ln; listNode *ln;
listIter li; listIter li;
int outc = 0, j; int j;
robj **outv;
/* We need 1+(ARGS*3) objects since commands are using the new protocol
* and we one 1 object for the first "*<count>\r\n" multibulk count, then
* for every additional object we have "$<count>\r\n" + object + "\r\n". */
robj *static_outv[REDIS_STATIC_ARGS*3+1];
robj *lenobj;
if (argc <= REDIS_STATIC_ARGS) {
outv = static_outv;
} else {
outv = zmalloc(sizeof(robj*)*(argc*3+1));
}
lenobj = createObject(REDIS_STRING,
sdscatprintf(sdsempty(), "*%d\r\n", argc));
lenobj->refcount = 0;
outv[outc++] = lenobj;
for (j = 0; j < argc; j++) {
lenobj = createObject(REDIS_STRING,
sdscatprintf(sdsempty(),"$%lu\r\n",
(unsigned long) stringObjectLen(argv[j])));
lenobj->refcount = 0;
outv[outc++] = lenobj;
outv[outc++] = argv[j];
outv[outc++] = shared.crlf;
}
/* Increment all the refcounts at start and decrement at end in order to
* be sure to free objects if there is no slave in a replication state
* able to be feed with commands */
for (j = 0; j < outc; j++) incrRefCount(outv[j]);
listRewind(slaves,&li); listRewind(slaves,&li);
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
redisClient *slave = ln->value; redisClient *slave = ln->value;
@ -49,7 +19,9 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
/* Don't feed slaves that are still waiting for BGSAVE to start */ /* Don't feed slaves that are still waiting for BGSAVE to start */
if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue; if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue;
/* Feed all the other slaves, MONITORs and so on */ /* Feed slaves that are waiting for the initial SYNC (so these commands
* are queued in the output buffer until the intial SYNC completes),
* or are already in sync with the master. */
if (slave->slaveseldb != dictid) { if (slave->slaveseldb != dictid) {
robj *selectcmd; robj *selectcmd;
@ -73,10 +45,9 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
addReply(slave,selectcmd); addReply(slave,selectcmd);
slave->slaveseldb = dictid; slave->slaveseldb = dictid;
} }
for (j = 0; j < outc; j++) addReply(slave,outv[j]); addReplyMultiBulkLen(slave,argc);
for (j = 0; j < argc; j++) addReplyBulk(slave,argv[j]);
} }
for (j = 0; j < outc; j++) decrRefCount(outv[j]);
if (outv != static_outv) zfree(outv);
} }
void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc) { void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc) {
@ -315,19 +286,18 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
/* If repl_transfer_left == -1 we still have to read the bulk length /* If repl_transfer_left == -1 we still have to read the bulk length
* from the master reply. */ * from the master reply. */
if (server.repl_transfer_left == -1) { if (server.repl_transfer_left == -1) {
if (syncReadLine(fd,buf,1024,3600) == -1) { if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout) == -1) {
redisLog(REDIS_WARNING, redisLog(REDIS_WARNING,
"I/O error reading bulk count from MASTER: %s", "I/O error reading bulk count from MASTER: %s",
strerror(errno)); strerror(errno));
replicationAbortSyncTransfer(); goto error;
return;
} }
if (buf[0] == '-') { if (buf[0] == '-') {
redisLog(REDIS_WARNING, redisLog(REDIS_WARNING,
"MASTER aborted replication with an error: %s", "MASTER aborted replication with an error: %s",
buf+1); buf+1);
replicationAbortSyncTransfer(); goto error;
return;
} else if (buf[0] == '\0') { } else if (buf[0] == '\0') {
/* At this stage just a newline works as a PING in order to take /* At this stage just a newline works as a PING in order to take
* the connection live. So we refresh our last interaction * the connection live. So we refresh our last interaction
@ -336,8 +306,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
return; return;
} else if (buf[0] != '$') { } else if (buf[0] != '$') {
redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?"); redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
replicationAbortSyncTransfer(); goto error;
return;
} }
server.repl_transfer_left = strtol(buf+1,NULL,10); server.repl_transfer_left = strtol(buf+1,NULL,10);
redisLog(REDIS_NOTICE, redisLog(REDIS_NOTICE,
@ -359,8 +328,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
server.repl_transfer_lastio = time(NULL); server.repl_transfer_lastio = time(NULL);
if (write(server.repl_transfer_fd,buf,nread) != nread) { if (write(server.repl_transfer_fd,buf,nread) != nread) {
redisLog(REDIS_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno)); redisLog(REDIS_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
replicationAbortSyncTransfer(); goto error;
return;
} }
server.repl_transfer_left -= nread; server.repl_transfer_left -= nread;
/* Check if the transfer is now complete */ /* Check if the transfer is now complete */
@ -391,48 +359,54 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
server.replstate = REDIS_REPL_CONNECTED; server.replstate = REDIS_REPL_CONNECTED;
redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success"); redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success");
} }
return;
error:
replicationAbortSyncTransfer();
return;
} }
int syncWithMaster(void) { void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
char buf[1024], tmpfile[256], authcmd[1024]; char buf[1024], tmpfile[256];
int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
int dfd, maxtries = 5; int dfd, maxtries = 5;
REDIS_NOTUSED(el);
REDIS_NOTUSED(privdata);
REDIS_NOTUSED(mask);
if (fd == -1) { /* This event should only be triggered once since it is used to have a
redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s", * non-blocking connect(2) to the master. It has been triggered when this
strerror(errno)); * function is called, so we can delete it. */
return REDIS_ERR; aeDeleteFileEvent(server.el,fd,AE_WRITABLE);
}
/* AUTH with the master if required. */ /* AUTH with the master if required. */
if(server.masterauth) { if(server.masterauth) {
snprintf(authcmd, 1024, "AUTH %s\r\n", server.masterauth); char authcmd[1024];
if (syncWrite(fd, authcmd, strlen(server.masterauth)+7, 5) == -1) { size_t authlen;
close(fd);
authlen = snprintf(authcmd,sizeof(authcmd),"AUTH %s\r\n",server.masterauth);
if (syncWrite(fd,authcmd,authlen,server.repl_syncio_timeout) == -1) {
redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s", redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s",
strerror(errno)); strerror(errno));
return REDIS_ERR; goto error;
} }
/* Read the AUTH result. */ /* Read the AUTH result. */
if (syncReadLine(fd,buf,1024,3600) == -1) { if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout) == -1) {
close(fd);
redisLog(REDIS_WARNING,"I/O error reading auth result from MASTER: %s", redisLog(REDIS_WARNING,"I/O error reading auth result from MASTER: %s",
strerror(errno)); strerror(errno));
return REDIS_ERR; goto error;
} }
if (buf[0] != '+') { if (buf[0] != '+') {
close(fd);
redisLog(REDIS_WARNING,"Cannot AUTH to MASTER, is the masterauth password correct?"); redisLog(REDIS_WARNING,"Cannot AUTH to MASTER, is the masterauth password correct?");
return REDIS_ERR; goto error;
} }
} }
/* Issue the SYNC command */ /* Issue the SYNC command */
if (syncWrite(fd,"SYNC \r\n",7,5) == -1) { if (syncWrite(fd,"SYNC \r\n",7,server.repl_syncio_timeout) == -1) {
close(fd);
redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s", redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
strerror(errno)); strerror(errno));
return REDIS_ERR; goto error;
} }
/* Prepare a suitable temp file for bulk transfer */ /* Prepare a suitable temp file for bulk transfer */
@ -444,25 +418,51 @@ int syncWithMaster(void) {
sleep(1); sleep(1);
} }
if (dfd == -1) { if (dfd == -1) {
close(fd);
redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno)); redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
return REDIS_ERR; goto error;
} }
/* Setup the non blocking download of the bulk file. */ /* Setup the non blocking download of the bulk file. */
if (aeCreateFileEvent(server.el, fd, AE_READABLE, readSyncBulkPayload, NULL) if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL)
== AE_ERR) == AE_ERR)
{
redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
goto error;
}
server.replstate = REDIS_REPL_TRANSFER;
server.repl_transfer_left = -1;
server.repl_transfer_fd = dfd;
server.repl_transfer_lastio = time(NULL);
server.repl_transfer_tmpfile = zstrdup(tmpfile);
return;
error:
server.replstate = REDIS_REPL_CONNECT;
close(fd);
return;
}
int connectWithMaster(void) {
int fd;
fd = anetTcpNonBlockConnect(NULL,server.masterhost,server.masterport);
if (fd == -1) {
redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
strerror(errno));
return REDIS_ERR;
}
if (aeCreateFileEvent(server.el,fd,AE_WRITABLE,syncWithMaster,NULL) ==
AE_ERR)
{ {
close(fd); close(fd);
redisLog(REDIS_WARNING,"Can't create readable event for SYNC"); redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
return REDIS_ERR; return REDIS_ERR;
} }
server.replstate = REDIS_REPL_TRANSFER;
server.repl_transfer_left = -1;
server.repl_transfer_s = fd; server.repl_transfer_s = fd;
server.repl_transfer_fd = dfd; server.replstate = REDIS_REPL_CONNECTING;
server.repl_transfer_lastio = time(NULL);
server.repl_transfer_tmpfile = zstrdup(tmpfile);
return REDIS_OK; return REDIS_OK;
} }
@ -517,8 +517,8 @@ void replicationCron(void) {
/* Check if we should connect to a MASTER */ /* Check if we should connect to a MASTER */
if (server.replstate == REDIS_REPL_CONNECT) { if (server.replstate == REDIS_REPL_CONNECT) {
redisLog(REDIS_NOTICE,"Connecting to MASTER..."); redisLog(REDIS_NOTICE,"Connecting to MASTER...");
if (syncWithMaster() == REDIS_OK) { if (connectWithMaster() == REDIS_OK) {
redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync started: SYNC sent"); redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync started");
if (server.appendonly) rewriteAppendOnlyFileBackground(); if (server.appendonly) rewriteAppendOnlyFileBackground();
} }
} }

5
src/t_set.c
View File

@ -249,8 +249,11 @@ void sremCommand(redisClient *c) {
for (j = 2; j < c->argc; j++) { for (j = 2; j < c->argc; j++) {
if (setTypeRemove(set,c->argv[j])) { if (setTypeRemove(set,c->argv[j])) {
if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
deleted++; deleted++;
if (setTypeSize(set) == 0) {
dbDelete(c->db,c->argv[1]);
break;
}
} }
} }
if (deleted) { if (deleted) {

233
src/t_zset.c
View File

@ -810,11 +810,29 @@ void zaddGenericCommand(redisClient *c, int incr) {
robj *ele; robj *ele;
robj *zobj; robj *zobj;
robj *curobj; robj *curobj;
double score, curscore = 0.0; double score = 0, *scores, curscore = 0.0;
int j, elements = (c->argc-2)/2;
int added = 0;
if (getDoubleFromObjectOrReply(c,c->argv[2],&score,NULL) != REDIS_OK) if (c->argc % 2) {
addReply(c,shared.syntaxerr);
return; return;
}
/* Start parsing all the scores, we need to emit any syntax error
* before executing additions to the sorted set, as the command should
* either execute fully or nothing at all. */
scores = zmalloc(sizeof(double)*elements);
for (j = 0; j < elements; j++) {
if (getDoubleFromObjectOrReply(c,c->argv[2+j*2],&scores[j],NULL)
!= REDIS_OK)
{
zfree(scores);
return;
}
}
/* Lookup the key and create the sorted set if does not exist. */
zobj = lookupKeyWrite(c->db,key); zobj = lookupKeyWrite(c->db,key);
if (zobj == NULL) { if (zobj == NULL) {
if (server.zset_max_ziplist_entries == 0 || if (server.zset_max_ziplist_entries == 0 ||
@ -828,111 +846,105 @@ void zaddGenericCommand(redisClient *c, int incr) {
} else { } else {
if (zobj->type != REDIS_ZSET) { if (zobj->type != REDIS_ZSET) {
addReply(c,shared.wrongtypeerr); addReply(c,shared.wrongtypeerr);
zfree(scores);
return; return;
} }
} }
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { for (j = 0; j < elements; j++) {
unsigned char *eptr; score = scores[j];
/* Prefer non-encoded element when dealing with ziplists. */ if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
ele = c->argv[3]; unsigned char *eptr;
if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
if (incr) { /* Prefer non-encoded element when dealing with ziplists. */
score += curscore; ele = c->argv[3+j*2];
if (isnan(score)) { if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
addReplyError(c,nanerr); if (incr) {
/* Don't need to check if the sorted set is empty, because score += curscore;
* we know it has at least one element. */ if (isnan(score)) {
return; addReplyError(c,nanerr);
/* Don't need to check if the sorted set is empty
* because we know it has at least one element. */
zfree(scores);
return;
}
} }
}
/* Remove and re-insert when score changed. */ /* Remove and re-insert when score changed. */
if (score != curscore) { if (score != curscore) {
zobj->ptr = zzlDelete(zobj->ptr,eptr); zobj->ptr = zzlDelete(zobj->ptr,eptr);
zobj->ptr = zzlInsert(zobj->ptr,ele,score);
signalModifiedKey(c->db,key);
server.dirty++;
}
} else {
/* Optimize: check if the element is too large or the list
* becomes too long *before* executing zzlInsert. */
zobj->ptr = zzlInsert(zobj->ptr,ele,score); zobj->ptr = zzlInsert(zobj->ptr,ele,score);
if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
signalModifiedKey(c->db,key); signalModifiedKey(c->db,key);
server.dirty++; server.dirty++;
if (!incr) added++;
} }
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplistNode *znode;
dictEntry *de;
if (incr) /* ZINCRBY */ ele = c->argv[3+j*2] = tryObjectEncoding(c->argv[3+j*2]);
addReplyDouble(c,score); de = dictFind(zs->dict,ele);
else /* ZADD */ if (de != NULL) {
addReply(c,shared.czero); curobj = dictGetEntryKey(de);
} else { curscore = *(double*)dictGetEntryVal(de);
/* Optimize: check if the element is too large or the list becomes
* too long *before* executing zzlInsert. */
zobj->ptr = zzlInsert(zobj->ptr,ele,score);
if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
if (sdslen(ele->ptr) > server.zset_max_ziplist_value)
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
signalModifiedKey(c->db,key); if (incr) {
server.dirty++; score += curscore;
if (isnan(score)) {
if (incr) /* ZINCRBY */ addReplyError(c,nanerr);
addReplyDouble(c,score); /* Don't need to check if the sorted set is empty
else /* ZADD */ * because we know it has at least one element. */
addReply(c,shared.cone); zfree(scores);
} return;
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { }
zset *zs = zobj->ptr;
zskiplistNode *znode;
dictEntry *de;
ele = c->argv[3] = tryObjectEncoding(c->argv[3]);
de = dictFind(zs->dict,ele);
if (de != NULL) {
curobj = dictGetEntryKey(de);
curscore = *(double*)dictGetEntryVal(de);
if (incr) {
score += curscore;
if (isnan(score)) {
addReplyError(c,nanerr);
/* Don't need to check if the sorted set is empty, because
* we know it has at least one element. */
return;
} }
}
/* Remove and re-insert when score changed. We can safely delete /* Remove and re-insert when score changed. We can safely
* the key object from the skiplist, since the dictionary still has * delete the key object from the skiplist, since the
* a reference to it. */ * dictionary still has a reference to it. */
if (score != curscore) { if (score != curscore) {
redisAssert(zslDelete(zs->zsl,curscore,curobj)); redisAssert(zslDelete(zs->zsl,curscore,curobj));
znode = zslInsert(zs->zsl,score,curobj); znode = zslInsert(zs->zsl,score,curobj);
incrRefCount(curobj); /* Re-inserted in skiplist. */ incrRefCount(curobj); /* Re-inserted in skiplist. */
dictGetEntryVal(de) = &znode->score; /* Update score ptr. */ dictGetEntryVal(de) = &znode->score; /* Update score ptr. */
signalModifiedKey(c->db,key);
server.dirty++;
}
} else {
znode = zslInsert(zs->zsl,score,ele);
incrRefCount(ele); /* Inserted in skiplist. */
redisAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
incrRefCount(ele); /* Added to dictionary. */
signalModifiedKey(c->db,key); signalModifiedKey(c->db,key);
server.dirty++; server.dirty++;
if (!incr) added++;
} }
if (incr) /* ZINCRBY */
addReplyDouble(c,score);
else /* ZADD */
addReply(c,shared.czero);
} else { } else {
znode = zslInsert(zs->zsl,score,ele); redisPanic("Unknown sorted set encoding");
incrRefCount(ele); /* Inserted in skiplist. */
redisAssert(dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
incrRefCount(ele); /* Added to dictionary. */
signalModifiedKey(c->db,key);
server.dirty++;
if (incr) /* ZINCRBY */
addReplyDouble(c,score);
else /* ZADD */
addReply(c,shared.cone);
} }
} else {
redisPanic("Unknown sorted set encoding");
} }
zfree(scores);
if (incr) /* ZINCRBY */
addReplyDouble(c,score);
else /* ZADD */
addReplyLongLong(c,added);
} }
void zaddCommand(redisClient *c) { void zaddCommand(redisClient *c) {
@ -945,8 +957,8 @@ void zincrbyCommand(redisClient *c) {
void zremCommand(redisClient *c) { void zremCommand(redisClient *c) {
robj *key = c->argv[1]; robj *key = c->argv[1];
robj *ele = c->argv[2];
robj *zobj; robj *zobj;
int deleted = 0, j;
if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL || if ((zobj = lookupKeyWriteOrReply(c,key,shared.czero)) == NULL ||
checkType(c,zobj,REDIS_ZSET)) return; checkType(c,zobj,REDIS_ZSET)) return;
@ -954,39 +966,48 @@ void zremCommand(redisClient *c) {
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *eptr; unsigned char *eptr;
if ((eptr = zzlFind(zobj->ptr,ele,NULL)) != NULL) { for (j = 2; j < c->argc; j++) {
zobj->ptr = zzlDelete(zobj->ptr,eptr); if ((eptr = zzlFind(zobj->ptr,c->argv[j],NULL)) != NULL) {
if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key); deleted++;
} else { zobj->ptr = zzlDelete(zobj->ptr,eptr);
addReply(c,shared.czero); if (zzlLength(zobj->ptr) == 0) {
return; dbDelete(c->db,key);
break;
}
}
} }
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr; zset *zs = zobj->ptr;
dictEntry *de; dictEntry *de;
double score; double score;
de = dictFind(zs->dict,ele); for (j = 2; j < c->argc; j++) {
if (de != NULL) { de = dictFind(zs->dict,c->argv[j]);
/* Delete from the skiplist */ if (de != NULL) {
score = *(double*)dictGetEntryVal(de); deleted++;
redisAssert(zslDelete(zs->zsl,score,ele));
/* Delete from the hash table */ /* Delete from the skiplist */
dictDelete(zs->dict,ele); score = *(double*)dictGetEntryVal(de);
if (htNeedsResize(zs->dict)) dictResize(zs->dict); redisAssert(zslDelete(zs->zsl,score,c->argv[j]));
if (dictSize(zs->dict) == 0) dbDelete(c->db,key);
} else { /* Delete from the hash table */
addReply(c,shared.czero); dictDelete(zs->dict,c->argv[j]);
return; if (htNeedsResize(zs->dict)) dictResize(zs->dict);
if (dictSize(zs->dict) == 0) {
dbDelete(c->db,key);
break;
}
}
} }
} else { } else {
redisPanic("Unknown sorted set encoding"); redisPanic("Unknown sorted set encoding");
} }
signalModifiedKey(c->db,key); if (deleted) {
server.dirty++; signalModifiedKey(c->db,key);
addReply(c,shared.cone); server.dirty += deleted;
}
addReplyLongLong(c,deleted);
} }
void zremrangebyscoreCommand(redisClient *c) { void zremrangebyscoreCommand(redisClient *c) {

7
tests/test_helper.tcl
View File

@ -110,6 +110,13 @@ proc cleanup {} {
} }
proc execute_everything {} { proc execute_everything {} {
if 0 {
# Use this when hacking on new tests.
set ::verbose 1
execute_tests "unit/first"
return
}
execute_tests "unit/printver" execute_tests "unit/printver"
execute_tests "unit/auth" execute_tests "unit/auth"
execute_tests "unit/protocol" execute_tests "unit/protocol"

6
tests/unit/type/set.tcl
View File

@ -105,6 +105,12 @@ start_server {
lsort [r smembers myset] lsort [r smembers myset]
} {a c} } {a c}
test {SREM variadic version with more args needed to destroy the key} {
r del myset
r sadd myset 1 2 3
r srem myset 1 2 3 4 5 6 7 8
} {3}
foreach {type} {hashtable intset} { foreach {type} {hashtable intset} {
for {set i 1} {$i <= 5} {incr i} { for {set i 1} {$i <= 5} {incr i} {
r del [format "set%d" $i] r del [format "set%d" $i]

43
tests/unit/type/zset.tcl
View File

@ -48,6 +48,34 @@ start_server {tags {"zset"}} {
assert_error "*NaN*" {r zincrby myzset -inf abc} assert_error "*NaN*" {r zincrby myzset -inf abc}
} }
test {ZADD - Variadic version base case} {
r del myzset
list [r zadd myzset 10 a 20 b 30 c] [r zrange myzset 0 -1 withscores]
} {3 {a 10 b 20 c 30}}
test {ZADD - Return value is the number of actually added items} {
list [r zadd myzset 5 x 20 b 30 c] [r zrange myzset 0 -1 withscores]
} {1 {x 5 a 10 b 20 c 30}}
test {ZADD - Variadic version does not add nothing on single parsing err} {
r del myzset
catch {r zadd myzset 10 a 20 b 30.badscore c} e
assert_match {*ERR*not*double*} $e
r exists myzset
} {0}
test {ZADD - Variadic version will raise error on missing arg} {
r del myzset
catch {r zadd myzset 10 a 20 b 30 c 40} e
assert_match {*ERR*syntax*} $e
}
test {ZINCRBY does not work variadic even if shares ZADD implementation} {
r del myzset
catch {r zincrby myzset 10 a 20 b 30 c} e
assert_match {*ERR*wrong*number*arg*} $e
}
test "ZCARD basics - $encoding" { test "ZCARD basics - $encoding" {
assert_equal 3 [r zcard ztmp] assert_equal 3 [r zcard ztmp]
assert_equal 0 [r zcard zdoesntexist] assert_equal 0 [r zcard zdoesntexist]
@ -65,6 +93,21 @@ start_server {tags {"zset"}} {
assert_equal 0 [r exists ztmp] assert_equal 0 [r exists ztmp]
} }
test "ZREM variadic version" {
r del ztmp
r zadd ztmp 10 a 20 b 30 c
assert_equal 2 [r zrem ztmp x y a b k]
assert_equal 0 [r zrem ztmp foo bar]
assert_equal 1 [r zrem ztmp c]
r exists ztmp
} {0}
test "ZREM variadic version -- remove elements after key deletion" {
r del ztmp
r zadd ztmp 10 a 20 b 30 c
r zrem ztmp a b c d e f g
} {3}
test "ZRANGE basics - $encoding" { test "ZRANGE basics - $encoding" {
r del ztmp r del ztmp
r zadd ztmp 1 a r zadd ztmp 1 a