Better SIGCHLD handling for #2897 debugging.

My guess was that wait3() with WNOHANG could never return -1 and an
error. However issue #2897 may possibly indicate that this could happen
under non clear conditions. While we try to understand this better,
better to handle a return value of -1 explicitly, otherwise in the
case a BGREWRITE is in progress but wait3() returns -1, the effect is to
match the first branch of the if/else block since server.rdb_child_pid
is -1, and call backgroundSaveDoneHandler() without a good reason, that
will, in turn, crash the Redis server with an assertion.

00-RELEASENOTES

@@ -1,8 +1,6 @@
 Redis 3.0 release notes
 =======================
 
-WARNING: Redis 3.0 is currently a BETA not suitable for production environments.
-
 --------------------------------------------------------------------------------
 Upgrade urgency levels:
 
@@ -12,6 +10,123 @@ HIGH:     There is a critical bug that may affect a subset of users. Upgrade!
 CRITICAL: There is a critical bug affecting MOST USERS. Upgrade ASAP.
 --------------------------------------------------------------------------------
 
+--[ Redis 3.0.5 ] Release date: 15 Oct 2015
+
+Upgrade urgency: MODERATE, the most important thing is a fix in the replication
+                 code that may make the slave hanging forever if the master
+                 remains with an open socket even if it is no longer able to
+                 reply.
+
+* [FIX] MOVE now moves the TTL as well. A bug lasting forever... finally
+        fixed thanks to Andy Grunwald that reported it.
+        (reported by Andy Grunwald, fixed by Salvatore Sanfilippo)
+* [FIX] Fix a false positive in HSTRLEN test.
+* [FIX] Fix a bug in redis-cli --pipe mode that was not able to read back
+        replies from the server incrementally. Now a mass import will use
+        a lot less memory, and you can use --pipe to do incremental streaming.
+        (reported by Twitter user @fsaintjacques, fixed by Salvatore
+        Sanfilippo)
+* [FIX] Slave detection of master timeout. (fixed by Kevin McGehee, refactoring
+        and regression test by Salvatore Sanfilippo)
+
+* [NEW] Cluster: redis-trib fix can fix an additional case for opens lots.
+        (Salvatore Sanfilippo)
+* [NEW] Cluster: redis-trib import support for --copy and --replace options
+        (David Thomson)
+
+--[ Redis 3.0.4 ] Release date: 8 Sep 2015
+
+Upgrade urgency: HIGH for Redis and Sentinel. However note that in order to
+                 fix certain replication bugs, the replication internals were
+                 modified in a very heavy way. So while this release is
+                 conceptually saner, it may contain regressions. For this
+                 reason, before the release, QA activities were performed by
+                 me (antirez) and Redis Labs and no evident bug was found.
+
+* [FIX] A number of bugs related to replication PSYNC and the (yet experimental)
+        diskless replication feature were fixed. The bugs could lead to
+        inconsistency between masters and slaves. (Salvatore Sanfilippo, Oran
+        Agra fixed the issue found by Yuval Inbar)
+* [FIX] A replication bug in the context of PSYNC partial resynchonization was
+        found and fixed. This bug happens even when diskless replication is off
+        in the case different slaves connect at different times while the master
+        is creating an RDB file, and later a partial resynchronization is
+        attempted by a slave that connected not as the first one. (Salvatore
+        Sanfilippo, Oran Agra)
+* [FIX] Chained replication and PSYNC interactions leading to potential stale
+        chained slaves data set, see issue #2694. (Salvatore Sanfilippo fixed
+        an issue reported by "GeorgeBJ" user at Github)
+* [FIX] redis-cli --scan iteration fixed when returned cursor overflows
+        32 bit signed integer. (Ofir Luzon, Yuval Inbar)
+* [FIX] Sentinel: fixed a bug during the master switch process, where for a
+        failed conditional check, the new configuration is rewritten, during
+        a small window of time, in a corrupted way where the master is
+        also reported to be one of the slaves. This bug is rare to trigger
+        but apparently it happens in the wild, and the effect is to see
+        a replication loop where the master will try to replicate with itself.
+        A detailed explanation of the bug and its effects can be found in
+        the commit message here: https://github.com/antirez/redis/commit/c20218eb5770b2cafb12bc7092313b8358fedc0a.
+        The bug was found by Jan-Erik Rediger using a static analyzer and
+        fixed by Salvatore Sanfilippo.
+* [FIX] Sentinel lack of arity checks for certain commands.
+        (Rogerio Goncalves, Salvatore Sanfilippo)
+
+* [NEW] Replication internals rewritten in order to be more resistant to bugs.
+        The replication handshake in the slave side was rewritten as a non
+        blocking state machine. (Salvatore Sanfilippo, Oran Agra)
+* [NEW] New "replication capabilities" feature introduced in order to signal
+        from the master to the slave what are the features supported, so that
+        the master can choose the kind of replication to start (diskless or
+        not) when master and slave are of different versions. (Oran Agra,
+        Salvatore Sanfilippo)
+* [NEW] Log clients details when SLAVEOF command is received. (Salvatore
+        Sanfilippo with inputs from Nick Craver and Marc Gravell).
+
+--[ Redis 3.0.3 ] Release date: 17 Jul 2015
+
+Upgrade urgency: LOW for Redis and Sentinel.
+
+* [FIX] Fix blocking operations timeout precision when HZ is at its default
+        value (not increased) and there are thousands of clients connected
+        at the same time. This bug affected Sidekiq users that experienced
+        a very long delay for BLPOP and similar commands to return for
+        timeout. Check commit b029ff1 for more info. (Salvatore Sanfilippo)
+* [FIX] MIGRATE "creating socket: Invalid argument" error fix. Check
+        issues #2609 and #2612 for more info. (Salvatore Sanfilippo)
+* [FIX] Be able to connect to the master even when the slave is bound to
+        just the loopback interface and has no valid public address in the
+        network the master is reacahble. (Salvatore Sanfilippo)
+* [FIX] ZADD with options encoding promotion fixed. (linfangrong)
+* [FIX] Reset aof_delayed_fsync on CONFIG RESETSTATS. (Tom Kiemes)
+* [FIX] PFCOUNT key parsing in cluster fixed. (MOON_CLJ)
+* [FIX] Fix Solaris compilation of Redis 3.0. (Jan-Erik Rediger)
+
+* [NEW] Variadic EXISTS command. Now the command accepts multiple arguments
+        and returns the total count of existing keys.
+
+--[ Redis 3.0.2 ] Release date: 4 Jun 2015
+
+Upgrade urgency: HIGH for Redis because of a security issue.
+                 LOW for Sentinel.
+
+* [FIX] Critical security issue fix by Ben Murphy: http://t.co/LpGTyZmfS7
+* [FIX] SMOVE reply fixed when src and dst keys are the same. (Glenn Nethercutt)
+* [FIX] Lua cmsgpack lib updated to support str8 type. (Sebastian Waisbrot)
+
+* [NEW] ZADD support for options: NX, XX, CH. See new doc at redis.io.
+        (Salvatore Sanfilippo)
+* [NEW] Senitnel: CKQUORUM and FLUSHCONFIG commands back ported.
+        (Salvatore Sanfilippo and Bill Anderson)
+
+--[ Redis 3.0.1 ] Release date: 5 May 2015
+
+Upgrade urgency: LOW for Redis and Cluster, MODERATE for Sentinel.
+
+* [FIX] Sentinel memory leak due to hiredis fixed. (Salvatore Sanfilippo)
+* [FIX] Sentinel memory leak on duplicated instance. (Charsyam)
+* [FIX] Redis crash on Lua reaching output buffer limits. (Yossi Gottlieb)
+* [FIX] Sentinel flushes config on +slave events. (Bill Anderson)
+
 --[ Redis 3.0.0 ] Release date: 1 Apr 2015
 
 >> What's new in Redis 3.0 compared to Redis 2.8?
@@ -25,7 +140,7 @@ CRITICAL: There is a critical bug affecting MOST USERS. Upgrade ASAP.
 * WAIT command to block waiting for a write to be transmitted to
   the specified number of slaves.
 * MIGRATE connection caching. Much faster keys migraitons.
-* MIGARTE new options COPY and REPLACE.
+* MIGRATE new options COPY and REPLACE.
 * CLIENT PAUSE command: stop processing client requests for a
   specified amount of time.
 * BITCOUNT performance improvements.
@@ -587,6 +702,11 @@ non-backward compatible changes introduced in the 3.0 release:
   '?' is actually the role of the instance. M for master, S for slave, C
   if this process is a saving child (for RDB/AOF), and X for Sentinel.
 
+* The default maxmemory policy in Redis 3.0 is no longer "volatile-lru" as
+  it used to be in 2.8, but "noeviction". The policies behavior is the same
+  (but LRU eviction is much more precise in 3.0), so only the default value
+  changed. Just make sure to specify in your redis.conf what you mean.
+
 --------------------------------------------------------------------------------
 
 Credits: Where not specified the implementation and design is done by

CONTRIBUTING

@@ -12,15 +12,17 @@ each source file that you contribute.
 
   PLEASE DO NOT POST GENERAL QUESTIONS that are not about bugs or suspected
   bugs in the Github issues system. We'll be very happy to help you and provide
-  all the support in the Redis Google Group.
+  all the support Reddit sub:
 
-  Redis Google Group address:
-  
-      https://groups.google.com/forum/?fromgroups#!forum/redis-db
+      http://reddit.com/r/redis
+
+  There is also an active community of Redis users at Stack Overflow:
+
+      http://stackoverflow.com/questions/tagged/redis
 
 # How to provide a patch for a new feature
 
-1. Drop a message to the Redis Google Group with a proposal of semantics/API.
+1. If it is a major feature or a semantical change, please post it as a new submission in r/redis on Reddit at http://reddit.com/r/redis. Try to be passionate about why the feature is needed, make users upvote your proposal to gain traction and so forth. Read feedbacks about the community. But in this first step **please don't write code yet**.
 
 2. If in step 1 you get an acknowledge from the project leaders, use the
    following procedure to submit a patch:
@@ -31,4 +33,6 @@ each source file that you contribute.
     d. Initiate a pull request on github ( http://help.github.com/send-pull-requests/ )
     e. Done :)
 
+For minor fixes just open a pull request on Github.
+
 Thanks!

COPYING

@@ -1,4 +1,4 @@
-Copyright (c) 2006-2014, Salvatore Sanfilippo
+Copyright (c) 2006-2015, Salvatore Sanfilippo
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:

deps/hiredis/async.c

@@ -443,6 +443,7 @@ void redisProcessCallbacks(redisAsyncContext *ac) {
             if (((redisReply*)reply)->type == REDIS_REPLY_ERROR) {
                 c->err = REDIS_ERR_OTHER;
                 snprintf(c->errstr,sizeof(c->errstr),"%s",((redisReply*)reply)->str);
+                c->reader->fn->freeObject(reply);
                 __redisAsyncDisconnect(ac);
                 return;
             }

deps/lua/src/ldo.c

@@ -495,7 +495,7 @@ static void f_parser (lua_State *L, void *ud) {
   struct SParser *p = cast(struct SParser *, ud);
   int c = luaZ_lookahead(p->z);
   luaC_checkGC(L);
-  tf = ((c == LUA_SIGNATURE[0]) ? luaU_undump : luaY_parser)(L, p->z,
+  tf = (luaY_parser)(L, p->z,
                                                              &p->buff, p->name);
   cl = luaF_newLclosure(L, tf->nups, hvalue(gt(L)));
   cl->l.p = tf;

deps/lua/src/lua_cmsgpack.c

@@ -66,7 +66,7 @@
 /* Reverse memory bytes if arch is little endian. Given the conceptual
  * simplicity of the Lua build system we prefer check for endianess at runtime.
  * The performance difference should be acceptable. */
-static void memrevifle(void *ptr, size_t len) {
+void memrevifle(void *ptr, size_t len) {
     unsigned char   *p = (unsigned char *)ptr,
                     *e = (unsigned char *)p+len-1,
                     aux;
@@ -96,7 +96,7 @@ typedef struct mp_buf {
     size_t len, free;
 } mp_buf;
 
-static void *mp_realloc(lua_State *L, void *target, size_t osize,size_t nsize) {
+void *mp_realloc(lua_State *L, void *target, size_t osize,size_t nsize) {
     void *(*local_realloc) (void *, void *, size_t osize, size_t nsize) = NULL;
     void *ud;
 
@@ -105,7 +105,7 @@ static void *mp_realloc(lua_State *L, void *target, size_t osize,size_t nsize) {
     return local_realloc(ud, target, osize, nsize);
 }
 
-static mp_buf *mp_buf_new(lua_State *L) {
+mp_buf *mp_buf_new(lua_State *L) {
     mp_buf *buf = NULL;
 
     /* Old size = 0; new size = sizeof(*buf) */
@@ -117,7 +117,7 @@ static mp_buf *mp_buf_new(lua_State *L) {
     return buf;
 }
 
-static void mp_buf_append(mp_buf *buf, const unsigned char *s, size_t len) {
+void mp_buf_append(mp_buf *buf, const unsigned char *s, size_t len) {
     if (buf->free < len) {
         size_t newlen = buf->len+len;
 
@@ -153,7 +153,7 @@ typedef struct mp_cur {
     int err;
 } mp_cur;
 
-static void mp_cur_init(mp_cur *cursor, const unsigned char *s, size_t len) {
+void mp_cur_init(mp_cur *cursor, const unsigned char *s, size_t len) {
     cursor->p = s;
     cursor->left = len;
     cursor->err = MP_CUR_ERROR_NONE;
@@ -173,13 +173,17 @@ static void mp_cur_init(mp_cur *cursor, const unsigned char *s, size_t len) {
 
 /* ------------------------- Low level MP encoding -------------------------- */
 
-static void mp_encode_bytes(mp_buf *buf, const unsigned char *s, size_t len) {
+void mp_encode_bytes(mp_buf *buf, const unsigned char *s, size_t len) {
     unsigned char hdr[5];
     int hdrlen;
 
     if (len < 32) {
         hdr[0] = 0xa0 | (len&0xff); /* fix raw */
         hdrlen = 1;
+    } else if (len <= 0xff) {
+        hdr[0] = 0xd9;
+        hdr[1] = len;
+        hdrlen = 2;
     } else if (len <= 0xffff) {
         hdr[0] = 0xda;
         hdr[1] = (len&0xff00)>>8;
@@ -198,7 +202,7 @@ static void mp_encode_bytes(mp_buf *buf, const unsigned char *s, size_t len) {
 }
 
 /* we assume IEEE 754 internal format for single and double precision floats. */
-static void mp_encode_double(mp_buf *buf, double d) {
+void mp_encode_double(mp_buf *buf, double d) {
     unsigned char b[9];
     float f = d;
 
@@ -216,7 +220,7 @@ static void mp_encode_double(mp_buf *buf, double d) {
     }
 }
 
-static void mp_encode_int(mp_buf *buf, int64_t n) {
+void mp_encode_int(mp_buf *buf, int64_t n) {
     unsigned char b[9];
     int enclen;
 
@@ -288,7 +292,7 @@ static void mp_encode_int(mp_buf *buf, int64_t n) {
     mp_buf_append(buf,b,enclen);
 }
 
-static void mp_encode_array(mp_buf *buf, int64_t n) {
+void mp_encode_array(mp_buf *buf, int64_t n) {
     unsigned char b[5];
     int enclen;
 
@@ -311,7 +315,7 @@ static void mp_encode_array(mp_buf *buf, int64_t n) {
     mp_buf_append(buf,b,enclen);
 }
 
-static void mp_encode_map(mp_buf *buf, int64_t n) {
+void mp_encode_map(mp_buf *buf, int64_t n) {
     unsigned char b[5];
     int enclen;
 
@@ -336,7 +340,7 @@ static void mp_encode_map(mp_buf *buf, int64_t n) {
 
 /* --------------------------- Lua types encoding --------------------------- */
 
-static void mp_encode_lua_string(lua_State *L, mp_buf *buf) {
+void mp_encode_lua_string(lua_State *L, mp_buf *buf) {
     size_t len;
     const char *s;
 
@@ -344,13 +348,13 @@ static void mp_encode_lua_string(lua_State *L, mp_buf *buf) {
     mp_encode_bytes(buf,(const unsigned char*)s,len);
 }
 
-static void mp_encode_lua_bool(lua_State *L, mp_buf *buf) {
+void mp_encode_lua_bool(lua_State *L, mp_buf *buf) {
     unsigned char b = lua_toboolean(L,-1) ? 0xc3 : 0xc2;
     mp_buf_append(buf,&b,1);
 }
 
 /* Lua 5.3 has a built in 64-bit integer type */
-static void mp_encode_lua_integer(lua_State *L, mp_buf *buf) {
+void mp_encode_lua_integer(lua_State *L, mp_buf *buf) {
 #if (LUA_VERSION_NUM < 503) && BITS_32
     lua_Number i = lua_tonumber(L,-1);
 #else
@@ -362,7 +366,7 @@ static void mp_encode_lua_integer(lua_State *L, mp_buf *buf) {
 /* Lua 5.2 and lower only has 64-bit doubles, so we need to
  * detect if the double may be representable as an int
  * for Lua < 5.3 */
-static void mp_encode_lua_number(lua_State *L, mp_buf *buf) {
+void mp_encode_lua_number(lua_State *L, mp_buf *buf) {
     lua_Number n = lua_tonumber(L,-1);
 
     if (IS_INT64_EQUIVALENT(n)) {
@@ -372,10 +376,10 @@ static void mp_encode_lua_number(lua_State *L, mp_buf *buf) {
     }
 }
 
-static void mp_encode_lua_type(lua_State *L, mp_buf *buf, int level);
+void mp_encode_lua_type(lua_State *L, mp_buf *buf, int level);
 
 /* Convert a lua table into a message pack list. */
-static void mp_encode_lua_table_as_array(lua_State *L, mp_buf *buf, int level) {
+void mp_encode_lua_table_as_array(lua_State *L, mp_buf *buf, int level) {
 #if LUA_VERSION_NUM < 502
     size_t len = lua_objlen(L,-1), j;
 #else
@@ -391,7 +395,7 @@ static void mp_encode_lua_table_as_array(lua_State *L, mp_buf *buf, int level) {
 }
 
 /* Convert a lua table into a message pack key-value map. */
-static void mp_encode_lua_table_as_map(lua_State *L, mp_buf *buf, int level) {
+void mp_encode_lua_table_as_map(lua_State *L, mp_buf *buf, int level) {
     size_t len = 0;
 
     /* First step: count keys into table. No other way to do it with the
@@ -418,7 +422,7 @@ static void mp_encode_lua_table_as_map(lua_State *L, mp_buf *buf, int level) {
 /* Returns true if the Lua table on top of the stack is exclusively composed
  * of keys from numerical keys from 1 up to N, with N being the total number
  * of elements, without any hole in the middle. */
-static int table_is_an_array(lua_State *L) {
+int table_is_an_array(lua_State *L) {
     int count = 0, max = 0;
 #if LUA_VERSION_NUM < 503
     lua_Number n;
@@ -461,14 +465,14 @@ static int table_is_an_array(lua_State *L) {
 /* If the length operator returns non-zero, that is, there is at least
  * an object at key '1', we serialize to message pack list. Otherwise
  * we use a map. */
-static void mp_encode_lua_table(lua_State *L, mp_buf *buf, int level) {
+void mp_encode_lua_table(lua_State *L, mp_buf *buf, int level) {
     if (table_is_an_array(L))
         mp_encode_lua_table_as_array(L,buf,level);
     else
         mp_encode_lua_table_as_map(L,buf,level);
 }
 
-static void mp_encode_lua_null(lua_State *L, mp_buf *buf) {
+void mp_encode_lua_null(lua_State *L, mp_buf *buf) {
     unsigned char b[1];
     (void)L;
 
@@ -476,7 +480,7 @@ static void mp_encode_lua_null(lua_State *L, mp_buf *buf) {
     mp_buf_append(buf,b,1);
 }
 
-static void mp_encode_lua_type(lua_State *L, mp_buf *buf, int level) {
+void mp_encode_lua_type(lua_State *L, mp_buf *buf, int level) {
     int t = lua_type(L,-1);
 
     /* Limit the encoding of nested tables to a specified maximum depth, so that
@@ -506,7 +510,7 @@ static void mp_encode_lua_type(lua_State *L, mp_buf *buf, int level) {
  * Packs all arguments as a stream for multiple upacking later.
  * Returns error if no arguments provided.
  */
-static int mp_pack(lua_State *L) {
+int mp_pack(lua_State *L) {
     int nargs = lua_gettop(L);
     int i;
     mp_buf *buf;
@@ -687,6 +691,15 @@ void mp_decode_to_lua_type(lua_State *L, mp_cur *c) {
             mp_cur_consume(c,9);
         }
         break;
+    case 0xd9:  /* raw 8 */
+        mp_cur_need(c,2);
+        {
+            size_t l = c->p[1];
+            mp_cur_need(c,2+l);
+            lua_pushlstring(L,(char*)c->p+2,l);
+            mp_cur_consume(c,2+l);
+        }
+        break;
     case 0xda:  /* raw 16 */
         mp_cur_need(c,3);
         {
@@ -773,7 +786,7 @@ void mp_decode_to_lua_type(lua_State *L, mp_cur *c) {
     }
 }
 
-static int mp_unpack_full(lua_State *L, int limit, int offset) {
+int mp_unpack_full(lua_State *L, int limit, int offset) {
     size_t len;
     const char *s;
     mp_cur c;
@@ -826,18 +839,18 @@ static int mp_unpack_full(lua_State *L, int limit, int offset) {
     return cnt;
 }
 
-static int mp_unpack(lua_State *L) {
+int mp_unpack(lua_State *L) {
     return mp_unpack_full(L, 0, 0);
 }
 
-static int mp_unpack_one(lua_State *L) {
+int mp_unpack_one(lua_State *L) {
     int offset = luaL_optinteger(L, 2, 0);
     /* Variable pop because offset may not exist */
     lua_pop(L, lua_gettop(L)-1);
     return mp_unpack_full(L, 1, offset);
 }
 
-static int mp_unpack_limit(lua_State *L) {
+int mp_unpack_limit(lua_State *L) {
     int limit = luaL_checkinteger(L, 2);
     int offset = luaL_optinteger(L, 3, 0);
     /* Variable pop because offset may not exist */
@@ -846,7 +859,7 @@ static int mp_unpack_limit(lua_State *L) {
     return mp_unpack_full(L, limit, offset);
 }
 
-static int mp_safe(lua_State *L) {
+int mp_safe(lua_State *L) {
     int argc, err, total_results;
 
     argc = lua_gettop(L);
@@ -869,7 +882,7 @@ static int mp_safe(lua_State *L) {
 }
 
 /* -------------------------------------------------------------------------- */
-static const struct luaL_Reg cmds[] = {
+const struct luaL_Reg cmds[] = {
     {"pack", mp_pack},
     {"unpack", mp_unpack},
     {"unpack_one", mp_unpack_one},
@@ -877,7 +890,7 @@ static const struct luaL_Reg cmds[] = {
     {0}
 };
 
-static int luaopen_create(lua_State *L) {
+int luaopen_create(lua_State *L) {
     int i;
     /* Manually construct our module table instead of
      * relying on _register or _newlib */

redis.conf

@@ -1,4 +1,9 @@
-# Redis configuration file example
+# Redis configuration file example.
+#
+# Note that in order to read the configuration file, Redis must be
+# started with the file path as first argument:
+#
+# ./redis-server /path/to/redis.conf
 
 # Note on units: when memory size is needed, it is possible to specify
 # it in the usual form of 1k 5GB 4M and so forth:

src/anet.c

@@ -264,6 +264,7 @@ static int anetCreateSocket(char *err, int domain) {
 
 #define ANET_CONNECT_NONE 0
 #define ANET_CONNECT_NONBLOCK 1
+#define ANET_CONNECT_BE_BINDING 2 /* Best effort binding. */
 static int anetTcpGenericConnect(char *err, char *addr, int port,
                                  char *source_addr, int flags)
 {
@@ -295,7 +296,7 @@ static int anetTcpGenericConnect(char *err, char *addr, int port,
             if ((rv = getaddrinfo(source_addr, NULL, &hints, &bservinfo)) != 0)
             {
                 anetSetError(err, "%s", gai_strerror(rv));
-                goto end;
+                goto error;
             }
             for (b = bservinfo; b != NULL; b = b->ai_next) {
                 if (bind(s,b->ai_addr,b->ai_addrlen) != -1) {
@@ -306,7 +307,7 @@ static int anetTcpGenericConnect(char *err, char *addr, int port,
             freeaddrinfo(bservinfo);
             if (!bound) {
                 anetSetError(err, "bind: %s", strerror(errno));
-                goto end;
+                goto error;
             }
         }
         if (connect(s,p->ai_addr,p->ai_addrlen) == -1) {
@@ -331,9 +332,17 @@ error:
         close(s);
         s = ANET_ERR;
     }
+
 end:
     freeaddrinfo(servinfo);
-    return s;
+
+    /* Handle best effort binding: if a binding address was used, but it is
+     * not possible to create a socket, try again without a binding address. */
+    if (s == ANET_ERR && source_addr && (flags & ANET_CONNECT_BE_BINDING)) {
+        return anetTcpGenericConnect(err,addr,port,NULL,flags);
+    } else {
+        return s;
+    }
 }
 
 int anetTcpConnect(char *err, char *addr, int port)
@@ -346,9 +355,18 @@ int anetTcpNonBlockConnect(char *err, char *addr, int port)
     return anetTcpGenericConnect(err,addr,port,NULL,ANET_CONNECT_NONBLOCK);
 }
 
-int anetTcpNonBlockBindConnect(char *err, char *addr, int port, char *source_addr)
+int anetTcpNonBlockBindConnect(char *err, char *addr, int port,
+                               char *source_addr)
 {
-    return anetTcpGenericConnect(err,addr,port,source_addr,ANET_CONNECT_NONBLOCK);
+    return anetTcpGenericConnect(err,addr,port,source_addr,
+            ANET_CONNECT_NONBLOCK);
+}
+
+int anetTcpNonBlockBestEffortBindConnect(char *err, char *addr, int port,
+                                         char *source_addr)
+{
+    return anetTcpGenericConnect(err,addr,port,source_addr,
+            ANET_CONNECT_NONBLOCK|ANET_CONNECT_BE_BINDING);
 }
 
 int anetUnixGenericConnect(char *err, char *path, int flags)

src/anet.h

@@ -50,6 +50,7 @@
 int anetTcpConnect(char *err, char *addr, int port);
 int anetTcpNonBlockConnect(char *err, char *addr, int port);
 int anetTcpNonBlockBindConnect(char *err, char *addr, int port, char *source_addr);
+int anetTcpNonBlockBestEffortBindConnect(char *err, char *addr, int port, char *source_addr);
 int anetUnixConnect(char *err, char *path);
 int anetUnixNonBlockConnect(char *err, char *path);
 int anetRead(int fd, char *buf, int count);

src/aof.c

@@ -1169,9 +1169,9 @@ int rewriteAppendOnlyFile(char *filename) {
     return REDIS_OK;
 
 werr:
+    redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
     fclose(fp);
     unlink(tmpfile);
-    redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
     if (di) dictReleaseIterator(di);
     return REDIS_ERR;
 }

src/bio.c

@@ -152,6 +152,7 @@ void *bioProcessBackgroundJobs(void *arg) {
      * receive the watchdog signal. */
     sigemptyset(&sigset);
     sigaddset(&sigset, SIGALRM);
+    sigaddset(&sigset, SIGCHLD);
     if (pthread_sigmask(SIG_BLOCK, &sigset, NULL))
         redisLog(REDIS_WARNING,
             "Warning: can't mask SIGALRM in bio.c thread: %s", strerror(errno));

src/cluster.c

@@ -358,6 +358,11 @@ void clusterSaveConfigOrDie(int do_fsync) {
  * On success REDIS_OK is returned, otherwise an error is logged and
  * the function returns REDIS_ERR to signal a lock was not acquired. */
 int clusterLockConfig(char *filename) {
+/* flock() does not exist on Solaris
+ * and a fcntl-based solution won't help, as we constantly re-open that file,
+ * which will release _all_ locks anyway
+ */
+#if !defined(__sun)
     /* To lock it, we need to open the file in a way it is created if
      * it does not exist, otherwise there is a race condition with other
      * processes. */
@@ -385,6 +390,8 @@ int clusterLockConfig(char *filename) {
     }
     /* Lock acquired: leak the 'fd' by not closing it, so that we'll retain the
      * lock to the file as long as the process exists. */
+#endif /* __sun */
+
     return REDIS_OK;
 }
 
@@ -2618,7 +2625,9 @@ void clusterLogCantFailover(int reason) {
 
     switch(reason) {
     case REDIS_CLUSTER_CANT_FAILOVER_DATA_AGE:
-        msg = "Disconnected from master for longer than allowed.";
+        msg = "Disconnected from master for longer than allowed. "
+              "Please check the 'cluster-slave-validity-factor' configuration "
+              "option.";
         break;
     case REDIS_CLUSTER_CANT_FAILOVER_WAITING_DELAY:
         msg = "Waiting the delay before I can start a new failover.";
@@ -4486,8 +4495,8 @@ migrateCachedSocket* migrateGetSocket(redisClient *c, robj *host, robj *port, lo
     }
 
     /* Create the socket */
-    fd = anetTcpNonBlockBindConnect(server.neterr,c->argv[1]->ptr,
-                atoi(c->argv[2]->ptr),REDIS_BIND_ADDR);
+    fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr,
+                                atoi(c->argv[2]->ptr));
     if (fd == -1) {
         sdsfree(name);
         addReplyErrorFormat(c,"Can't connect to target node: %s",

src/db.c

@@ -293,13 +293,17 @@ void delCommand(redisClient *c) {
     addReplyLongLong(c,deleted);
 }
 
+/* EXISTS key1 key2 ... key_N.
+ * Return value is the number of keys existing. */
 void existsCommand(redisClient *c) {
-    expireIfNeeded(c->db,c->argv[1]);
-    if (dbExists(c->db,c->argv[1])) {
-        addReply(c, shared.cone);
-    } else {
-        addReply(c, shared.czero);
+    long long count = 0;
+    int j;
+
+    for (j = 1; j < c->argc; j++) {
+        expireIfNeeded(c->db,c->argv[j]);
+        if (dbExists(c->db,c->argv[j])) count++;
     }
+    addReplyLongLong(c,count);
 }
 
 void selectCommand(redisClient *c) {
@@ -710,7 +714,7 @@ void moveCommand(redisClient *c) {
     robj *o;
     redisDb *src, *dst;
     int srcid;
-    long long dbid;
+    long long dbid, expire;
 
     if (server.cluster_enabled) {
         addReplyError(c,"MOVE is not allowed in cluster mode");
@@ -744,6 +748,7 @@ void moveCommand(redisClient *c) {
         addReply(c,shared.czero);
         return;
     }
+    expire = getExpire(c->db,c->argv[1]);
 
     /* Return zero if the key already exists in the target DB */
     if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
@@ -751,6 +756,7 @@ void moveCommand(redisClient *c) {
         return;
     }
     dbAdd(dst,c->argv[1],o);
+    if (expire != -1) setExpire(dst,c->argv[1],expire);
     incrRefCount(o);
 
     /* OK! key moved, free the entry in the source DB */

src/help.h

@@ -1,4 +1,4 @@
-/* Automatically generated by utils/generate-command-help.rb, do not edit. */
+/* Automatically generated by generate-command-help.rb, do not edit. */
 
 #ifndef __REDIS_HELP_H
 #define __REDIS_HELP_H
@@ -15,7 +15,9 @@ static char *commandGroups[] = {
     "connection",
     "server",
     "scripting",
-    "hyperloglog"
+    "hyperloglog",
+    "cluster",
+    "geo"
 };
 
 struct commandHelp {
@@ -46,7 +48,7 @@ struct commandHelp {
     9,
     "1.0.0" },
     { "BITCOUNT",
-    "key [start] [end]",
+    "key [start end]",
     "Count set bits in a string",
     1,
     "2.6.0" },
@@ -81,7 +83,7 @@ struct commandHelp {
     9,
     "2.6.9" },
     { "CLIENT KILL",
-    "ip:port",
+    "[ip:port] [ID client-id] [TYPE normal|slave|pubsub] [ADDR ip:port] [SKIPME yes/no]",
     "Kill the connection of a client",
     9,
     "2.4.0" },
@@ -100,6 +102,116 @@ struct commandHelp {
     "Set the current connection name",
     9,
     "2.6.9" },
+    { "CLUSTER ADDSLOTS",
+    "slot [slot ...]",
+    "Assign new hash slots to receiving node",
+    12,
+    "3.0.0" },
+    { "CLUSTER COUNT-FAILURE-REPORTS",
+    "node-id",
+    "Return the number of failure reports active for a given node",
+    12,
+    "3.0.0" },
+    { "CLUSTER COUNTKEYSINSLOT",
+    "slot",
+    "Return the number of local keys in the specified hash slot",
+    12,
+    "3.0.0" },
+    { "CLUSTER DELSLOTS",
+    "slot [slot ...]",
+    "Set hash slots as unbound in receiving node",
+    12,
+    "3.0.0" },
+    { "CLUSTER FAILOVER",
+    "[FORCE|TAKEOVER]",
+    "Forces a slave to perform a manual failover of its master.",
+    12,
+    "3.0.0" },
+    { "CLUSTER FORGET",
+    "node-id",
+    "Remove a node from the nodes table",
+    12,
+    "3.0.0" },
+    { "CLUSTER GETKEYSINSLOT",
+    "slot count",
+    "Return local key names in the specified hash slot",
+    12,
+    "3.0.0" },
+    { "CLUSTER INFO",
+    "-",
+    "Provides info about Redis Cluster node state",
+    12,
+    "3.0.0" },
+    { "CLUSTER KEYSLOT",
+    "key",
+    "Returns the hash slot of the specified key",
+    12,
+    "3.0.0" },
+    { "CLUSTER MEET",
+    "ip port",
+    "Force a node cluster to handshake with another node",
+    12,
+    "3.0.0" },
+    { "CLUSTER NODES",
+    "-",
+    "Get Cluster config for the node",
+    12,
+    "3.0.0" },
+    { "CLUSTER REPLICATE",
+    "node-id",
+    "Reconfigure a node as a slave of the specified master node",
+    12,
+    "3.0.0" },
+    { "CLUSTER RESET",
+    "[HARD|SOFT]",
+    "Reset a Redis Cluster node",
+    12,
+    "3.0.0" },
+    { "CLUSTER SAVECONFIG",
+    "-",
+    "Forces the node to save cluster state on disk",
+    12,
+    "3.0.0" },
+    { "CLUSTER SET-CONFIG-EPOCH",
+    "config-epoch",
+    "Set the configuration epoch in a new node",
+    12,
+    "3.0.0" },
+    { "CLUSTER SETSLOT",
+    "slot IMPORTING|MIGRATING|STABLE|NODE [node-id]",
+    "Bind an hash slot to a specific node",
+    12,
+    "3.0.0" },
+    { "CLUSTER SLAVES",
+    "node-id",
+    "List slave nodes of the specified master node",
+    12,
+    "3.0.0" },
+    { "CLUSTER SLOTS",
+    "-",
+    "Get array of Cluster slot to node mappings",
+    12,
+    "3.0.0" },
+    { "COMMAND",
+    "-",
+    "Get array of Redis command details",
+    9,
+    "2.8.13" },
+    { "COMMAND COUNT",
+    "-",
+    "Get total number of Redis commands",
+    9,
+    "2.8.13" },
+    { "COMMAND GETKEYS",
+    "-",
+    "Extract keys given a full Redis command",
+    9,
+    "2.8.13" },
+    { "COMMAND INFO",
+    "command-name [command-name ...]",
+    "Get array of specific Redis command details",
+    9,
+    "2.8.13" },
     { "CONFIG GET",
     "parameter",
     "Get the value of a configuration parameter",
@@ -181,7 +293,7 @@ struct commandHelp {
     7,
     "1.2.0" },
     { "EXISTS",
-    "key",
+    "key [key ...]",
     "Determine if a key exists",
     0,
     "1.0.0" },
@@ -205,6 +317,36 @@ struct commandHelp {
     "Remove all keys from the current database",
     9,
     "1.0.0" },
+    { "GEOADD",
+    "key longitude latitude member [longitude latitude member ...]",
+    "Add one or more geospatial items in the geospatial index represented using a sorted set",
+    13,
+    "" },
+    { "GEODIST",
+    "key member1 member2 [unit]",
+    "Returns the distance between two members of a geospatial index",
+    13,
+    "" },
+    { "GEOHASH",
+    "key member [member ...]",
+    "Returns members of a geospatial index as standard geohash strings",
+    13,
+    "" },
+    { "GEOPOS",
+    "key member [member ...]",
+    "Returns longitude and latitude of members of a geospatial index",
+    13,
+    "" },
+    { "GEORADIUS",
+    "key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count]",
+    "Query a sorted set representing a geospatial index to fetch members matching a given maximum distance from a point",
+    13,
+    "" },
+    { "GEORADIUSBYMEMBER",
+    "key member radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count]",
+    "Query a sorted set representing a geospatial index to fetch members matching a given maximum distance from a member",
+    13,
+    "" },
     { "GET",
     "key",
     "Get the value of a key",
@@ -290,6 +432,11 @@ struct commandHelp {
     "Set the value of a hash field, only if the field does not exist",
     5,
     "2.0.0" },
+    { "HSTRLEN",
+    "key field",
+    "Get the length of the value of a hash field",
+    5,
+    "3.2.0" },
     { "HVALS",
     "key",
     "Get all the values in a hash",
@@ -490,6 +637,16 @@ struct commandHelp {
     "Return a random key from the keyspace",
     0,
     "1.0.0" },
+    { "READONLY",
+    "-",
+    "Enables read queries for a connection to a cluster slave node",
+    12,
+    "3.0.0" },
+    { "READWRITE",
+    "-",
+    "Disables read queries for a connection to a cluster slave node",
+    12,
+    "3.0.0" },
     { "RENAME",
     "key newkey",
     "Rename a key",
@@ -501,10 +658,15 @@ struct commandHelp {
     0,
     "1.0.0" },
     { "RESTORE",
-    "key ttl serialized-value",
+    "key ttl serialized-value [REPLACE]",
     "Create a key using the provided serialized value, previously obtained using DUMP.",
     0,
     "2.6.0" },
+    { "ROLE",
+    "-",
+    "Return the role of the instance in the context of replication",
+    9,
+    "2.8.12" },
     { "RPOP",
     "key",
     "Remove and get the last element in a list",
@@ -512,7 +674,7 @@ struct commandHelp {
     "1.0.0" },
     { "RPOPLPUSH",
     "source destination",
-    "Remove the last element in a list, append it to another list and return it",
+    "Remove the last element in a list, prepend it to another list and return it",
     2,
     "1.2.0" },
     { "RPUSH",
@@ -720,13 +882,18 @@ struct commandHelp {
     "Forget about all watched keys",
     7,
     "2.2.0" },
+    { "WAIT",
+    "numslaves timeout",
+    "Wait for the synchronous replication of all the write commands sent in the context of the current connection",
+    0,
+    "3.0.0" },
     { "WATCH",
     "key [key ...]",
     "Watch the given keys to determine execution of the MULTI/EXEC block",
     7,
     "2.2.0" },
     { "ZADD",
-    "key score member [score member ...]",
+    "key [NX|XX] [CH] [INCR] score member [score member ...]",
     "Add one or more members to a sorted set, or update its score if it already exists",
     4,
     "1.2.0" },
@@ -800,6 +967,11 @@ struct commandHelp {
     "Return a range of members in a sorted set, by index, with scores ordered from high to low",
     4,
     "1.2.0" },
+    { "ZREVRANGEBYLEX",
+    "key max min [LIMIT offset count]",
+    "Return a range of members in a sorted set, by lexicographical range, ordered from higher to lower strings.",
+    4,
+    "2.8.9" },
     { "ZREVRANGEBYSCORE",
     "key max min [WITHSCORES] [LIMIT offset count]",
     "Return a range of members in a sorted set, by score, with scores ordered from high to low",

src/networking.c

@@ -105,6 +105,7 @@ redisClient *createClient(int fd) {
     c->repl_ack_off = 0;
     c->repl_ack_time = 0;
     c->slave_listening_port = 0;
+    c->slave_capa = SLAVE_CAPA_NONE;
     c->reply = listCreate();
     c->reply_bytes = 0;
     c->obuf_soft_limit_reached_time = 0;
@@ -666,20 +667,6 @@ void disconnectSlaves(void) {
     }
 }
 
-/* This function is called when the slave lose the connection with the
- * master into an unexpected way. */
-void replicationHandleMasterDisconnection(void) {
-    server.master = NULL;
-    server.repl_state = REDIS_REPL_CONNECT;
-    server.repl_down_since = server.unixtime;
-    /* We lost connection with our master, force our slaves to resync
-     * with us as well to load the new data set.
-     *
-     * If server.masterhost is NULL the user called SLAVEOF NO ONE so
-     * slave resync is not needed. */
-    if (server.masterhost != NULL) disconnectSlaves();
-}
-
 void freeClient(redisClient *c) {
     listNode *ln;
 
@@ -797,7 +784,7 @@ void freeClient(redisClient *c) {
  * a context where calling freeClient() is not possible, because the client
  * should be valid for the continuation of the flow of the program. */
 void freeClientAsync(redisClient *c) {
-    if (c->flags & REDIS_CLOSE_ASAP) return;
+    if (c->flags & REDIS_CLOSE_ASAP || c->flags & REDIS_LUA_CLIENT) return;
     c->flags |= REDIS_CLOSE_ASAP;
     listAddNodeTail(server.clients_to_close,c);
 }
@@ -974,7 +961,7 @@ int processInlineBuffer(redisClient *c) {
 /* Helper function. Trims query buffer to make the function that processes
  * multi bulk requests idempotent. */
 static void setProtocolError(redisClient *c, int pos) {
-    if (server.verbosity >= REDIS_VERBOSE) {
+    if (server.verbosity <= REDIS_VERBOSE) {
         sds client = catClientInfoString(sdsempty(),c);
         redisLog(REDIS_VERBOSE,
             "Protocol error from client: %s", client);
@@ -1679,6 +1666,12 @@ void flushSlavesOutputBuffers(void) {
         redisClient *slave = listNodeValue(ln);
         int events;
 
+        /* Note that the following will not flush output buffers of slaves
+         * in STATE_ONLINE but having put_online_on_ack set to true: in this
+         * case the writable event is never installed, since the purpose
+         * of put_online_on_ack is to postpone the moment it is installed.
+         * This is what we want since slaves in this state should not receive
+         * writes before the first ACK. */
         events = aeGetFileEvents(server.el,slave->fd);
         if (events & AE_WRITABLE &&
             slave->replstate == REDIS_REPL_ONLINE &&

src/rdb.c

@@ -755,9 +755,9 @@ int rdbSave(char *filename) {
     return REDIS_OK;
 
 werr:
+    redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
     fclose(fp);
     unlink(tmpfile);
-    redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
     return REDIS_ERR;
 }
 
@@ -1420,7 +1420,7 @@ int rdbSaveToSlavesSockets(void) {
         if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
             clientids[numfds] = slave->id;
             fds[numfds++] = slave->fd;
-            slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
+            replicationSetupSlaveForFullResync(slave,getPsyncInitialOffset());
             /* Put the socket in non-blocking mode to simplify RDB transfer.
              * We'll restore it when the children returns (since duped socket
              * will share the O_NONBLOCK attribute with the parent). */
@@ -1498,27 +1498,43 @@ int rdbSaveToSlavesSockets(void) {
         exitFromChild((retval == REDIS_OK) ? 0 : 1);
     } else {
         /* Parent */
-        zfree(clientids); /* Not used by parent. Free ASAP. */
         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) {
             redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
                 strerror(errno));
-            zfree(fds);
+
+            /* Undo the state change. The caller will perform cleanup on
+             * all the slaves in BGSAVE_START state, but an early call to
+             * replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
+            listRewind(server.slaves,&li);
+            while((ln = listNext(&li))) {
+                redisClient *slave = ln->value;
+                int j;
+
+                for (j = 0; j < numfds; j++) {
+                    if (slave->id == clientids[j]) {
+                        slave->replstate = REDIS_REPL_WAIT_BGSAVE_START;
+                        break;
+                    }
+                }
+            }
             close(pipefds[0]);
             close(pipefds[1]);
-            return REDIS_ERR;
+        } else {
+            redisLog(REDIS_NOTICE,"Background RDB transfer started by pid %d",
+                childpid);
+            server.rdb_save_time_start = time(NULL);
+            server.rdb_child_pid = childpid;
+            server.rdb_child_type = REDIS_RDB_CHILD_TYPE_SOCKET;
+            updateDictResizePolicy();
         }
-        redisLog(REDIS_NOTICE,"Background RDB transfer started by pid %d",childpid);
-        server.rdb_save_time_start = time(NULL);
-        server.rdb_child_pid = childpid;
-        server.rdb_child_type = REDIS_RDB_CHILD_TYPE_SOCKET;
-        updateDictResizePolicy();
+        zfree(clientids);
         zfree(fds);
-        return REDIS_OK;
+        return (childpid == -1) ? REDIS_ERR : REDIS_OK;
     }
-    return REDIS_OK; /* unreached */
+    return REDIS_OK; /* Unreached. */
 }
 
 void saveCommand(redisClient *c) {

src/redis-cli.c

@@ -1397,6 +1397,7 @@ static void getRDB(void) {
  * Bulk import (pipe) mode
  *--------------------------------------------------------------------------- */
 
+#define PIPEMODE_WRITE_LOOP_MAX_BYTES (128*1024)
 static void pipeMode(void) {
     int fd = context->fd;
     long long errors = 0, replies = 0, obuf_len = 0, obuf_pos = 0;
@@ -1473,6 +1474,8 @@ static void pipeMode(void) {
 
         /* Handle the writable state: we can send protocol to the server. */
         if (mask & AE_WRITABLE) {
+            ssize_t loop_nwritten = 0;
+
             while(1) {
                 /* Transfer current buffer to server. */
                 if (obuf_len != 0) {
@@ -1489,6 +1492,7 @@ static void pipeMode(void) {
                     }
                     obuf_len -= nwritten;
                     obuf_pos += nwritten;
+                    loop_nwritten += nwritten;
                     if (obuf_len != 0) break; /* Can't accept more data. */
                 }
                 /* If buffer is empty, load from stdin. */
@@ -1524,7 +1528,8 @@ static void pipeMode(void) {
                         obuf_pos = 0;
                     }
                 }
-                if (obuf_len == 0 && eof) break;
+                if ((obuf_len == 0 && eof) ||
+                    loop_nwritten > PIPEMODE_WRITE_LOOP_MAX_BYTES) break;
             }
         }
 
@@ -1582,7 +1587,7 @@ static redisReply *sendScan(unsigned long long *it) {
     assert(reply->element[1]->type == REDIS_REPLY_ARRAY);
 
     /* Update iterator */
-    *it = atoi(reply->element[0]->str);
+    *it = strtoull(reply->element[0]->str, NULL, 10);
 
     return reply;
 }

src/redis-trib.rb

@@ -496,6 +496,10 @@ class RedisTrib
         #         importing state in 1 slot. That's trivial to address.
         if migrating.length == 1 && importing.length == 1
             move_slot(migrating[0],importing[0],slot,:verbose=>true,:fix=>true)
+        # Case 2: There are multiple nodes that claim the slot as importing,
+        # they probably got keys about the slot after a restart so opened
+        # the slot. In this case we just move all the keys to the owner
+        # according to the configuration.
         elsif migrating.length == 0 && importing.length > 0
             xputs ">>> Moving all the #{slot} slot keys to its owner #{owner}"
             importing.each {|node|
@@ -504,8 +508,14 @@ class RedisTrib
                 xputs ">>> Setting #{slot} as STABLE in #{node}"
                 node.r.cluster("setslot",slot,"stable")
             }
+        # Case 3: There are no slots claiming to be in importing state, but
+        # there is a migrating node that actually don't have any key. We
+        # can just close the slot, probably a reshard interrupted in the middle.
+        elsif importing.length == 0 && migrating.length == 1 &&
+              migrating[0].r.cluster("getkeysinslot",slot,10).length == 0
+            migrating[0].r.cluster("setslot",slot,"stable")
         else
-            xputs "[ERR] Sorry, Redis-trib can't fix this slot yet (work in progress)"
+            xputs "[ERR] Sorry, Redis-trib can't fix this slot yet (work in progress). Slot is set as migrating in #{migrating.join(",")}, as importing in #{importing.join(",")}, owner is #{owner}"
         end
     end
 
@@ -812,7 +822,7 @@ class RedisTrib
                     source.r.client.call(["migrate",target.info[:host],target.info[:port],key,0,15000])
                 rescue => e
                     if o[:fix] && e.to_s =~ /BUSYKEY/
-                        xputs "*** Target key #{key} exists. Replace it for FIX."
+                        xputs "*** Target key #{key} exists. Replacing it for FIX."
                         source.r.client.call(["migrate",target.info[:host],target.info[:port],key,0,15000,:replace])
                     else
                         puts ""
@@ -1139,7 +1149,9 @@ class RedisTrib
     def import_cluster_cmd(argv,opt)
         source_addr = opt['from']
         xputs ">>> Importing data from #{source_addr} to cluster #{argv[1]}"
-
+        use_copy = opt['copy']
+        use_replace = opt['replace']
+        
         # Check the existing cluster.
         load_cluster_info_from_node(argv[0])
         check_cluster
@@ -1174,7 +1186,10 @@ class RedisTrib
                 print "Migrating #{k} to #{target}: "
                 STDOUT.flush
                 begin
-                    source.client.call(["migrate",target.info[:host],target.info[:port],k,0,15000])
+                    cmd = ["migrate",target.info[:host],target.info[:port],k,0,15000]
+                    cmd << :copy if use_copy
+                    cmd << :replace if use_replace
+                    source.client.call(cmd)
                 rescue => e
                     puts e
                 else
@@ -1334,7 +1349,7 @@ COMMANDS={
 ALLOWED_OPTIONS={
     "create" => {"replicas" => true},
     "add-node" => {"slave" => false, "master-id" => true},
-    "import" => {"from" => :required},
+    "import" => {"from" => :required, "copy" => false, "replace" => false},
     "reshard" => {"from" => true, "to" => true, "slots" => true, "yes" => false}
 }
 

src/redis.c

@@ -129,7 +129,7 @@ struct redisCommand redisCommandTable[] = {
     {"append",appendCommand,3,"wm",0,NULL,1,1,1,0,0},
     {"strlen",strlenCommand,2,"rF",0,NULL,1,1,1,0,0},
     {"del",delCommand,-2,"w",0,NULL,1,-1,1,0,0},
-    {"exists",existsCommand,2,"rF",0,NULL,1,1,1,0,0},
+    {"exists",existsCommand,-2,"rF",0,NULL,1,-1,1,0,0},
     {"setbit",setbitCommand,4,"wm",0,NULL,1,1,1,0,0},
     {"getbit",getbitCommand,3,"rF",0,NULL,1,1,1,0,0},
     {"setrange",setrangeCommand,4,"wm",0,NULL,1,1,1,0,0},
@@ -261,7 +261,7 @@ struct redisCommand redisCommandTable[] = {
     {"cluster",clusterCommand,-2,"ar",0,NULL,0,0,0,0,0},
     {"restore",restoreCommand,-4,"wm",0,NULL,1,1,1,0,0},
     {"restore-asking",restoreCommand,-4,"wmk",0,NULL,1,1,1,0,0},
-    {"migrate",migrateCommand,-6,"w",0,NULL,0,0,0,0,0},
+    {"migrate",migrateCommand,-6,"w",0,NULL,3,3,1,0,0},
     {"asking",askingCommand,1,"r",0,NULL,0,0,0,0,0},
     {"readonly",readonlyCommand,1,"rF",0,NULL,0,0,0,0,0},
     {"readwrite",readwriteCommand,1,"rF",0,NULL,0,0,0,0,0},
@@ -280,7 +280,7 @@ struct redisCommand redisCommandTable[] = {
     {"command",commandCommand,0,"rlt",0,NULL,0,0,0,0,0},
     {"pfselftest",pfselftestCommand,1,"r",0,NULL,0,0,0,0,0},
     {"pfadd",pfaddCommand,-2,"wmF",0,NULL,1,1,1,0,0},
-    {"pfcount",pfcountCommand,-2,"r",0,NULL,1,1,1,0,0},
+    {"pfcount",pfcountCommand,-2,"r",0,NULL,1,-1,1,0,0},
     {"pfmerge",pfmergeCommand,-2,"wm",0,NULL,1,-1,1,0,0},
     {"pfdebug",pfdebugCommand,-3,"w",0,NULL,0,0,0,0,0},
     {"latency",latencyCommand,-2,"arslt",0,NULL,0,0,0,0,0}
@@ -902,9 +902,12 @@ long long getInstantaneousMetric(int metric) {
     return sum / REDIS_METRIC_SAMPLES;
 }
 
-/* Check for timeouts. Returns non-zero if the client was terminated */
-int clientsCronHandleTimeout(redisClient *c) {
-    time_t now = server.unixtime;
+/* Check for timeouts. Returns non-zero if the client was terminated.
+ * The function gets the current time in milliseconds as argument since
+ * it gets called multiple times in a loop, so calling gettimeofday() for
+ * each iteration would be costly without any actual gain. */
+int clientsCronHandleTimeout(redisClient *c, mstime_t now_ms) {
+    time_t now = now_ms/1000;
 
     if (server.maxidletime &&
         !(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
@@ -920,7 +923,6 @@ int clientsCronHandleTimeout(redisClient *c) {
         /* Blocked OPS timeout is handled with milliseconds resolution.
          * However note that the actual resolution is limited by
          * server.hz. */
-        mstime_t now_ms = mstime();
 
         if (c->bpop.timeout != 0 && c->bpop.timeout < now_ms) {
             /* Handle blocking operation specific timeout. */
@@ -962,17 +964,23 @@ int clientsCronResizeQueryBuffer(redisClient *c) {
     return 0;
 }
 
+#define CLIENTS_CRON_MIN_ITERATIONS 5
 void clientsCron(void) {
-    /* Make sure to process at least 1/(server.hz*10) of clients per call.
-     * Since this function is called server.hz times per second we are sure that
-     * in the worst case we process all the clients in 10 seconds.
-     * In normal conditions (a reasonable number of clients) we process
-     * all the clients in a shorter time. */
+    /* Make sure to process at least numclients/server.hz of clients
+     * per call. Since this function is called server.hz times per second
+     * we are sure that in the worst case we process all the clients in 1
+     * second. */
     int numclients = listLength(server.clients);
-    int iterations = numclients/(server.hz*10);
+    int iterations = numclients/server.hz;
+    mstime_t now = mstime();
+
+    /* Process at least a few clients while we are at it, even if we need
+     * to process less than CLIENTS_CRON_MIN_ITERATIONS to meet our contract
+     * of processing each client once per second. */
+    if (iterations < CLIENTS_CRON_MIN_ITERATIONS)
+        iterations = (numclients < CLIENTS_CRON_MIN_ITERATIONS) ?
+                     numclients : CLIENTS_CRON_MIN_ITERATIONS;
 
-    if (iterations < 50)
-        iterations = (numclients < 50) ? numclients : 50;
     while(listLength(server.clients) && iterations--) {
         redisClient *c;
         listNode *head;
@@ -986,7 +994,7 @@ void clientsCron(void) {
         /* The following functions do different service checks on the client.
          * The protocol is that they return non-zero if the client was
          * terminated. */
-        if (clientsCronHandleTimeout(c)) continue;
+        if (clientsCronHandleTimeout(c,now)) continue;
         if (clientsCronResizeQueryBuffer(c)) continue;
     }
 }
@@ -1164,7 +1172,13 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 
             if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc);
 
-            if (pid == server.rdb_child_pid) {
+            if (pid == -1) {
+                redisLog(LOG_WARNING,"wait3() returned an error: %s. "
+                    "rdb_child_pid = %d, aof_child_pid = %d",
+                    strerror(errno),
+                    (int) server.rdb_child_pid,
+                    (int) server.aof_child_pid);
+            } else if (pid == server.rdb_child_pid) {
                 backgroundSaveDoneHandler(exitcode,bysignal);
             } else if (pid == server.aof_child_pid) {
                 backgroundRewriteDoneHandler(exitcode,bysignal);
@@ -1740,13 +1754,15 @@ void resetServerStats(void) {
     }
     server.stat_net_input_bytes = 0;
     server.stat_net_output_bytes = 0;
+    server.aof_delayed_fsync = 0;
 }
 
 void initServer(void) {
     int j;
 
     signal(SIGHUP, SIG_IGN);
-    signal(SIGPIPE, SIG_IGN);
+    signal(SIGPIPE, SIG_IGN); /* No write(2) generated signals. */
+    signal(SIGCHLD, SIG_DFL); /* We want zombies to queue for waitpid(). */
     setupSignalHandlers();
 
     if (server.syslog_enabled) {
@@ -2250,7 +2266,7 @@ int processCommand(redisClient *c) {
         c->cmd->proc != unsubscribeCommand &&
         c->cmd->proc != psubscribeCommand &&
         c->cmd->proc != punsubscribeCommand) {
-        addReplyError(c,"only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context");
+        addReplyError(c,"only (P)SUBSCRIBE / (P)UNSUBSCRIBE / PING / QUIT allowed in this context");
         return REDIS_OK;
     }
 
@@ -3651,6 +3667,7 @@ int main(int argc, char **argv) {
     if (server.daemonize) createPidFile();
     redisSetProcTitle(argv[0]);
     redisAsciiArt();
+    checkTcpBacklogSettings();
 
     if (!server.sentinel_mode) {
         /* Things not needed when running in Sentinel mode. */
@@ -3658,7 +3675,6 @@ int main(int argc, char **argv) {
     #ifdef __linux__
         linuxMemoryWarnings();
     #endif
-        checkTcpBacklogSettings();
         loadDataFromDisk();
         if (server.cluster_enabled) {
             if (verifyClusterConfigWithData() == REDIS_ERR) {

src/redis.h

@@ -267,22 +267,37 @@ typedef long long mstime_t; /* millisecond time type. */
 #define REDIS_CLIENT_TYPE_PUBSUB 2 /* Clients subscribed to PubSub channels. */
 #define REDIS_CLIENT_TYPE_COUNT 3
 
-/* Slave replication state - from the point of view of the slave. */
+/* Slave replication state. Used in server.repl_state for slaves to remember
+ * what to do next. */
 #define REDIS_REPL_NONE 0 /* No active replication */
 #define REDIS_REPL_CONNECT 1 /* Must connect to master */
 #define REDIS_REPL_CONNECTING 2 /* Connecting to master */
+/* --- Handshake states, must be ordered --- */
 #define REDIS_REPL_RECEIVE_PONG 3 /* Wait for PING reply */
-#define REDIS_REPL_TRANSFER 4 /* Receiving .rdb from master */
-#define REDIS_REPL_CONNECTED 5 /* Connected to master */
+#define REDIS_REPL_SEND_AUTH 4 /* Send AUTH to master */
+#define REDIS_REPL_RECEIVE_AUTH 5 /* Wait for AUTH reply */
+#define REDIS_REPL_SEND_PORT 6 /* Send REPLCONF listening-port */
+#define REDIS_REPL_RECEIVE_PORT 7 /* Wait for REPLCONF reply */
+#define REDIS_REPL_SEND_CAPA 8 /* Send REPLCONF capa */
+#define REDIS_REPL_RECEIVE_CAPA 9 /* Wait for REPLCONF reply */
+#define REDIS_REPL_SEND_PSYNC 10 /* Send PSYNC */
+#define REDIS_REPL_RECEIVE_PSYNC 11 /* Wait for PSYNC reply */
+/* --- End of handshake states --- */
+#define REDIS_REPL_TRANSFER 12 /* Receiving .rdb from master */
+#define REDIS_REPL_CONNECTED 13 /* Connected to master */
 
-/* Slave replication state - from the point of view of the master.
+/* State of slaves from the POV of the master. Used in client->replstate.
  * In SEND_BULK and ONLINE state the slave receives new updates
  * in its output queue. In the WAIT_BGSAVE state instead the server is waiting
  * to start the next background saving in order to send updates to it. */
-#define REDIS_REPL_WAIT_BGSAVE_START 6 /* We need to produce a new RDB file. */
-#define REDIS_REPL_WAIT_BGSAVE_END 7 /* Waiting RDB file creation to finish. */
-#define REDIS_REPL_SEND_BULK 8 /* Sending RDB file to slave. */
-#define REDIS_REPL_ONLINE 9 /* RDB file transmitted, sending just updates. */
+#define REDIS_REPL_WAIT_BGSAVE_START 14 /* We need to produce a new RDB file. */
+#define REDIS_REPL_WAIT_BGSAVE_END 15 /* Waiting RDB file creation to finish. */
+#define REDIS_REPL_SEND_BULK 16 /* Sending RDB file to slave. */
+#define REDIS_REPL_ONLINE 17 /* RDB file transmitted, sending just updates. */
+
+/* Slave capabilities. */
+#define SLAVE_CAPA_NONE 0
+#define SLAVE_CAPA_EOF (1<<0)   /* Can parse the RDB EOF streaming format. */
 
 /* Synchronous read timeout - slave side */
 #define REDIS_REPL_SYNCIO_TIMEOUT 5
@@ -543,8 +558,12 @@ typedef struct redisClient {
     long long reploff;      /* replication offset if this is our master */
     long long repl_ack_off; /* replication ack offset, if this is a slave */
     long long repl_ack_time;/* replication ack time, if this is a slave */
+    long long psync_initial_offset; /* FULLRESYNC reply offset other slaves
+                                       copying this slave output buffer
+                                       should use. */
     char replrunid[REDIS_RUN_ID_SIZE+1]; /* master run id if this is a master */
     int slave_listening_port; /* As configured with: SLAVECONF listening-port */
+    int slave_capa;         /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
     multiState mstate;      /* MULTI/EXEC state */
     int btype;              /* Type of blocking op if REDIS_BLOCKED. */
     blockingState bpop;     /* blocking state */
@@ -1171,6 +1190,8 @@ int replicationCountAcksByOffset(long long offset);
 void replicationSendNewlineToMaster(void);
 long long replicationGetSlaveOffset(void);
 char *replicationGetSlaveName(redisClient *c);
+long long getPsyncInitialOffset(void);
+int replicationSetupSlaveForFullResync(redisClient *slave, long long offset);
 
 /* Generic persistence functions */
 void startLoading(FILE *fp);

src/replication.c

@@ -201,6 +201,7 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
         listRewind(slaves,&li);
         while((ln = listNext(&li))) {
             redisClient *slave = ln->value;
+            if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue;
             addReply(slave,selectcmd);
         }
 
@@ -349,6 +350,58 @@ long long addReplyReplicationBacklog(redisClient *c, long long offset) {
     return server.repl_backlog_histlen - skip;
 }
 
+/* Return the offset to provide as reply to the PSYNC command received
+ * from the slave. The returned value is only valid immediately after
+ * the BGSAVE process started and before executing any other command
+ * from clients. */
+long long getPsyncInitialOffset(void) {
+    long long psync_offset = server.master_repl_offset;
+    /* Add 1 to psync_offset if it the replication backlog does not exists
+     * as when it will be created later we'll increment the offset by one. */
+    if (server.repl_backlog == NULL) psync_offset++;
+    return psync_offset;
+}
+
+/* Send a FULLRESYNC reply in the specific case of a full resynchronization,
+ * as a side effect setup the slave for a full sync in different ways:
+ *
+ * 1) Remember, into the slave client structure, the offset we sent
+ *    here, so that if new slaves will later attach to the same
+ *    background RDB saving process (by duplicating this client output
+ *    buffer), we can get the right offset from this slave.
+ * 2) Set the replication state of the slave to WAIT_BGSAVE_END so that
+ *    we start accumulating differences from this point.
+ * 3) Force the replication stream to re-emit a SELECT statement so
+ *    the new slave incremental differences will start selecting the
+ *    right database number.
+ *
+ * Normally this function should be called immediately after a successful
+ * BGSAVE for replication was started, or when there is one already in
+ * progress that we attached our slave to. */
+int replicationSetupSlaveForFullResync(redisClient *slave, long long offset) {
+    char buf[128];
+    int buflen;
+
+    slave->psync_initial_offset = offset;
+    slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
+    /* We are going to accumulate the incremental changes for this
+     * slave as well. Set slaveseldb to -1 in order to force to re-emit
+     * a SLEECT statement in the replication stream. */
+    server.slaveseldb = -1;
+
+    /* Don't send this reply to slaves that approached us with
+     * the old SYNC command. */
+    if (!(slave->flags & REDIS_PRE_PSYNC)) {
+        buflen = snprintf(buf,sizeof(buf),"+FULLRESYNC %s %lld\r\n",
+                          server.runid,offset);
+        if (write(slave->fd,buf,buflen) != buflen) {
+            freeClientAsync(slave);
+            return REDIS_ERR;
+        }
+    }
+    return REDIS_OK;
+}
+
 /* This function handles the PSYNC command from the point of view of a
  * master receiving a request for partial resynchronization.
  *
@@ -422,18 +475,10 @@ int masterTryPartialResynchronization(redisClient *c) {
     return REDIS_OK; /* The caller can return, no full resync needed. */
 
 need_full_resync:
-    /* We need a full resync for some reason... notify the client. */
-    psync_offset = server.master_repl_offset;
-    /* Add 1 to psync_offset if it the replication backlog does not exists
-     * as when it will be created later we'll increment the offset by one. */
-    if (server.repl_backlog == NULL) psync_offset++;
-    /* Again, we can't use the connection buffers (see above). */
-    buflen = snprintf(buf,sizeof(buf),"+FULLRESYNC %s %lld\r\n",
-                      server.runid,psync_offset);
-    if (write(c->fd,buf,buflen) != buflen) {
-        freeClientAsync(c);
-        return REDIS_OK;
-    }
+    /* We need a full resync for some reason... Note that we can't
+     * reply to PSYNC right now if a full SYNC is needed. The reply
+     * must include the master offset at the time the RDB file we transfer
+     * is generated, so we need to delay the reply to that moment. */
     return REDIS_ERR;
 }
 
@@ -441,18 +486,68 @@ need_full_resync:
  * socket target depending on the configuration, and making sure that
  * the script cache is flushed before to start.
  *
+ * The mincapa argument is the bitwise AND among all the slaves capabilities
+ * of the slaves waiting for this BGSAVE, so represents the slave capabilities
+ * all the slaves support. Can be tested via SLAVE_CAPA_* macros.
+ *
+ * Side effects, other than starting a BGSAVE:
+ *
+ * 1) Handle the slaves in WAIT_START state, by preparing them for a full
+ *    sync if the BGSAVE was succesfully started, or sending them an error
+ *    and dropping them from the list of slaves.
+ *
+ * 2) Flush the Lua scripting script cache if the BGSAVE was actually
+ *    started.
+ *
  * Returns REDIS_OK on success or REDIS_ERR otherwise. */
-int startBgsaveForReplication(void) {
+int startBgsaveForReplication(int mincapa) {
     int retval;
+    int socket_target = server.repl_diskless_sync && (mincapa & SLAVE_CAPA_EOF);
+    listIter li;
+    listNode *ln;
 
     redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC with target: %s",
-        server.repl_diskless_sync ? "slaves sockets" : "disk");
+        socket_target ? "slaves sockets" : "disk");
 
-    if (server.repl_diskless_sync)
+    if (socket_target)
         retval = rdbSaveToSlavesSockets();
     else
         retval = rdbSaveBackground(server.rdb_filename);
 
+    /* If we failed to BGSAVE, remove the slaves waiting for a full
+     * resynchorinization from the list of salves, inform them with
+     * an error about what happened, close the connection ASAP. */
+    if (retval == REDIS_ERR) {
+        redisLog(REDIS_WARNING,"BGSAVE for replication failed");
+        listRewind(server.slaves,&li);
+        while((ln = listNext(&li))) {
+            redisClient *slave = ln->value;
+
+            if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
+                slave->flags &= ~REDIS_SLAVE;
+                listDelNode(server.slaves,ln);
+                addReplyError(slave,
+                    "BGSAVE failed, replication can't continue");
+                slave->flags |= REDIS_CLOSE_AFTER_REPLY;
+            }
+        }
+        return retval;
+    }
+
+    /* If the target is socket, rdbSaveToSlavesSockets() already setup
+     * the salves for a full resync. Otherwise for disk target do it now.*/
+    if (!socket_target) {
+        listRewind(server.slaves,&li);
+        while((ln = listNext(&li))) {
+            redisClient *slave = ln->value;
+
+            if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
+                    replicationSetupSlaveForFullResync(slave,
+                            getPsyncInitialOffset());
+            }
+        }
+    }
+
     /* Flush the script cache, since we need that slave differences are
      * accumulated without requiring slaves to match our cached scripts. */
     if (retval == REDIS_OK) replicationScriptCacheFlush();
@@ -515,8 +610,16 @@ void syncCommand(redisClient *c) {
     /* Full resynchronization. */
     server.stat_sync_full++;
 
-    /* Here we need to check if there is a background saving operation
-     * in progress, or if it is required to start one */
+    /* Setup the slave as one waiting for BGSAVE to start. The following code
+     * paths will change the state if we handle the slave differently. */
+    c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
+    if (server.repl_disable_tcp_nodelay)
+        anetDisableTcpNoDelay(NULL, c->fd); /* Non critical if it fails. */
+    c->repldbfd = -1;
+    c->flags |= REDIS_SLAVE;
+    listAddNodeTail(server.slaves,c);
+
+    /* CASE 1: BGSAVE is in progress, with disk target. */
     if (server.rdb_child_pid != -1 &&
         server.rdb_child_type == REDIS_RDB_CHILD_TYPE_DISK)
     {
@@ -532,51 +635,45 @@ void syncCommand(redisClient *c) {
             slave = ln->value;
             if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
         }
-        if (ln) {
+        /* To attach this slave, we check that it has at least all the
+         * capabilities of the slave that triggered the current BGSAVE. */
+        if (ln && ((c->slave_capa & slave->slave_capa) == slave->slave_capa)) {
             /* Perfect, the server is already registering differences for
              * another slave. Set the right state, and copy the buffer. */
             copyClientOutputBuffer(c,slave);
-            c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
+            replicationSetupSlaveForFullResync(c,slave->psync_initial_offset);
             redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
         } else {
             /* No way, we need to wait for the next BGSAVE in order to
              * register differences. */
-            c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
             redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
         }
+
+    /* CASE 2: BGSAVE is in progress, with socket target. */
     } else if (server.rdb_child_pid != -1 &&
                server.rdb_child_type == REDIS_RDB_CHILD_TYPE_SOCKET)
     {
         /* There is an RDB child process but it is writing directly to
          * children sockets. We need to wait for the next BGSAVE
          * in order to synchronize. */
-        c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
         redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
+
+    /* CASE 3: There is no BGSAVE is progress. */
     } else {
-        if (server.repl_diskless_sync) {
+        if (server.repl_diskless_sync && (c->slave_capa & SLAVE_CAPA_EOF)) {
             /* Diskless replication RDB child is created inside
              * replicationCron() since we want to delay its start a
              * few seconds to wait for more slaves to arrive. */
-            c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
             if (server.repl_diskless_sync_delay)
                 redisLog(REDIS_NOTICE,"Delay next BGSAVE for SYNC");
         } else {
-            /* Ok we don't have a BGSAVE in progress, let's start one. */
-            if (startBgsaveForReplication() != REDIS_OK) {
-                redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
-                addReplyError(c,"Unable to perform background save");
-                return;
-            }
-            c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
+            /* Target is disk (or the slave is not capable of supporting
+             * diskless replication) and we don't have a BGSAVE in progress,
+             * let's start one. */
+            if (startBgsaveForReplication(c->slave_capa) != REDIS_OK) return;
         }
     }
 
-    if (server.repl_disable_tcp_nodelay)
-        anetDisableTcpNoDelay(NULL, c->fd); /* Non critical if it fails. */
-    c->repldbfd = -1;
-    c->flags |= REDIS_SLAVE;
-    server.slaveseldb = -1; /* Force to re-emit the SELECT command. */
-    listAddNodeTail(server.slaves,c);
     if (listLength(server.slaves) == 1 && server.repl_backlog == NULL)
         createReplicationBacklog();
     return;
@@ -613,6 +710,10 @@ void replconfCommand(redisClient *c) {
                     &port,NULL) != REDIS_OK))
                 return;
             c->slave_listening_port = port;
+        } else if (!strcasecmp(c->argv[j]->ptr,"capa")) {
+            /* Ignore capabilities not understood by this master. */
+            if (!strcasecmp(c->argv[j+1]->ptr,"eof"))
+                c->slave_capa |= SLAVE_CAPA_EOF;
         } else if (!strcasecmp(c->argv[j]->ptr,"ack")) {
             /* REPLCONF ACK is used by slave to inform the master the amount
              * of replication stream that it processed so far. It is an
@@ -746,6 +847,7 @@ void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
 void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
     listNode *ln;
     int startbgsave = 0;
+    int mincapa = -1;
     listIter li;
 
     listRewind(server.slaves,&li);
@@ -754,7 +856,8 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
 
         if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
             startbgsave = 1;
-            slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
+            mincapa = (mincapa == -1) ? slave->slave_capa :
+                                        (mincapa & slave->slave_capa);
         } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
             struct redis_stat buf;
 
@@ -801,24 +904,18 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
             }
         }
     }
-    if (startbgsave) {
-        if (startBgsaveForReplication() != REDIS_OK) {
-            listIter li;
-
-            listRewind(server.slaves,&li);
-            redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
-            while((ln = listNext(&li))) {
-                redisClient *slave = ln->value;
-
-                if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
-                    freeClient(slave);
-            }
-        }
-    }
+    if (startbgsave) startBgsaveForReplication(mincapa);
 }
 
 /* ----------------------------------- SLAVE -------------------------------- */
 
+/* Returns 1 if the given replication state is a handshake state,
+ * 0 otherwise. */
+int slaveIsInHandshakeState(void) {
+    return server.repl_state >= REDIS_REPL_RECEIVE_PONG &&
+           server.repl_state <= REDIS_REPL_RECEIVE_PSYNC;
+}
+
 /* Abort the async download of the bulk dataset while SYNC-ing with master */
 void replicationAbortSyncTransfer(void) {
     redisAssert(server.repl_state == REDIS_REPL_TRANSFER);
@@ -1062,38 +1159,54 @@ error:
  * The command returns an sds string representing the result of the
  * operation. On error the first byte is a "-".
  */
-char *sendSynchronousCommand(int fd, ...) {
-    va_list ap;
-    sds cmd = sdsempty();
-    char *arg, buf[256];
+#define SYNC_CMD_READ (1<<0)
+#define SYNC_CMD_WRITE (1<<1)
+#define SYNC_CMD_FULL (SYNC_CMD_READ|SYNC_CMD_WRITE)
+char *sendSynchronousCommand(int flags, int fd, ...) {
 
     /* Create the command to send to the master, we use simple inline
      * protocol for simplicity as currently we only send simple strings. */
-    va_start(ap,fd);
-    while(1) {
-        arg = va_arg(ap, char*);
-        if (arg == NULL) break;
+    if (flags & SYNC_CMD_WRITE) {
+        char *arg;
+        va_list ap;
+        sds cmd = sdsempty();
+        va_start(ap,fd);
 
-        if (sdslen(cmd) != 0) cmd = sdscatlen(cmd," ",1);
-        cmd = sdscat(cmd,arg);
-    }
-    cmd = sdscatlen(cmd,"\r\n",2);
+        while(1) {
+            arg = va_arg(ap, char*);
+            if (arg == NULL) break;
 
-    /* Transfer command to the server. */
-    if (syncWrite(fd,cmd,sdslen(cmd),server.repl_syncio_timeout*1000) == -1) {
+            if (sdslen(cmd) != 0) cmd = sdscatlen(cmd," ",1);
+            cmd = sdscat(cmd,arg);
+        }
+        cmd = sdscatlen(cmd,"\r\n",2);
+
+        /* Transfer command to the server. */
+        if (syncWrite(fd,cmd,sdslen(cmd),server.repl_syncio_timeout*1000)
+            == -1)
+        {
+            sdsfree(cmd);
+            return sdscatprintf(sdsempty(),"-Writing to master: %s",
+                    strerror(errno));
+        }
         sdsfree(cmd);
-        return sdscatprintf(sdsempty(),"-Writing to master: %s",
-                strerror(errno));
+        va_end(ap);
     }
-    sdsfree(cmd);
 
     /* Read the reply from the server. */
-    if (syncReadLine(fd,buf,sizeof(buf),server.repl_syncio_timeout*1000) == -1)
-    {
-        return sdscatprintf(sdsempty(),"-Reading from master: %s",
-                strerror(errno));
+    if (flags & SYNC_CMD_READ) {
+        char buf[256];
+
+        if (syncReadLine(fd,buf,sizeof(buf),server.repl_syncio_timeout*1000)
+            == -1)
+        {
+            return sdscatprintf(sdsempty(),"-Reading from master: %s",
+                    strerror(errno));
+        }
+        server.repl_transfer_lastio = server.unixtime;
+        return sdsnew(buf);
     }
-    return sdsnew(buf);
+    return NULL;
 }
 
 /* Try a partial resynchronization with the master if we are about to reconnect.
@@ -1110,6 +1223,19 @@ char *sendSynchronousCommand(int fd, ...) {
  *    of successful partial resynchronization, the function will reuse
  *    'fd' as file descriptor of the server.master client structure.
  *
+ * The function is split in two halves: if read_reply is 0, the function
+ * writes the PSYNC command on the socket, and a new function call is
+ * needed, with read_reply set to 1, in order to read the reply of the
+ * command. This is useful in order to support non blocking operations, so
+ * that we write, return into the event loop, and read when there are data.
+ *
+ * When read_reply is 0 the function returns PSYNC_WRITE_ERR if there
+ * was a write error, or PSYNC_WAIT_REPLY to signal we need another call
+ * with read_reply set to 1. However even when read_reply is set to 1
+ * the function may return PSYNC_WAIT_REPLY again to signal there were
+ * insufficient data to read to complete its work. We should re-enter
+ * into the event loop and wait in such a case.
+ *
  * The function returns:
  *
  * PSYNC_CONTINUE: If the PSYNC command succeded and we can continue.
@@ -1118,35 +1244,68 @@ char *sendSynchronousCommand(int fd, ...) {
  *                   offset is saved.
  * PSYNC_NOT_SUPPORTED: If the server does not understand PSYNC at all and
  *                      the caller should fall back to SYNC.
+ * PSYNC_WRITE_ERR: There was an error writing the command to the socket.
+ * PSYNC_WAIT_REPLY: Call again the function with read_reply set to 1.
+ *
+ * Notable side effects:
+ *
+ * 1) As a side effect of the function call the function removes the readable
+ *    event handler from "fd", unless the return value is PSYNC_WAIT_REPLY.
+ * 2) server.repl_master_initial_offset is set to the right value according
+ *    to the master reply. This will be used to populate the 'server.master'
+ *    structure replication offset.
  */
 
-#define PSYNC_CONTINUE 0
-#define PSYNC_FULLRESYNC 1
-#define PSYNC_NOT_SUPPORTED 2
-int slaveTryPartialResynchronization(int fd) {
+#define PSYNC_WRITE_ERROR 0
+#define PSYNC_WAIT_REPLY 1
+#define PSYNC_CONTINUE 2
+#define PSYNC_FULLRESYNC 3
+#define PSYNC_NOT_SUPPORTED 4
+int slaveTryPartialResynchronization(int fd, int read_reply) {
     char *psync_runid;
     char psync_offset[32];
     sds reply;
 
-    /* Initially set repl_master_initial_offset to -1 to mark the current
-     * master run_id and offset as not valid. Later if we'll be able to do
-     * a FULL resync using the PSYNC command we'll set the offset at the
-     * right value, so that this information will be propagated to the
-     * client structure representing the master into server.master. */
-    server.repl_master_initial_offset = -1;
+    /* Writing half */
+    if (!read_reply) {
+        /* Initially set repl_master_initial_offset to -1 to mark the current
+         * master run_id and offset as not valid. Later if we'll be able to do
+         * a FULL resync using the PSYNC command we'll set the offset at the
+         * right value, so that this information will be propagated to the
+         * client structure representing the master into server.master. */
+        server.repl_master_initial_offset = -1;
 
-    if (server.cached_master) {
-        psync_runid = server.cached_master->replrunid;
-        snprintf(psync_offset,sizeof(psync_offset),"%lld", server.cached_master->reploff+1);
-        redisLog(REDIS_NOTICE,"Trying a partial resynchronization (request %s:%s).", psync_runid, psync_offset);
-    } else {
-        redisLog(REDIS_NOTICE,"Partial resynchronization not possible (no cached master)");
-        psync_runid = "?";
-        memcpy(psync_offset,"-1",3);
+        if (server.cached_master) {
+            psync_runid = server.cached_master->replrunid;
+            snprintf(psync_offset,sizeof(psync_offset),"%lld", server.cached_master->reploff+1);
+            redisLog(REDIS_NOTICE,"Trying a partial resynchronization (request %s:%s).", psync_runid, psync_offset);
+        } else {
+            redisLog(REDIS_NOTICE,"Partial resynchronization not possible (no cached master)");
+            psync_runid = "?";
+            memcpy(psync_offset,"-1",3);
+        }
+
+        /* Issue the PSYNC command */
+        reply = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"PSYNC",psync_runid,psync_offset,NULL);
+        if (reply != NULL) {
+            redisLog(REDIS_WARNING,"Unable to send PSYNC to master: %s",reply);
+            sdsfree(reply);
+            aeDeleteFileEvent(server.el,fd,AE_READABLE);
+            return PSYNC_WRITE_ERROR;
+        }
+        return PSYNC_WAIT_REPLY;
     }
 
-    /* Issue the PSYNC command */
-    reply = sendSynchronousCommand(fd,"PSYNC",psync_runid,psync_offset,NULL);
+    /* Reading half */
+    reply = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
+    if (sdslen(reply) == 0) {
+        /* The master may send empty newlines after it receives PSYNC
+         * and before to reply, just to keep the connection alive. */
+        sdsfree(reply);
+        return PSYNC_WAIT_REPLY;
+    }
+
+    aeDeleteFileEvent(server.el,fd,AE_READABLE);
 
     if (!strncmp(reply,"+FULLRESYNC",11)) {
         char *runid = NULL, *offset = NULL;
@@ -1190,7 +1349,7 @@ int slaveTryPartialResynchronization(int fd) {
         return PSYNC_CONTINUE;
     }
 
-    /* If we reach this point we receied either an error since the master does
+    /* If we reach this point we received either an error since the master does
      * not understand PSYNC, or an unexpected reply from the master.
      * Return PSYNC_NOT_SUPPORTED to the caller in both cases. */
 
@@ -1209,7 +1368,7 @@ int slaveTryPartialResynchronization(int fd) {
 }
 
 void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
-    char tmpfile[256], *err;
+    char tmpfile[256], *err = NULL;
     int dfd, maxtries = 5;
     int sockerr = 0, psync_result;
     socklen_t errlen = sizeof(sockerr);
@@ -1228,16 +1387,12 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
     if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &sockerr, &errlen) == -1)
         sockerr = errno;
     if (sockerr) {
-        aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);
         redisLog(REDIS_WARNING,"Error condition on socket for SYNC: %s",
             strerror(sockerr));
         goto error;
     }
 
-    /* If we were connecting, it's time to send a non blocking PING, we want to
-     * make sure the master is able to reply before going into the actual
-     * replication process where we have long timeouts in the order of
-     * seconds (in the meantime the slave would block). */
+    /* Send a PING to check the master is able to reply without errors. */
     if (server.repl_state == REDIS_REPL_CONNECTING) {
         redisLog(REDIS_NOTICE,"Non blocking connect for SYNC fired the event.");
         /* Delete the writable event so that the readable event remains
@@ -1246,70 +1401,109 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
         server.repl_state = REDIS_REPL_RECEIVE_PONG;
         /* Send the PING, don't check for errors at all, we have the timeout
          * that will take care about this. */
-        syncWrite(fd,"PING\r\n",6,100);
+        err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"PING",NULL);
+        if (err) goto write_error;
         return;
     }
 
     /* Receive the PONG command. */
     if (server.repl_state == REDIS_REPL_RECEIVE_PONG) {
-        char buf[1024];
-
-        /* Delete the readable event, we no longer need it now that there is
-         * the PING reply to read. */
-        aeDeleteFileEvent(server.el,fd,AE_READABLE);
-
-        /* Read the reply with explicit timeout. */
-        buf[0] = '\0';
-        if (syncReadLine(fd,buf,sizeof(buf),
-            server.repl_syncio_timeout*1000) == -1)
-        {
-            redisLog(REDIS_WARNING,
-                "I/O error reading PING reply from master: %s",
-                strerror(errno));
-            goto error;
-        }
+        err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
 
         /* We accept only two replies as valid, a positive +PONG reply
          * (we just check for "+") or an authentication error.
          * Note that older versions of Redis replied with "operation not
          * permitted" instead of using a proper error code, so we test
          * both. */
-        if (buf[0] != '+' &&
-            strncmp(buf,"-NOAUTH",7) != 0 &&
-            strncmp(buf,"-ERR operation not permitted",28) != 0)
+        if (err[0] != '+' &&
+            strncmp(err,"-NOAUTH",7) != 0 &&
+            strncmp(err,"-ERR operation not permitted",28) != 0)
         {
-            redisLog(REDIS_WARNING,"Error reply to PING from master: '%s'",buf);
+            redisLog(REDIS_WARNING,"Error reply to PING from master: '%s'",err);
+            sdsfree(err);
             goto error;
         } else {
             redisLog(REDIS_NOTICE,
                 "Master replied to PING, replication can continue...");
         }
+        sdsfree(err);
+        server.repl_state = REDIS_REPL_SEND_AUTH;
     }
 
     /* AUTH with the master if required. */
-    if(server.masterauth) {
-        err = sendSynchronousCommand(fd,"AUTH",server.masterauth,NULL);
+    if (server.repl_state == REDIS_REPL_SEND_AUTH) {
+        if (server.masterauth) {
+            err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"AUTH",server.masterauth,NULL);
+            if (err) goto write_error;
+            server.repl_state = REDIS_REPL_RECEIVE_AUTH;
+            return;
+        } else {
+            server.repl_state = REDIS_REPL_SEND_PORT;
+        }
+    }
+
+    /* Receive AUTH reply. */
+    if (server.repl_state == REDIS_REPL_RECEIVE_AUTH) {
+        err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
         if (err[0] == '-') {
             redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s",err);
             sdsfree(err);
             goto error;
         }
         sdsfree(err);
+        server.repl_state = REDIS_REPL_SEND_PORT;
     }
 
     /* Set the slave port, so that Master's INFO command can list the
      * slave listening port correctly. */
-    {
+    if (server.repl_state == REDIS_REPL_SEND_PORT) {
         sds port = sdsfromlonglong(server.port);
-        err = sendSynchronousCommand(fd,"REPLCONF","listening-port",port,
-                                         NULL);
+        err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF",
+                "listening-port",port, NULL);
         sdsfree(port);
+        if (err) goto write_error;
+        sdsfree(err);
+        server.repl_state = REDIS_REPL_RECEIVE_PORT;
+        return;
+    }
+
+    /* Receive REPLCONF listening-port reply. */
+    if (server.repl_state == REDIS_REPL_RECEIVE_PORT) {
+        err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
         /* Ignore the error if any, not all the Redis versions support
          * REPLCONF listening-port. */
         if (err[0] == '-') {
-            redisLog(REDIS_NOTICE,"(Non critical) Master does not understand REPLCONF listening-port: %s", err);
+            redisLog(REDIS_NOTICE,"(Non critical) Master does not understand "
+                                  "REPLCONF listening-port: %s", err);
         }
         sdsfree(err);
+        server.repl_state = REDIS_REPL_SEND_CAPA;
+    }
+
+    /* Inform the master of our capabilities. While we currently send
+     * just one capability, it is possible to chain new capabilities here
+     * in the form of REPLCONF capa X capa Y capa Z ...
+     * The master will ignore capabilities it does not understand. */
+    if (server.repl_state == REDIS_REPL_SEND_CAPA) {
+        err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF",
+                "capa","eof",NULL);
+        if (err) goto write_error;
+        sdsfree(err);
+        server.repl_state = REDIS_REPL_RECEIVE_CAPA;
+        return;
+    }
+
+    /* Receive CAPA reply. */
+    if (server.repl_state == REDIS_REPL_RECEIVE_CAPA) {
+        err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
+        /* Ignore the error if any, not all the Redis versions support
+         * REPLCONF capa. */
+        if (err[0] == '-') {
+            redisLog(REDIS_NOTICE,"(Non critical) Master does not understand "
+                                  "REPLCONF capa: %s", err);
+        }
+        sdsfree(err);
+        server.repl_state = REDIS_REPL_SEND_PSYNC;
     }
 
     /* Try a partial resynchonization. If we don't have a cached master
@@ -1317,12 +1511,41 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
      * to start a full resynchronization so that we get the master run id
      * and the global offset, to try a partial resync at the next
      * reconnection attempt. */
-    psync_result = slaveTryPartialResynchronization(fd);
+    if (server.repl_state == REDIS_REPL_SEND_PSYNC) {
+        if (slaveTryPartialResynchronization(fd,0) == PSYNC_WRITE_ERROR) {
+            err = sdsnew("Write error sending the PSYNC command.");
+            goto write_error;
+        }
+        server.repl_state = REDIS_REPL_RECEIVE_PSYNC;
+        return;
+    }
+
+    /* If reached this point, we should be in REDIS_REPL_RECEIVE_PSYNC. */
+    if (server.repl_state != REDIS_REPL_RECEIVE_PSYNC) {
+        redisLog(REDIS_WARNING,"syncWithMaster(): state machine error, "
+                             "state should be RECEIVE_PSYNC but is %d",
+                             server.repl_state);
+        goto error;
+    }
+
+    psync_result = slaveTryPartialResynchronization(fd,1);
+    if (psync_result == PSYNC_WAIT_REPLY) return; /* Try again later... */
+
+    /* Note: if PSYNC does not return WAIT_REPLY, it will take care of
+     * uninstalling the read handler from the file descriptor. */
+
     if (psync_result == PSYNC_CONTINUE) {
         redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Master accepted a Partial Resynchronization.");
         return;
     }
 
+    /* PSYNC failed or is not supported: we want our slaves to resync with us
+     * as well, if we have any (chained replication case). The mater may
+     * transfer us an entirely different data set and we have no way to
+     * incrementally feed our slaves after that. */
+    disconnectSlaves(); /* Force our slaves to resync with us as well. */
+    freeReplicationBacklog(); /* Don't allow our chained slaves to PSYNC. */
+
     /* Fall back to SYNC if needed. Otherwise psync_result == PSYNC_FULLRESYNC
      * and the server.repl_master_runid and repl_master_initial_offset are
      * already populated. */
@@ -1368,16 +1591,22 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
     return;
 
 error:
+    aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);
     close(fd);
     server.repl_transfer_s = -1;
     server.repl_state = REDIS_REPL_CONNECT;
     return;
+
+write_error: /* Handle sendSynchronousCommand(SYNC_CMD_WRITE) errors. */
+    redisLog(REDIS_WARNING,"Sending command to master in replication handshake: %s", err);
+    sdsfree(err);
+    goto error;
 }
 
 int connectWithMaster(void) {
     int fd;
 
-    fd = anetTcpNonBlockBindConnect(NULL,
+    fd = anetTcpNonBlockBestEffortBindConnect(NULL,
         server.masterhost,server.masterport,REDIS_BIND_ADDR);
     if (fd == -1) {
         redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
@@ -1405,7 +1634,7 @@ void undoConnectWithMaster(void) {
     int fd = server.repl_transfer_s;
 
     redisAssert(server.repl_state == REDIS_REPL_CONNECTING ||
-                server.repl_state == REDIS_REPL_RECEIVE_PONG);
+                slaveIsInHandshakeState());
     aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);
     close(fd);
     server.repl_transfer_s = -1;
@@ -1424,7 +1653,7 @@ int cancelReplicationHandshake(void) {
     if (server.repl_state == REDIS_REPL_TRANSFER) {
         replicationAbortSyncTransfer();
     } else if (server.repl_state == REDIS_REPL_CONNECTING ||
-             server.repl_state == REDIS_REPL_RECEIVE_PONG)
+               slaveIsInHandshakeState())
     {
         undoConnectWithMaster();
     } else {
@@ -1470,6 +1699,17 @@ void replicationUnsetMaster(void) {
     server.repl_state = REDIS_REPL_NONE;
 }
 
+/* This function is called when the slave lose the connection with the
+ * master into an unexpected way. */
+void replicationHandleMasterDisconnection(void) {
+    server.master = NULL;
+    server.repl_state = REDIS_REPL_CONNECT;
+    server.repl_down_since = server.unixtime;
+    /* We lost connection with our master, don't disconnect slaves yet,
+     * maybe we'll be able to PSYNC with our master later. We'll disconnect
+     * the slaves only if we'll have to do a full resync with our master. */
+}
+
 void slaveofCommand(redisClient *c) {
     /* SLAVEOF is not allowed in cluster mode as replication is automatically
      * configured using the current address of the master node. */
@@ -1484,7 +1724,10 @@ void slaveofCommand(redisClient *c) {
         !strcasecmp(c->argv[2]->ptr,"one")) {
         if (server.masterhost) {
             replicationUnsetMaster();
-            redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
+            sds client = catClientInfoString(sdsempty(),c);
+            redisLog(REDIS_NOTICE,
+                "MASTER MODE enabled (user request from '%s')",client);
+            sdsfree(client);
         }
     } else {
         long port;
@@ -1502,8 +1745,10 @@ void slaveofCommand(redisClient *c) {
         /* There was no previous master or the user specified a different one,
          * we can continue. */
         replicationSetMaster(c->argv[1]->ptr, port);
-        redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
-            server.masterhost, server.masterport);
+        sds client = catClientInfoString(sdsempty(),c);
+        redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request from '%s')",
+            server.masterhost, server.masterport, client);
+        sdsfree(client);
     }
     addReply(c,shared.ok);
 }
@@ -1543,14 +1788,17 @@ void roleCommand(redisClient *c) {
         addReplyBulkCBuffer(c,"slave",5);
         addReplyBulkCString(c,server.masterhost);
         addReplyLongLong(c,server.masterport);
-        switch(server.repl_state) {
-        case REDIS_REPL_NONE: slavestate = "none"; break;
-        case REDIS_REPL_CONNECT: slavestate = "connect"; break;
-        case REDIS_REPL_CONNECTING: slavestate = "connecting"; break;
-        case REDIS_REPL_RECEIVE_PONG: /* see next */
-        case REDIS_REPL_TRANSFER: slavestate = "sync"; break;
-        case REDIS_REPL_CONNECTED: slavestate = "connected"; break;
-        default: slavestate = "unknown"; break;
+        if (slaveIsInHandshakeState()) {
+            slavestate = "handshake";
+        } else {
+            switch(server.repl_state) {
+            case REDIS_REPL_NONE: slavestate = "none"; break;
+            case REDIS_REPL_CONNECT: slavestate = "connect"; break;
+            case REDIS_REPL_CONNECTING: slavestate = "connecting"; break;
+            case REDIS_REPL_TRANSFER: slavestate = "sync"; break;
+            case REDIS_REPL_CONNECTED: slavestate = "connected"; break;
+            default: slavestate = "unknown"; break;
+            }
         }
         addReplyBulkCString(c,slavestate);
         addReplyLongLong(c,server.master ? server.master->reploff : -1);
@@ -1936,11 +2184,13 @@ long long replicationGetSlaveOffset(void) {
 
 /* Replication cron function, called 1 time per second. */
 void replicationCron(void) {
+    static long long replication_cron_loops = 0;
+
     /* Non blocking connection timeout? */
     if (server.masterhost &&
         (server.repl_state == REDIS_REPL_CONNECTING ||
-         server.repl_state == REDIS_REPL_RECEIVE_PONG) &&
-        (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
+         slaveIsInHandshakeState()) &&
+         (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout)
     {
         redisLog(REDIS_WARNING,"Timeout connecting to the MASTER...");
         undoConnectWithMaster();
@@ -1982,31 +2232,34 @@ void replicationCron(void) {
      * So slaves can implement an explicit timeout to masters, and will
      * be able to detect a link disconnection even if the TCP connection
      * will not actually go down. */
-    if (!(server.cronloops % (server.repl_ping_slave_period * server.hz))) {
-        listIter li;
-        listNode *ln;
-        robj *ping_argv[1];
+    listIter li;
+    listNode *ln;
+    robj *ping_argv[1];
 
-        /* First, send PING */
+    /* First, send PING according to ping_slave_period. */
+    if ((replication_cron_loops % server.repl_ping_slave_period) == 0) {
         ping_argv[0] = createStringObject("PING",4);
-        replicationFeedSlaves(server.slaves, server.slaveseldb, ping_argv, 1);
+        replicationFeedSlaves(server.slaves, server.slaveseldb,
+            ping_argv, 1);
         decrRefCount(ping_argv[0]);
+    }
 
-        /* Second, send a newline to all the slaves in pre-synchronization
-         * stage, that is, slaves waiting for the master to create the RDB file.
-         * The newline will be ignored by the slave but will refresh the
-         * last-io timer preventing a timeout. */
-        listRewind(server.slaves,&li);
-        while((ln = listNext(&li))) {
-            redisClient *slave = ln->value;
+    /* Second, send a newline to all the slaves in pre-synchronization
+     * stage, that is, slaves waiting for the master to create the RDB file.
+     * The newline will be ignored by the slave but will refresh the
+     * last-io timer preventing a timeout. In this case we ignore the
+     * ping period and refresh the connection once per second since certain
+     * timeouts are set at a few seconds (example: PSYNC response). */
+    listRewind(server.slaves,&li);
+    while((ln = listNext(&li))) {
+        redisClient *slave = ln->value;
 
-            if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START ||
-                (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END &&
-                 server.rdb_child_type != REDIS_RDB_CHILD_TYPE_SOCKET))
-            {
-                if (write(slave->fd, "\n", 1) == -1) {
-                    /* Don't worry, it's just a ping. */
-                }
+        if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START ||
+            (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END &&
+             server.rdb_child_type != REDIS_RDB_CHILD_TYPE_SOCKET))
+        {
+            if (write(slave->fd, "\n", 1) == -1) {
+                /* Don't worry, it's just a ping. */
             }
         }
     }
@@ -2067,6 +2320,7 @@ void replicationCron(void) {
     if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) {
         time_t idle, max_idle = 0;
         int slaves_waiting = 0;
+        int mincapa = -1;
         listNode *ln;
         listIter li;
 
@@ -2077,28 +2331,19 @@ void replicationCron(void) {
                 idle = server.unixtime - slave->lastinteraction;
                 if (idle > max_idle) max_idle = idle;
                 slaves_waiting++;
+                mincapa = (mincapa == -1) ? slave->slave_capa :
+                                            (mincapa & slave->slave_capa);
             }
         }
 
         if (slaves_waiting && max_idle > server.repl_diskless_sync_delay) {
             /* Start a BGSAVE. Usually with socket target, or with disk target
              * if there was a recent socket -> disk config change. */
-            if (startBgsaveForReplication() == REDIS_OK) {
-                /* It started! We need to change the state of slaves
-                 * from WAIT_BGSAVE_START to WAIT_BGSAVE_END in case
-                 * the current target is disk. Otherwise it was already done
-                 * by rdbSaveToSlavesSockets() which is called by
-                 * startBgsaveForReplication(). */
-                listRewind(server.slaves,&li);
-                while((ln = listNext(&li))) {
-                    redisClient *slave = ln->value;
-                    if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
-                        slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
-                }
-            }
+            startBgsaveForReplication(mincapa);
         }
     }
 
     /* Refresh the number of slaves with lag <= min-slaves-max-lag. */
     refreshGoodSlavesCount();
+    replication_cron_loops++; /* Incremented with frequency 1 HZ. */
 }

src/scripting.c

@@ -612,11 +612,12 @@ void scriptingEnableGlobalsProtection(lua_State *lua) {
 
     /* strict.lua from: http://metalua.luaforge.net/src/lib/strict.lua.html.
      * Modified to be adapted to Redis. */
+    s[j++]="local dbg=debug\n";
     s[j++]="local mt = {}\n";
     s[j++]="setmetatable(_G, mt)\n";
     s[j++]="mt.__newindex = function (t, n, v)\n";
-    s[j++]="  if debug.getinfo(2) then\n";
-    s[j++]="    local w = debug.getinfo(2, \"S\").what\n";
+    s[j++]="  if dbg.getinfo(2) then\n";
+    s[j++]="    local w = dbg.getinfo(2, \"S\").what\n";
     s[j++]="    if w ~= \"main\" and w ~= \"C\" then\n";
     s[j++]="      error(\"Script attempted to create global variable '\"..tostring(n)..\"'\", 2)\n";
     s[j++]="    end\n";
@@ -624,11 +625,12 @@ void scriptingEnableGlobalsProtection(lua_State *lua) {
     s[j++]="  rawset(t, n, v)\n";
     s[j++]="end\n";
     s[j++]="mt.__index = function (t, n)\n";
-    s[j++]="  if debug.getinfo(2) and debug.getinfo(2, \"S\").what ~= \"C\" then\n";
+    s[j++]="  if dbg.getinfo(2) and dbg.getinfo(2, \"S\").what ~= \"C\" then\n";
     s[j++]="    error(\"Script attempted to access unexisting global variable '\"..tostring(n)..\"'\", 2)\n";
     s[j++]="  end\n";
     s[j++]="  return rawget(t, n)\n";
     s[j++]="end\n";
+    s[j++]="debug = nil\n";
     s[j++]=NULL;
 
     for (j = 0; s[j] != NULL; j++) code = sdscatlen(code,s[j],strlen(s[j]));
@@ -732,10 +734,11 @@ void scriptingInit(void) {
      * information about the caller, that's what makes sense from the point
      * of view of the user debugging a script. */
     {
-        char *errh_func =       "function __redis__err__handler(err)\n"
-                                "  local i = debug.getinfo(2,'nSl')\n"
+        char *errh_func =       "local dbg = debug\n"
+                                "function __redis__err__handler(err)\n"
+                                "  local i = dbg.getinfo(2,'nSl')\n"
                                 "  if i and i.what == 'C' then\n"
-                                "    i = debug.getinfo(3,'nSl')\n"
+                                "    i = dbg.getinfo(3,'nSl')\n"
                                 "  end\n"
                                 "  if i then\n"
                                 "    return i.source .. ':' .. i.currentline .. ': ' .. err\n"

src/sds.c

@@ -71,7 +71,7 @@ sds sdsempty(void) {
     return sdsnewlen("",0);
 }
 
-/* Create a new sds string starting from a null termined C string. */
+/* Create a new sds string starting from a null terminated C string. */
 sds sdsnew(const char *init) {
     size_t initlen = (init == NULL) ? 0 : strlen(init);
     return sdsnewlen(init, initlen);
@@ -557,7 +557,7 @@ sds sdscatfmt(sds s, char const *fmt, ...) {
  * Example:
  *
  * s = sdsnew("AA...AA.a.aa.aHelloWorld     :::");
- * s = sdstrim(s,"A. :");
+ * s = sdstrim(s,"Aa. :");
  * printf("%s\n", s);
  *
  * Output will be just "Hello World".
@@ -1083,6 +1083,7 @@ int main(void) {
             int oldfree;
 
             sdsfree(x);
+            sdsfree(y);
             x = sdsnew("0");
             sh = (void*) (x-(sizeof(struct sdshdr)));
             test_cond("sdsnew() free/len buffers", sh->len == 1 && sh->free == 0);
@@ -1095,6 +1096,8 @@ int main(void) {
             test_cond("sdsIncrLen() -- content", x[0] == '0' && x[1] == '1');
             test_cond("sdsIncrLen() -- len", sh->len == 2);
             test_cond("sdsIncrLen() -- free", sh->free == oldfree-1);
+
+            sdsfree(x);
         }
     }
     test_report()

src/sentinel.c

@@ -923,6 +923,7 @@ sentinelRedisInstance *createSentinelRedisInstance(char *name, int flags, char *
     else if (flags & SRI_SENTINEL) table = master->sentinels;
     sdsname = sdsnew(name);
     if (dictFind(table,sdsname)) {
+        releaseSentinelAddr(addr);
         sdsfree(sdsname);
         errno = EBUSY;
         return NULL;
@@ -1156,7 +1157,9 @@ void sentinelDelFlagsToDictOfRedisInstances(dict *instances, int flags) {
  * 1) Remove all slaves.
  * 2) Remove all sentinels.
  * 3) Remove most of the flags resulting from runtime operations.
- * 4) Reset timers to their default value.
+ * 4) Reset timers to their default value. For example after a reset it will be
+ *    possible to failover again the same master ASAP, without waiting the
+ *    failover timeout delay.
  * 5) In the process of doing this undo the failover if in progress.
  * 6) Disconnect the connections with the master (will reconnect automatically).
  */
@@ -1179,7 +1182,7 @@ void sentinelResetMaster(sentinelRedisInstance *ri, int flags) {
     }
     ri->failover_state = SENTINEL_FAILOVER_STATE_NONE;
     ri->failover_state_change_time = 0;
-    ri->failover_start_time = 0;
+    ri->failover_start_time = 0; /* We can failover again ASAP. */
     ri->promoted_slave = NULL;
     sdsfree(ri->runid);
     sdsfree(ri->slave_master_host);
@@ -1269,10 +1272,7 @@ int sentinelResetMasterAndChangeAddress(sentinelRedisInstance *master, char *ip,
         slave = createSentinelRedisInstance(NULL,SRI_SLAVE,slaves[j]->ip,
                     slaves[j]->port, master->quorum, master);
         releaseSentinelAddr(slaves[j]);
-        if (slave) {
-            sentinelEvent(REDIS_NOTICE,"+slave",slave,"%@");
-            sentinelFlushConfig();
-        }
+        if (slave) sentinelEvent(REDIS_NOTICE,"+slave",slave,"%@");
     }
     zfree(slaves);
 
@@ -1546,7 +1546,7 @@ void rewriteConfigSentinelOption(struct rewriteConfigState *state) {
                 slave_addr = master->addr;
             line = sdscatprintf(sdsempty(),
                 "sentinel known-slave %s %s %d",
-                master->name, ri->addr->ip, ri->addr->port);
+                master->name, slave_addr->ip, slave_addr->port);
             rewriteConfigRewriteLine(state,"sentinel",line,1);
         }
         dictReleaseIterator(di2);
@@ -1844,6 +1844,7 @@ void sentinelRefreshInstanceInfo(sentinelRedisInstance *ri, const char *info) {
                             atoi(port), ri->quorum, ri)) != NULL)
                 {
                     sentinelEvent(REDIS_NOTICE,"+slave",slave,"%@");
+                    sentinelFlushConfig();
                 }
             }
         }
@@ -2615,6 +2616,31 @@ sentinelRedisInstance *sentinelGetMasterByNameOrReplyError(redisClient *c,
     return ri;
 }
 
+#define SENTINEL_ISQR_OK 0
+#define SENTINEL_ISQR_NOQUORUM (1<<0)
+#define SENTINEL_ISQR_NOAUTH (1<<1)
+int sentinelIsQuorumReachable(sentinelRedisInstance *master, int *usableptr) {
+    dictIterator *di;
+    dictEntry *de;
+    int usable = 1; /* Number of usable Sentinels. Init to 1 to count myself. */
+    int result = SENTINEL_ISQR_OK;
+    int voters = dictSize(master->sentinels)+1; /* Known Sentinels + myself. */
+
+    di = dictGetIterator(master->sentinels);
+    while((de = dictNext(di)) != NULL) {
+        sentinelRedisInstance *ri = dictGetVal(de);
+
+        if (ri->flags & (SRI_S_DOWN|SRI_O_DOWN)) continue;
+        usable++;
+    }
+    dictReleaseIterator(di);
+
+    if (usable < (int)master->quorum) result |= SENTINEL_ISQR_NOQUORUM;
+    if (usable < voters/2+1) result |= SENTINEL_ISQR_NOAUTH;
+    if (usableptr) *usableptr = usable;
+    return result;
+}
+
 void sentinelCommand(redisClient *c) {
     if (!strcasecmp(c->argv[1]->ptr,"masters")) {
         /* SENTINEL MASTERS */
@@ -2765,16 +2791,49 @@ void sentinelCommand(redisClient *c) {
             sentinelEvent(REDIS_WARNING,"+monitor",ri,"%@ quorum %d",ri->quorum);
             addReply(c,shared.ok);
         }
+    } else if (!strcasecmp(c->argv[1]->ptr,"flushconfig")) {
+        if (c->argc != 2) goto numargserr;
+        sentinelFlushConfig();
+        addReply(c,shared.ok);
+        return;
     } else if (!strcasecmp(c->argv[1]->ptr,"remove")) {
         /* SENTINEL REMOVE <name> */
         sentinelRedisInstance *ri;
 
+        if (c->argc != 3) goto numargserr;
         if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2]))
             == NULL) return;
         sentinelEvent(REDIS_WARNING,"-monitor",ri,"%@");
         dictDelete(sentinel.masters,c->argv[2]->ptr);
         sentinelFlushConfig();
         addReply(c,shared.ok);
+    } else if (!strcasecmp(c->argv[1]->ptr,"ckquorum")) {
+        /* SENTINEL CKQUORUM <name> */
+        sentinelRedisInstance *ri;
+        int usable;
+
+        if (c->argc != 3) goto numargserr;
+        if ((ri = sentinelGetMasterByNameOrReplyError(c,c->argv[2]))
+            == NULL) return;
+        int result = sentinelIsQuorumReachable(ri,&usable);
+        if (result == SENTINEL_ISQR_OK) {
+            addReplySds(c, sdscatfmt(sdsempty(),
+                "+OK %i usable Sentinels. Quorum and failover authorization "
+                "can be reached\r\n",usable));
+        } else {
+            sds e = sdscatfmt(sdsempty(),
+                "-NOQUORUM %i usable Sentinels. ",usable);
+            if (result & SENTINEL_ISQR_NOQUORUM)
+                e = sdscat(e,"Not enough available Sentinels to reach the"
+                             " specified quorum for this master");
+            if (result & SENTINEL_ISQR_NOAUTH) {
+                if (result & SENTINEL_ISQR_NOQUORUM) e = sdscat(e,". ");
+                e = sdscat(e, "Not enough available Sentinels to reach the"
+                              " majority and authorize a failover");
+            }
+            e = sdscat(e,"\r\n");
+            addReplySds(c,e);
+        }
     } else if (!strcasecmp(c->argv[1]->ptr,"set")) {
         if (c->argc < 3 || c->argc % 2 == 0) goto numargserr;
         sentinelSetCommand(c);

src/sha1.c

@@ -23,7 +23,7 @@ A million repetitions of "a"
 
 #include <stdio.h>
 #include <string.h>
-#include <sys/types.h>	/* for u_int*_t */
+#include <stdint.h>
 #include "solarisfixes.h"
 #include "sha1.h"
 #include "config.h"
@@ -53,12 +53,12 @@ A million repetitions of "a"
 
 /* Hash a single 512-bit block. This is the core of the algorithm. */
 
-void SHA1Transform(u_int32_t state[5], const unsigned char buffer[64])
+void SHA1Transform(uint32_t state[5], const unsigned char buffer[64])
 {
-    u_int32_t a, b, c, d, e;
+    uint32_t a, b, c, d, e;
     typedef union {
         unsigned char c[64];
-        u_int32_t l[16];
+        uint32_t l[16];
     } CHAR64LONG16;
 #ifdef SHA1HANDSOFF
     CHAR64LONG16 block[1];  /* use array to appear as a pointer */
@@ -128,9 +128,9 @@ void SHA1Init(SHA1_CTX* context)
 
 /* Run your data through this. */
 
-void SHA1Update(SHA1_CTX* context, const unsigned char* data, u_int32_t len)
+void SHA1Update(SHA1_CTX* context, const unsigned char* data, uint32_t len)
 {
-    u_int32_t i, j;
+    uint32_t i, j;
 
     j = context->count[0];
     if ((context->count[0] += len << 3) < j)
@@ -168,7 +168,7 @@ void SHA1Final(unsigned char digest[20], SHA1_CTX* context)
 
     for (i = 0; i < 2; i++)
        {
-        u_int32_t t = context->count[i];
+        uint32_t t = context->count[i];
         int j;
 
         for (j = 0; j < 4; t >>= 8, j++)

src/sha1.h

@@ -6,12 +6,12 @@ By Steve Reid <steve@edmweb.com>
 */
 
 typedef struct {
-    u_int32_t state[5];
-    u_int32_t count[2];
+    uint32_t state[5];
+    uint32_t count[2];
     unsigned char buffer[64];
 } SHA1_CTX;
 
-void SHA1Transform(u_int32_t state[5], const unsigned char buffer[64]);
+void SHA1Transform(uint32_t state[5], const unsigned char buffer[64]);
 void SHA1Init(SHA1_CTX* context);
-void SHA1Update(SHA1_CTX* context, const unsigned char* data, u_int32_t len);
+void SHA1Update(SHA1_CTX* context, const unsigned char* data, uint32_t len);
 void SHA1Final(unsigned char digest[20], SHA1_CTX* context);

src/sort.c

@@ -209,7 +209,7 @@ void sortCommand(redisClient *c) {
     }
 
     /* Create a list of operations to perform for every sorted element.
-     * Operations can be GET/DEL/INCR/DECR */
+     * Operations can be GET */
     operations = listCreate();
     listSetFreeMethod(operations,zfree);
     j = 2; /* options start at argv[2] */

src/t_set.c

@@ -321,7 +321,7 @@ void smoveCommand(redisClient *c) {
 
     /* If srcset and dstset are equal, SMOVE is a no-op */
     if (srcset == dstset) {
-        addReply(c,shared.cone);
+        addReply(c,setTypeIsMember(srcset,ele) ? shared.cone : shared.czero);
         return;
     }
 

src/t_zset.c

@@ -1171,35 +1171,84 @@ void zsetConvert(robj *zobj, int encoding) {
  *----------------------------------------------------------------------------*/
 
 /* This generic command implements both ZADD and ZINCRBY. */
-void zaddGenericCommand(redisClient *c, int incr) {
+#define ZADD_NONE 0
+#define ZADD_INCR (1<<0)    /* Increment the score instead of setting it. */
+#define ZADD_NX (1<<1)      /* Don't touch elements not already existing. */
+#define ZADD_XX (1<<2)      /* Only touch elements already exisitng. */
+#define ZADD_CH (1<<3)      /* Return num of elements added or updated. */
+void zaddGenericCommand(redisClient *c, int flags) {
     static char *nanerr = "resulting score is not a number (NaN)";
     robj *key = c->argv[1];
     robj *ele;
     robj *zobj;
     robj *curobj;
     double score = 0, *scores = NULL, curscore = 0.0;
-    int j, elements = (c->argc-2)/2;
-    int added = 0, updated = 0;
+    int j, elements;
+    int scoreidx = 0;
+    /* The following vars are used in order to track what the command actually
+     * did during the execution, to reply to the client and to trigger the
+     * notification of keyspace change. */
+    int added = 0;      /* Number of new elements added. */
+    int updated = 0;    /* Number of elements with updated score. */
+    int processed = 0;  /* Number of elements processed, may remain zero with
+                           options like XX. */
 
-    if (c->argc % 2) {
+    /* Parse options. At the end 'scoreidx' is set to the argument position
+     * of the score of the first score-element pair. */
+    scoreidx = 2;
+    while(scoreidx < c->argc) {
+        char *opt = c->argv[scoreidx]->ptr;
+        if (!strcasecmp(opt,"nx")) flags |= ZADD_NX;
+        else if (!strcasecmp(opt,"xx")) flags |= ZADD_XX;
+        else if (!strcasecmp(opt,"ch")) flags |= ZADD_CH;
+        else if (!strcasecmp(opt,"incr")) flags |= ZADD_INCR;
+        else break;
+        scoreidx++;
+    }
+
+    /* Turn options into simple to check vars. */
+    int incr = (flags & ZADD_INCR) != 0;
+    int nx = (flags & ZADD_NX) != 0;
+    int xx = (flags & ZADD_XX) != 0;
+    int ch = (flags & ZADD_CH) != 0;
+
+    /* After the options, we expect to have an even number of args, since
+     * we expect any number of score-element pairs. */
+    elements = c->argc-scoreidx;
+    if (elements % 2) {
         addReply(c,shared.syntaxerr);
         return;
     }
+    elements /= 2; /* Now this holds the number of score-element pairs. */
+
+    /* Check for incompatible options. */
+    if (nx && xx) {
+        addReplyError(c,
+            "XX and NX options at the same time are not compatible");
+        return;
+    }
+
+    if (incr && elements > 1) {
+        addReplyError(c,
+            "INCR option supports a single increment-element pair");
+        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)
+        if (getDoubleFromObjectOrReply(c,c->argv[scoreidx+j*2],&scores[j],NULL)
             != REDIS_OK) goto cleanup;
     }
 
     /* Lookup the key and create the sorted set if does not exist. */
     zobj = lookupKeyWrite(c->db,key);
     if (zobj == NULL) {
+        if (xx) goto reply_to_client; /* No key + XX option: nothing to do. */
         if (server.zset_max_ziplist_entries == 0 ||
-            server.zset_max_ziplist_value < sdslen(c->argv[3]->ptr))
+            server.zset_max_ziplist_value < sdslen(c->argv[scoreidx+1]->ptr))
         {
             zobj = createZsetObject();
         } else {
@@ -1220,8 +1269,9 @@ void zaddGenericCommand(redisClient *c, int incr) {
             unsigned char *eptr;
 
             /* Prefer non-encoded element when dealing with ziplists. */
-            ele = c->argv[3+j*2];
+            ele = c->argv[scoreidx+1+j*2];
             if ((eptr = zzlFind(zobj->ptr,ele,&curscore)) != NULL) {
+                if (nx) continue;
                 if (incr) {
                     score += curscore;
                     if (isnan(score)) {
@@ -1237,7 +1287,8 @@ void zaddGenericCommand(redisClient *c, int incr) {
                     server.dirty++;
                     updated++;
                 }
-            } else {
+                processed++;
+            } else if (!xx) {
                 /* Optimize: check if the element is too large or the list
                  * becomes too long *before* executing zzlInsert. */
                 zobj->ptr = zzlInsert(zobj->ptr,ele,score);
@@ -1247,15 +1298,18 @@ void zaddGenericCommand(redisClient *c, int incr) {
                     zsetConvert(zobj,REDIS_ENCODING_SKIPLIST);
                 server.dirty++;
                 added++;
+                processed++;
             }
         } else if (zobj->encoding == REDIS_ENCODING_SKIPLIST) {
             zset *zs = zobj->ptr;
             zskiplistNode *znode;
             dictEntry *de;
 
-            ele = c->argv[3+j*2] = tryObjectEncoding(c->argv[3+j*2]);
+            ele = c->argv[scoreidx+1+j*2] =
+                tryObjectEncoding(c->argv[scoreidx+1+j*2]);
             de = dictFind(zs->dict,ele);
             if (de != NULL) {
+                if (nx) continue;
                 curobj = dictGetKey(de);
                 curscore = *(double*)dictGetVal(de);
 
@@ -1280,22 +1334,30 @@ void zaddGenericCommand(redisClient *c, int incr) {
                     server.dirty++;
                     updated++;
                 }
-            } else {
+                processed++;
+            } else if (!xx) {
                 znode = zslInsert(zs->zsl,score,ele);
                 incrRefCount(ele); /* Inserted in skiplist. */
                 redisAssertWithInfo(c,NULL,dictAdd(zs->dict,ele,&znode->score) == DICT_OK);
                 incrRefCount(ele); /* Added to dictionary. */
                 server.dirty++;
                 added++;
+                processed++;
             }
         } else {
             redisPanic("Unknown sorted set encoding");
         }
     }
-    if (incr) /* ZINCRBY */
-        addReplyDouble(c,score);
-    else /* ZADD */
-        addReplyLongLong(c,added);
+
+reply_to_client:
+    if (incr) { /* ZINCRBY or INCR option. */
+        if (processed)
+            addReplyDouble(c,score);
+        else
+            addReply(c,shared.nullbulk);
+    } else { /* ZADD. */
+        addReplyLongLong(c,ch ? added+updated : added);
+    }
 
 cleanup:
     zfree(scores);
@@ -1307,11 +1369,11 @@ cleanup:
 }
 
 void zaddCommand(redisClient *c) {
-    zaddGenericCommand(c,0);
+    zaddGenericCommand(c,ZADD_NONE);
 }
 
 void zincrbyCommand(redisClient *c) {
-    zaddGenericCommand(c,1);
+    zaddGenericCommand(c,ZADD_INCR);
 }
 
 void zremCommand(redisClient *c) {

src/version.h

@@ -1 +1 @@
-#define REDIS_VERSION "3.0.0"
+#define REDIS_VERSION "3.0.5"

tests/integration/rdb.tcl

@@ -7,19 +7,19 @@ start_server [list overrides [list "dir" $server_path "dbfilename" "encodings.rd
   test "RDB encoding loading test" {
     r select 0
     csvdump r
-  } {"compressible","string","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
-"hash","hash","a","1","aa","10","aaa","100","b","2","bb","20","bbb","200","c","3","cc","30","ccc","300","ddd","400","eee","5000000000",
-"hash_zipped","hash","a","1","b","2","c","3",
-"list","list","1","2","3","a","b","c","100000","6000000000","1","2","3","a","b","c","100000","6000000000","1","2","3","a","b","c","100000","6000000000",
-"list_zipped","list","1","2","3","a","b","c","100000","6000000000",
-"number","string","10"
-"set","set","1","100000","2","3","6000000000","a","b","c",
-"set_zipped_1","set","1","2","3","4",
-"set_zipped_2","set","100000","200000","300000","400000",
-"set_zipped_3","set","1000000000","2000000000","3000000000","4000000000","5000000000","6000000000",
-"string","string","Hello World"
-"zset","zset","a","1","b","2","c","3","aa","10","bb","20","cc","30","aaa","100","bbb","200","ccc","300","aaaa","1000","cccc","123456789","bbbb","5000000000",
-"zset_zipped","zset","a","1","b","2","c","3",
+  } {"0","compressible","string","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+"0","hash","hash","a","1","aa","10","aaa","100","b","2","bb","20","bbb","200","c","3","cc","30","ccc","300","ddd","400","eee","5000000000",
+"0","hash_zipped","hash","a","1","b","2","c","3",
+"0","list","list","1","2","3","a","b","c","100000","6000000000","1","2","3","a","b","c","100000","6000000000","1","2","3","a","b","c","100000","6000000000",
+"0","list_zipped","list","1","2","3","a","b","c","100000","6000000000",
+"0","number","string","10"
+"0","set","set","1","100000","2","3","6000000000","a","b","c",
+"0","set_zipped_1","set","1","2","3","4",
+"0","set_zipped_2","set","100000","200000","300000","400000",
+"0","set_zipped_3","set","1000000000","2000000000","3000000000","4000000000","5000000000","6000000000",
+"0","string","string","Hello World"
+"0","zset","zset","a","1","b","2","c","3","aa","10","bb","20","cc","30","aaa","100","bbb","200","ccc","300","aaaa","1000","cccc","123456789","bbbb","5000000000",
+"0","zset_zipped","zset","a","1","b","2","c","3",
 }
 }
 

tests/integration/replication-psync.tcl

@@ -13,7 +13,11 @@ proc stop_bg_complex_data {handle} {
 #
 # You can specifiy backlog size, ttl, delay before reconnection, test duration
 # in seconds, and an additional condition to verify at the end.
-proc test_psync {descr duration backlog_size backlog_ttl delay cond} {
+#
+# If reconnect is > 0, the test actually try to break the connection and
+# reconnect with the master, otherwise just the initial synchronization is
+# checked for consistency.
+proc test_psync {descr duration backlog_size backlog_ttl delay cond diskless reconnect} {
     start_server {tags {"repl"}} {
         start_server {} {
 
@@ -24,6 +28,8 @@ proc test_psync {descr duration backlog_size backlog_ttl delay cond} {
 
             $master config set repl-backlog-size $backlog_size
             $master config set repl-backlog-ttl $backlog_ttl
+            $master config set repl-diskless-sync $diskless
+            $master config set repl-diskless-sync-delay 1
 
             set load_handle0 [start_bg_complex_data $master_host $master_port 9 100000]
             set load_handle1 [start_bg_complex_data $master_host $master_port 11 100000]
@@ -48,22 +54,24 @@ proc test_psync {descr duration backlog_size backlog_ttl delay cond} {
                 }
             }
 
-            test "Test replication partial resync: $descr" {
+            test "Test replication partial resync: $descr (diskless: $diskless, reconnect: $reconnect)" {
                 # Now while the clients are writing data, break the maste-slave
                 # link multiple times.
-                for {set j 0} {$j < $duration*10} {incr j} {
-                    after 100
-                    # catch {puts "MASTER [$master dbsize] keys, SLAVE [$slave dbsize] keys"}
+                if ($reconnect) {
+                    for {set j 0} {$j < $duration*10} {incr j} {
+                        after 100
+                        # catch {puts "MASTER [$master dbsize] keys, SLAVE [$slave dbsize] keys"}
 
-                    if {($j % 20) == 0} {
-                        catch {
-                            if {$delay} {
-                                $slave multi
-                                $slave client kill $master_host:$master_port
-                                $slave debug sleep $delay
-                                $slave exec
-                            } else {
-                                $slave client kill $master_host:$master_port
+                        if {($j % 20) == 0} {
+                            catch {
+                                if {$delay} {
+                                    $slave multi
+                                    $slave client kill $master_host:$master_port
+                                    $slave debug sleep $delay
+                                    $slave exec
+                                } else {
+                                    $slave client kill $master_host:$master_port
+                                }
                             }
                         }
                     }
@@ -98,18 +106,23 @@ proc test_psync {descr duration backlog_size backlog_ttl delay cond} {
     }
 }
 
-test_psync {ok psync} 6 1000000 3600 0 {
-    assert {[s -1 sync_partial_ok] > 0}
-}
+foreach diskless {no yes} {
+    test_psync {no reconnection, just sync} 6 1000000 3600 0 {
+    } $diskless 0
 
-test_psync {no backlog} 6 100 3600 0.5 {
-    assert {[s -1 sync_partial_err] > 0}
-}
+    test_psync {ok psync} 6 1000000 3600 0 {
+        assert {[s -1 sync_partial_ok] > 0}
+    } $diskless 1
 
-test_psync {ok after delay} 3 100000000 3600 3 {
-    assert {[s -1 sync_partial_ok] > 0}
-}
+    test_psync {no backlog} 6 100 3600 0.5 {
+        assert {[s -1 sync_partial_err] > 0}
+    } $diskless 1
 
-test_psync {backlog expired} 3 100000000 1 3 {
-    assert {[s -1 sync_partial_err] > 0}
+    test_psync {ok after delay} 3 100000000 3600 3 {
+        assert {[s -1 sync_partial_ok] > 0}
+    } $diskless 1
+
+    test_psync {backlog expired} 3 100000000 1 3 {
+        assert {[s -1 sync_partial_err] > 0}
+    } $diskless 1
 }

tests/integration/replication.tcl

@@ -1,3 +1,56 @@
+proc log_file_matches {log pattern} {
+    set fp [open $log r]
+    set content [read $fp]
+    close $fp
+    string match $pattern $content
+}
+
+start_server {tags {"repl"}} {
+    set slave [srv 0 client]
+    set slave_host [srv 0 host]
+    set slave_port [srv 0 port]
+    set slave_log [srv 0 stdout]
+    start_server {} {
+        set master [srv 0 client]
+        set master_host [srv 0 host]
+        set master_port [srv 0 port]
+
+        # Configure the master in order to hang waiting for the BGSAVE
+        # operation, so that the slave remains in the handshake state.
+        $master config set repl-diskless-sync yes
+        $master config set repl-diskless-sync-delay 1000
+
+        # Use a short replication timeout on the slave, so that if there
+        # are no bugs the timeout is triggered in a reasonable amount
+        # of time.
+        $slave config set repl-timeout 5
+
+        # Start the replication process...
+        $slave slaveof $master_host $master_port
+
+        test {Slave enters handshake} {
+            wait_for_condition 50 1000 {
+                [string match *handshake* [$slave role]]
+            } else {
+                fail "Slave does not enter handshake state"
+            }
+        }
+
+        # But make the master unable to send
+        # the periodic newlines to refresh the connection. The slave
+        # should detect the timeout.
+        $master debug sleep 10
+
+        test {Slave is able to detect timeout during handshake} {
+            wait_for_condition 50 1000 {
+                [log_file_matches $slave_log "*Timeout connecting to the MASTER*"]
+            } else {
+                fail "Slave is not able to detect timeout"
+            }
+        }
+    }
+}
+
 start_server {tags {"repl"}} {
     set A [srv 0 client]
     set A_host [srv 0 host]

tests/sentinel/tests/06-ckquorum.tcl

@@ -0,0 +1,34 @@
+# Test for the SENTINEL CKQUORUM command
+
+source "../tests/includes/init-tests.tcl"
+set num_sentinels [llength $::sentinel_instances]
+
+test "CKQUORUM reports OK and the right amount of Sentinels" {
+    foreach_sentinel_id id {
+        assert_match "*OK $num_sentinels usable*" [S $id SENTINEL CKQUORUM mymaster]
+    }
+}
+
+test "CKQUORUM detects quorum cannot be reached" {
+    set orig_quorum [expr {$num_sentinels/2+1}]
+    S 0 SENTINEL SET mymaster quorum [expr {$num_sentinels+1}]
+    catch {[S 0 SENTINEL CKQUORUM mymaster]} err
+    assert_match "*NOQUORUM*" $err
+    S 0 SENTINEL SET mymaster quorum $orig_quorum
+}
+
+test "CKQUORUM detects failover authorization cannot be reached" {
+    set orig_quorum [expr {$num_sentinels/2+1}]
+    S 0 SENTINEL SET mymaster quorum 1
+    kill_instance sentinel 1
+    kill_instance sentinel 2
+    kill_instance sentinel 3
+    after 5000
+    catch {[S 0 SENTINEL CKQUORUM mymaster]} err
+    assert_match "*NOQUORUM*" $err
+    S 0 SENTINEL SET mymaster quorum $orig_quorum
+    restart_instance sentinel 1
+    restart_instance sentinel 2
+    restart_instance sentinel 3
+}
+

tests/support/util.tcl

@@ -262,46 +262,50 @@ proc formatCommand {args} {
 
 proc csvdump r {
     set o {}
-    foreach k [lsort [{*}$r keys *]] {
-        set type [{*}$r type $k]
-        append o [csvstring $k] , [csvstring $type] ,
-        switch $type {
-            string {
-                append o [csvstring [{*}$r get $k]] "\n"
-            }
-            list {
-                foreach e [{*}$r lrange $k 0 -1] {
-                    append o [csvstring $e] ,
+    for {set db 0} {$db < 16} {incr db} {
+        {*}$r select $db
+        foreach k [lsort [{*}$r keys *]] {
+            set type [{*}$r type $k]
+            append o [csvstring $db] , [csvstring $k] , [csvstring $type] ,
+            switch $type {
+                string {
+                    append o [csvstring [{*}$r get $k]] "\n"
                 }
-                append o "\n"
-            }
-            set {
-                foreach e [lsort [{*}$r smembers $k]] {
-                    append o [csvstring $e] ,
+                list {
+                    foreach e [{*}$r lrange $k 0 -1] {
+                        append o [csvstring $e] ,
+                    }
+                    append o "\n"
                 }
-                append o "\n"
-            }
-            zset {
-                foreach e [{*}$r zrange $k 0 -1 withscores] {
-                    append o [csvstring $e] ,
+                set {
+                    foreach e [lsort [{*}$r smembers $k]] {
+                        append o [csvstring $e] ,
+                    }
+                    append o "\n"
                 }
-                append o "\n"
-            }
-            hash {
-                set fields [{*}$r hgetall $k]
-                set newfields {}
-                foreach {k v} $fields {
-                    lappend newfields [list $k $v]
+                zset {
+                    foreach e [{*}$r zrange $k 0 -1 withscores] {
+                        append o [csvstring $e] ,
+                    }
+                    append o "\n"
                 }
-                set fields [lsort -index 0 $newfields]
-                foreach kv $fields {
-                    append o [csvstring [lindex $kv 0]] ,
-                    append o [csvstring [lindex $kv 1]] ,
+                hash {
+                    set fields [{*}$r hgetall $k]
+                    set newfields {}
+                    foreach {k v} $fields {
+                        lappend newfields [list $k $v]
+                    }
+                    set fields [lsort -index 0 $newfields]
+                    foreach kv $fields {
+                        append o [csvstring [lindex $kv 0]] ,
+                        append o [csvstring [lindex $kv 1]] ,
+                    }
+                    append o "\n"
                 }
-                append o "\n"
             }
         }
     }
+    {*}$r select 9
     return $o
 }
 

tests/unit/basic.tcl

@@ -433,6 +433,32 @@ start_server {tags {"basic"}} {
         set e
     } {*ERR*index out of range}
 
+    test {MOVE can move key expire metadata as well} {
+        r select 10
+        r flushdb
+        r select 9
+        r set mykey foo ex 100
+        r move mykey 10
+        assert {[r ttl mykey] == -2}
+        r select 10
+        assert {[r ttl mykey] > 0 && [r ttl mykey] <= 100}
+        assert {[r get mykey] eq "foo"}
+        r select 9
+    }
+
+    test {MOVE does not create an expire if it does not exist} {
+        r select 10
+        r flushdb
+        r select 9
+        r set mykey foo
+        r move mykey 10
+        assert {[r ttl mykey] == -2}
+        r select 10
+        assert {[r ttl mykey] == -1}
+        assert {[r get mykey] eq "foo"}
+        r select 9
+    }
+
     test {SET/GET keys in different DBs} {
         r set a hello
         r set b world

tests/unit/type/hash.tcl

@@ -2,8 +2,8 @@ start_server {tags {"hash"}} {
     test {HSET/HLEN - Small hash creation} {
         array set smallhash {}
         for {set i 0} {$i < 8} {incr i} {
-            set key [randstring 0 8 alpha]
-            set val [randstring 0 8 alpha]
+            set key __avoid_collisions__[randstring 0 8 alpha]
+            set val __avoid_collisions__[randstring 0 8 alpha]
             if {[info exists smallhash($key)]} {
                 incr i -1
                 continue
@@ -21,8 +21,8 @@ start_server {tags {"hash"}} {
     test {HSET/HLEN - Big hash creation} {
         array set bighash {}
         for {set i 0} {$i < 1024} {incr i} {
-            set key [randstring 0 8 alpha]
-            set val [randstring 0 8 alpha]
+            set key __avoid_collisions__[randstring 0 8 alpha]
+            set val __avoid_collisions__[randstring 0 8 alpha]
             if {[info exists bighash($key)]} {
                 incr i -1
                 continue
@@ -33,7 +33,7 @@ start_server {tags {"hash"}} {
         list [r hlen bighash]
     } {1024}
 
-    test {Is the big hash encoded with a ziplist?} {
+    test {Is the big hash encoded with an hash table?} {
         assert_encoding hashtable bighash
     }
 

tests/unit/type/set.tcl

@@ -450,6 +450,7 @@ start_server {
     test "SMOVE non existing key" {
         setup_move
         assert_equal 0 [r smove myset1 myset2 foo]
+        assert_equal 0 [r smove myset1 myset1 foo]
         assert_equal {1 a b} [lsort [r smembers myset1]]
         assert_equal {2 3 4} [lsort [r smembers myset2]]
     }

tests/unit/type/zset.tcl

@@ -43,6 +43,84 @@ start_server {tags {"zset"}} {
             assert_error "*not*float*" {r zadd myzset nan abc}
         }
 
+        test "ZADD with options syntax error with incomplete pair" {
+            r del ztmp
+            catch {r zadd ztmp xx 10 x 20} err
+            set err
+        } {ERR*}
+
+        test "ZADD XX option without key - $encoding" {
+            r del ztmp
+            assert {[r zadd ztmp xx 10 x] == 0}
+            assert {[r type ztmp] eq {none}}
+        }
+
+        test "ZADD XX existing key - $encoding" {
+            r del ztmp
+            r zadd ztmp 10 x
+            assert {[r zadd ztmp xx 20 y] == 0}
+            assert {[r zcard ztmp] == 1}
+        }
+
+        test "ZADD XX returns the number of elements actually added" {
+            r del ztmp
+            r zadd ztmp 10 x
+            set retval [r zadd ztmp 10 x 20 y 30 z]
+            assert {$retval == 2}
+        }
+
+        test "ZADD XX updates existing elements score" {
+            r del ztmp
+            r zadd ztmp 10 x 20 y 30 z
+            r zadd ztmp xx 5 foo 11 x 21 y 40 zap
+            assert {[r zcard ztmp] == 3}
+            assert {[r zscore ztmp x] == 11}
+            assert {[r zscore ztmp y] == 21}
+        }
+
+        test "ZADD XX and NX are not compatible" {
+            r del ztmp
+            catch {r zadd ztmp xx nx 10 x} err
+            set err
+        } {ERR*}
+
+        test "ZADD NX with non exisitng key" {
+            r del ztmp
+            r zadd ztmp nx 10 x 20 y 30 z
+            assert {[r zcard ztmp] == 3}
+        }
+
+        test "ZADD NX only add new elements without updating old ones" {
+            r del ztmp
+            r zadd ztmp 10 x 20 y 30 z
+            assert {[r zadd ztmp nx 11 x 21 y 100 a 200 b] == 2}
+            assert {[r zscore ztmp x] == 10}
+            assert {[r zscore ztmp y] == 20}
+            assert {[r zscore ztmp a] == 100}
+            assert {[r zscore ztmp b] == 200}
+        }
+
+        test "ZADD INCR works like ZINCRBY" {
+            r del ztmp
+            r zadd ztmp 10 x 20 y 30 z
+            r zadd ztmp INCR 15 x
+            assert {[r zscore ztmp x] == 25}
+        }
+
+        test "ZADD INCR works with a single score-elemenet pair" {
+            r del ztmp
+            r zadd ztmp 10 x 20 y 30 z
+            catch {r zadd ztmp INCR 15 x 10 y} err
+            set err
+        } {ERR*}
+
+        test "ZADD CH option changes return value to all changed elements" {
+            r del ztmp
+            r zadd ztmp 10 x 20 y 30 z
+            assert {[r zadd ztmp 11 x 21 y 30 z] == 0}
+            assert {[r zadd ztmp ch 12 x 22 y 30 z] == 2}
+        }
+
         test "ZINCRBY calls leading to NaN result in error" {
             r zincrby myzset +inf abc
             assert_error "*NaN*" {r zincrby myzset -inf abc}
@@ -77,6 +155,8 @@ start_server {tags {"zset"}} {
         }
 
         test "ZCARD basics - $encoding" {
+            r del ztmp
+            r zadd ztmp 10 a 20 b 30 c
             assert_equal 3 [r zcard ztmp]
             assert_equal 0 [r zcard zdoesntexist]
         }

utils/generate-command-help.rb

@@ -12,7 +12,9 @@ GROUPS = [
   "connection",
   "server",
   "scripting",
-  "hyperloglog"
+  "hyperloglog",
+  "cluster",
+  "geo"
 ].freeze
 
 GROUPS_BY_NAME = Hash[*