Commit 0d1650f8 authored by antirez's avatar antirez
Browse files

Updated to unstable

parents 3bb818df 55937b79
...@@ -116,9 +116,9 @@ struct redisCommand redisCommandTable[] = { ...@@ -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},
...@@ -635,6 +635,14 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { ...@@ -635,6 +635,14 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
if ((server.maxidletime && !(loops % 100)) || server.bpop_blocked_clients) if ((server.maxidletime && !(loops % 100)) || server.bpop_blocked_clients)
closeTimedoutClients(); closeTimedoutClients();
/* Start a scheduled AOF rewrite if this was requested by the user while
* a BGSAVE was in progress. */
if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1 &&
server.aofrewrite_scheduled)
{
rewriteAppendOnlyFileBackground();
}
/* Check if a background saving or AOF rewrite in progress terminated. */ /* Check if a background saving or AOF rewrite in progress terminated. */
if (server.bgsavechildpid != -1 || server.bgrewritechildpid != -1) { if (server.bgsavechildpid != -1 || server.bgrewritechildpid != -1) {
int statloc; int statloc;
...@@ -669,9 +677,10 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { ...@@ -669,9 +677,10 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
} }
} }
} else if (!server.ds_enabled) { } else if (!server.ds_enabled) {
/* If there is not a background saving in progress check if
* we have to save now */
time_t now = time(NULL); time_t now = time(NULL);
/* If there is not a background saving/rewrite in progress check if
* we have to save/rewrite now */
for (j = 0; j < server.saveparamslen; j++) { for (j = 0; j < server.saveparamslen; j++) {
struct saveparam *sp = server.saveparams+j; struct saveparam *sp = server.saveparams+j;
...@@ -683,6 +692,21 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { ...@@ -683,6 +692,21 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
break; break;
} }
} }
/* Trigger an AOF rewrite if needed */
if (server.bgsavechildpid == -1 &&
server.bgrewritechildpid == -1 &&
server.auto_aofrewrite_perc &&
server.appendonly_current_size > server.auto_aofrewrite_min_size)
{
int base = server.auto_aofrewrite_base_size ?
server.auto_aofrewrite_base_size : 1;
long long growth = (server.appendonly_current_size*100/base) - 100;
if (growth >= server.auto_aofrewrite_perc) {
redisLog(REDIS_NOTICE,"Starting automatic rewriting of AOF on %lld%% growth",growth);
rewriteAppendOnlyFileBackground();
}
}
} }
/* Expire a few keys per cycle, only if this is a master. /* Expire a few keys per cycle, only if this is a master.
...@@ -832,6 +856,10 @@ void initServerConfig() { ...@@ -832,6 +856,10 @@ void initServerConfig() {
server.appendonly = 0; server.appendonly = 0;
server.appendfsync = APPENDFSYNC_EVERYSEC; server.appendfsync = APPENDFSYNC_EVERYSEC;
server.no_appendfsync_on_rewrite = 0; server.no_appendfsync_on_rewrite = 0;
server.auto_aofrewrite_perc = REDIS_AUTO_AOFREWRITE_PERC;
server.auto_aofrewrite_min_size = REDIS_AUTO_AOFREWRITE_MIN_SIZE;
server.auto_aofrewrite_base_size = 0;
server.aofrewrite_scheduled = 0;
server.lastfsync = time(NULL); server.lastfsync = time(NULL);
server.appendfd = -1; server.appendfd = -1;
server.appendseldb = -1; /* Make sure the first time will not match */ server.appendseldb = -1; /* Make sure the first time will not match */
...@@ -876,6 +904,7 @@ void initServerConfig() { ...@@ -876,6 +904,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 */
...@@ -968,6 +997,7 @@ void initServer() { ...@@ -968,6 +997,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,
...@@ -1373,6 +1403,16 @@ sds genRedisInfoString(char *section) { ...@@ -1373,6 +1403,16 @@ sds genRedisInfoString(char *section) {
server.lastsave, server.lastsave,
server.bgrewritechildpid != -1); server.bgrewritechildpid != -1);
if (server.appendonly) {
info = sdscatprintf(info,
"aof_current_size:%lld\r\n"
"aof_base_size:%lld\r\n"
"aof_pending_rewrite:%d\r\n",
(long long) server.appendonly_current_size,
(long long) server.auto_aofrewrite_base_size,
server.aofrewrite_scheduled);
}
if (server.loading) { if (server.loading) {
double perc; double perc;
time_t eta, elapsed; time_t eta, elapsed;
...@@ -1446,7 +1486,8 @@ sds genRedisInfoString(char *section) { ...@@ -1446,7 +1486,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,
...@@ -1454,7 +1495,8 @@ sds genRedisInfoString(char *section) { ...@@ -1454,7 +1495,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 */
......
...@@ -42,7 +42,6 @@ ...@@ -42,7 +42,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 */
...@@ -52,6 +51,8 @@ ...@@ -52,6 +51,8 @@
#define REDIS_SHARED_INTEGERS 10000 #define REDIS_SHARED_INTEGERS 10000
#define REDIS_REPLY_CHUNK_BYTES (5*1500) /* 5 TCP packets with default MTU */ #define REDIS_REPLY_CHUNK_BYTES (5*1500) /* 5 TCP packets with default MTU */
#define REDIS_MAX_LOGMSG_LEN 1024 /* Default maximum length of syslog messages */ #define REDIS_MAX_LOGMSG_LEN 1024 /* Default maximum length of syslog messages */
#define REDIS_AUTO_AOFREWRITE_PERC 100
#define REDIS_AUTO_AOFREWRITE_MIN_SIZE (1024*1024)
/* Hash table parameters */ /* Hash table parameters */
#define REDIS_HT_MINFILL 10 /* Minimal hash table fill 10% */ #define REDIS_HT_MINFILL 10 /* Minimal hash table fill 10% */
...@@ -157,8 +158,12 @@ ...@@ -157,8 +158,12 @@
/* 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
...@@ -548,6 +553,7 @@ struct redisServer { ...@@ -548,6 +553,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;
...@@ -556,6 +562,11 @@ struct redisServer { ...@@ -556,6 +562,11 @@ struct redisServer {
int appendonly; int appendonly;
int appendfsync; int appendfsync;
int no_appendfsync_on_rewrite; int no_appendfsync_on_rewrite;
int auto_aofrewrite_perc; /* Rewrite AOF if % growth is > M and... */
off_t auto_aofrewrite_min_size; /* the AOF file is at least N bytes. */
off_t auto_aofrewrite_base_size;/* AOF size on latest startup or rewrite. */
off_t appendonly_current_size; /* AOF current size. */
int aofrewrite_scheduled; /* Rewrite once BGSAVE terminates. */
int shutdown_asap; int shutdown_asap;
int activerehashing; int activerehashing;
char *requirepass; char *requirepass;
...@@ -590,6 +601,7 @@ struct redisServer { ...@@ -590,6 +601,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 */
...@@ -898,7 +910,6 @@ int fwriteBulkCount(FILE *fp, char prefix, int count); ...@@ -898,7 +910,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);
......
...@@ -10,38 +10,8 @@ ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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) { ...@@ -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 */
...@@ -390,49 +358,58 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -390,49 +358,58 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
server.master->authenticated = 1; server.master->authenticated = 1;
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");
/* Rewrite the AOF file now that the dataset changed. */
if (server.appendonly) rewriteAppendOnlyFileBackground();
} }
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) { redisLog(REDIS_NOTICE,"Non blocking connect for SYNC fired the event.");
redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s", /* This event should only be triggered once since it is used to have a
strerror(errno)); * non-blocking connect(2) to the master. It has been triggered when this
return REDIS_ERR; * function is called, so we can delete it. */
} aeDeleteFileEvent(server.el,fd,AE_READABLE|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 +421,51 @@ int syncWithMaster(void) { ...@@ -444,25 +421,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)
{ {
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; goto error;
} }
server.replstate = REDIS_REPL_TRANSFER; server.replstate = REDIS_REPL_TRANSFER;
server.repl_transfer_left = -1; server.repl_transfer_left = -1;
server.repl_transfer_s = fd;
server.repl_transfer_fd = dfd; server.repl_transfer_fd = dfd;
server.repl_transfer_lastio = time(NULL); server.repl_transfer_lastio = time(NULL);
server.repl_transfer_tmpfile = zstrdup(tmpfile); 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_READABLE|AE_WRITABLE,syncWithMaster,NULL) ==
AE_ERR)
{
close(fd);
redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
return REDIS_ERR;
}
server.repl_transfer_s = fd;
server.replstate = REDIS_REPL_CONNECTING;
return REDIS_OK; return REDIS_OK;
} }
...@@ -517,9 +520,8 @@ void replicationCron(void) { ...@@ -517,9 +520,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();
} }
} }
......
...@@ -249,8 +249,11 @@ void sremCommand(redisClient *c) { ...@@ -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) {
......
...@@ -810,11 +810,29 @@ void zaddGenericCommand(redisClient *c, int incr) { ...@@ -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,22 +846,27 @@ void zaddGenericCommand(redisClient *c, int incr) { ...@@ -828,22 +846,27 @@ 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;
} }
} }
for (j = 0; j < elements; j++) {
score = scores[j];
if (zobj->encoding == REDIS_ENCODING_ZIPLIST) { if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
unsigned char *eptr; unsigned char *eptr;
/* Prefer non-encoded element when dealing with ziplists. */ /* Prefer non-encoded element when dealing with ziplists. */
ele = c->argv[3]; ele = c->argv[3+j*2];
if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) { if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
if (incr) { if (incr) {
score += curscore; score += curscore;
if (isnan(score)) { if (isnan(score)) {
addReplyError(c,nanerr); addReplyError(c,nanerr);
/* Don't need to check if the sorted set is empty, because /* Don't need to check if the sorted set is empty
* we know it has at least one element. */ * because we know it has at least one element. */
zfree(scores);
return; return;
} }
} }
...@@ -856,14 +879,9 @@ void zaddGenericCommand(redisClient *c, int incr) { ...@@ -856,14 +879,9 @@ void zaddGenericCommand(redisClient *c, int incr) {
signalModifiedKey(c->db,key); signalModifiedKey(c->db,key);
server.dirty++; server.dirty++;
} }
if (incr) /* ZINCRBY */
addReplyDouble(c,score);
else /* ZADD */
addReply(c,shared.czero);
} else { } else {
/* Optimize: check if the element is too large or the list becomes /* Optimize: check if the element is too large or the list
* too long *before* executing zzlInsert. */ * 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) if (zzlLength(zobj->ptr) > server.zset_max_ziplist_entries)
zsetConvert(zobj,REDIS_ENCODING_SKIPLIST); zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
...@@ -872,18 +890,14 @@ void zaddGenericCommand(redisClient *c, int incr) { ...@@ -872,18 +890,14 @@ void zaddGenericCommand(redisClient *c, int incr) {
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.cone);
} }
} else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) { } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr; zset *zs = zobj->ptr;
zskiplistNode *znode; zskiplistNode *znode;
dictEntry *de; dictEntry *de;
ele = c->argv[3] = tryObjectEncoding(c->argv[3]); ele = c->argv[3+j*2] = tryObjectEncoding(c->argv[3+j*2]);
de = dictFind(zs->dict,ele); de = dictFind(zs->dict,ele);
if (de != NULL) { if (de != NULL) {
curobj = dictGetEntryKey(de); curobj = dictGetEntryKey(de);
...@@ -893,15 +907,16 @@ void zaddGenericCommand(redisClient *c, int incr) { ...@@ -893,15 +907,16 @@ void zaddGenericCommand(redisClient *c, int incr) {
score += curscore; score += curscore;
if (isnan(score)) { if (isnan(score)) {
addReplyError(c,nanerr); addReplyError(c,nanerr);
/* Don't need to check if the sorted set is empty, because /* Don't need to check if the sorted set is empty
* we know it has at least one element. */ * because we know it has at least one element. */
zfree(scores);
return; 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);
...@@ -911,11 +926,6 @@ void zaddGenericCommand(redisClient *c, int incr) { ...@@ -911,11 +926,6 @@ void zaddGenericCommand(redisClient *c, int incr) {
signalModifiedKey(c->db,key); signalModifiedKey(c->db,key);
server.dirty++; server.dirty++;
} }
if (incr) /* ZINCRBY */
addReplyDouble(c,score);
else /* ZADD */
addReply(c,shared.czero);
} else { } else {
znode = zslInsert(zs->zsl,score,ele); znode = zslInsert(zs->zsl,score,ele);
incrRefCount(ele); /* Inserted in skiplist. */ incrRefCount(ele); /* Inserted in skiplist. */
...@@ -924,15 +934,17 @@ void zaddGenericCommand(redisClient *c, int incr) { ...@@ -924,15 +934,17 @@ void zaddGenericCommand(redisClient *c, int incr) {
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.cone);
} }
} else { } else {
redisPanic("Unknown sorted set encoding"); 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) { ...@@ -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) { ...@@ -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++) {
if ((eptr = zzlFind(zobj->ptr,c->argv[j],NULL)) != NULL) {
deleted++;
zobj->ptr = zzlDelete(zobj->ptr,eptr); zobj->ptr = zzlDelete(zobj->ptr,eptr);
if (zzlLength(zobj->ptr) == 0) dbDelete(c->db,key); if (zzlLength(zobj->ptr) == 0) {
} else { dbDelete(c->db,key);
addReply(c,shared.czero); break;
return; }
}
} }
} 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++) {
de = dictFind(zs->dict,c->argv[j]);
if (de != NULL) { if (de != NULL) {
deleted++;
/* Delete from the skiplist */ /* Delete from the skiplist */
score = *(double*)dictGetEntryVal(de); score = *(double*)dictGetEntryVal(de);
redisAssert(zslDelete(zs->zsl,score,ele)); redisAssert(zslDelete(zs->zsl,score,c->argv[j]));
/* Delete from the hash table */ /* Delete from the hash table */
dictDelete(zs->dict,ele); dictDelete(zs->dict,c->argv[j]);
if (htNeedsResize(zs->dict)) dictResize(zs->dict); if (htNeedsResize(zs->dict)) dictResize(zs->dict);
if (dictSize(zs->dict) == 0) dbDelete(c->db,key); if (dictSize(zs->dict) == 0) {
} else { dbDelete(c->db,key);
addReply(c,shared.czero); break;
return; }
}
} }
} else { } else {
redisPanic("Unknown sorted set encoding"); redisPanic("Unknown sorted set encoding");
} }
if (deleted) {
signalModifiedKey(c->db,key); signalModifiedKey(c->db,key);
server.dirty++; server.dirty += deleted;
addReply(c,shared.cone); }
addReplyLongLong(c,deleted);
} }
void zremrangebyscoreCommand(redisClient *c) { void zremrangebyscoreCommand(redisClient *c) {
......
...@@ -110,6 +110,13 @@ proc cleanup {} { ...@@ -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"
......
...@@ -105,6 +105,12 @@ start_server { ...@@ -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]
......
...@@ -48,6 +48,34 @@ start_server {tags {"zset"}} { ...@@ -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"}} { ...@@ -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
......
#!/bin/sh #!/bin/sh
#
# Simple Redis init.d script conceived to work on Linux systems
# as it does use of the /proc filesystem.
REDISPORT=6379 REDISPORT=6379
EXEC=/usr/local/bin/redis-server EXEC=/usr/local/bin/redis-server
CLIEXEC=/usr/local/bin/redis-cli
PIDFILE=/var/run/redis_${REDISPORT}.pid PIDFILE=/var/run/redis_${REDISPORT}.pid
CONF="/etc/redis/${REDISPORT}.conf" CONF="/etc/redis/${REDISPORT}.conf"
...@@ -10,21 +14,21 @@ case "$1" in ...@@ -10,21 +14,21 @@ case "$1" in
start) start)
if [ -f $PIDFILE ] if [ -f $PIDFILE ]
then then
echo -n "$PIDFILE exists, process is already running or crashed\n" echo "$PIDFILE exists, process is already running or crashed"
else else
echo -n "Starting Redis server...\n" echo "Starting Redis server..."
$EXEC $CONF $EXEC $CONF
fi fi
;; ;;
stop) stop)
if [ ! -f $PIDFILE ] if [ ! -f $PIDFILE ]
then then
echo -n "$PIDFILE does not exist, process is not running\n" echo "$PIDFILE does not exist, process is not running"
else else
PID=$(cat $PIDFILE) PID=$(cat $PIDFILE)
echo -n "Stopping ...\n" echo "Stopping ..."
echo -n "SHUTDOWN\r\n" | nc localhost $REDISPORT & $CLIEXEC -p $REDISPORT shutdown
while [ -x /proc/${PIDFILE} ] while [ -x /proc/${PID} ]
do do
echo "Waiting for Redis to shutdown ..." echo "Waiting for Redis to shutdown ..."
sleep 1 sleep 1
...@@ -32,4 +36,7 @@ case "$1" in ...@@ -32,4 +36,7 @@ case "$1" in
echo "Redis stopped" echo "Redis stopped"
fi fi
;; ;;
*)
echo "Please use start or stop as first argument"
;;
esac esac
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