RSS aware maxmemory: algorith #1 implemented.

RDB EOF detection was relying on the final part of the RDB transfer to
be a magic 40 bytes EOF marker. However as the slave is put online
immediately, and because of sockets timeouts, the replication stream is
actually contiguous with the RDB file.

This means that to detect the EOF correctly we should either:

1) Scan all the stream searching for the mark. Sucks CPU-wise.
2) Start to send the replication stream only after an acknowledge.
3) Implement a proper chunked encoding.

For now solution "2" was picked, so the master does not start to send
ASAP the stream of commands in the case of diskless replication. We wait
for the first REPLCONF ACK command from the slave, that certifies us
that the slave correctly loaded the RDB file and is ready to get more
data.

src/cluster.c

@@ -178,7 +178,7 @@ int clusterLoadConfig(char *filename) {
             clusterAddNode(n);
         }
         /* Address and port */
-        if ((p = strchr(argv[1],':')) == NULL) goto fmterr;
+        if ((p = strrchr(argv[1],':')) == NULL) goto fmterr;
         *p = '\0';
         memcpy(n->ip,argv[1],strlen(argv[1])+1);
         n->port = atoi(p+1);
@@ -1533,7 +1533,7 @@ int clusterProcessPacket(clusterLink *link) {
         }
     }
 
-    /* Process packets by type. */
+    /* Initial processing of PING and MEET requests replying with a PONG. */
     if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_MEET) {
         redisLog(REDIS_DEBUG,"Ping packet received: %p", (void*)link->node);
 
@@ -1571,14 +1571,17 @@ int clusterProcessPacket(clusterLink *link) {
             clusterDoBeforeSleep(CLUSTER_TODO_SAVE_CONFIG);
         }
 
-        /* Get info from the gossip section */
-        clusterProcessGossipSection(hdr,link);
+        /* If this is a MEET packet from an unknown node, we still process
+         * the gossip section here since we have to trust the sender because
+         * of the message type. */
+        if (!sender && type == CLUSTERMSG_TYPE_MEET)
+            clusterProcessGossipSection(hdr,link);
 
         /* Anyway reply with a PONG */
         clusterSendPing(link,CLUSTERMSG_TYPE_PONG);
     }
 
-    /* PING or PONG: process config information. */
+    /* PING, PONG, MEET: process config information. */
     if (type == CLUSTERMSG_TYPE_PING || type == CLUSTERMSG_TYPE_PONG ||
         type == CLUSTERMSG_TYPE_MEET)
     {
@@ -1775,7 +1778,7 @@ int clusterProcessPacket(clusterLink *link) {
         }
 
         /* Get info from the gossip section */
-        clusterProcessGossipSection(hdr,link);
+        if (sender) clusterProcessGossipSection(hdr,link);
     } else if (type == CLUSTERMSG_TYPE_FAIL) {
         clusterNode *failing;
 
@@ -3235,7 +3238,7 @@ void clusterCloseAllSlots(void) {
 
 void clusterUpdateState(void) {
     int j, new_state;
-    int unreachable_masters = 0;
+    int reachable_masters = 0;
     static mstime_t among_minority_time;
     static mstime_t first_call_time = 0;
 
@@ -3271,8 +3274,8 @@ void clusterUpdateState(void) {
     /* Compute the cluster size, that is the number of master nodes
      * serving at least a single slot.
      *
-     * At the same time count the number of unreachable masters with
-     * at least one node. */
+     * At the same time count the number of reachable masters having
+     * at least one slot. */
     {
         dictIterator *di;
         dictEntry *de;
@@ -3284,20 +3287,19 @@ void clusterUpdateState(void) {
 
             if (nodeIsMaster(node) && node->numslots) {
                 server.cluster->size++;
-                if (node->flags & (REDIS_NODE_FAIL|REDIS_NODE_PFAIL))
-                    unreachable_masters++;
+                if ((node->flags & (REDIS_NODE_FAIL|REDIS_NODE_PFAIL)) == 0)
+                    reachable_masters++;
             }
         }
         dictReleaseIterator(di);
     }
 
-    /* If we can't reach at least half the masters, change the cluster state
-     * to FAIL, as we are not even able to mark nodes as FAIL in this side
-     * of the netsplit because of lack of majority. */
+    /* If we are in a minority partition, change the cluster state
+     * to FAIL. */
     {
         int needed_quorum = (server.cluster->size / 2) + 1;
 
-        if (unreachable_masters >= needed_quorum) {
+        if (reachable_masters < needed_quorum) {
             new_state = REDIS_CLUSTER_FAIL;
             among_minority_time = mstime();
         }
@@ -4305,8 +4307,8 @@ int migrateGetSocket(redisClient *c, robj *host, robj *port, long timeout) {
     }
 
     /* Create the socket */
-    fd = anetTcpNonBlockConnect(server.neterr,c->argv[1]->ptr,
-                atoi(c->argv[2]->ptr));
+    fd = anetTcpNonBlockBindConnect(server.neterr,c->argv[1]->ptr,
+                atoi(c->argv[2]->ptr),REDIS_BIND_ADDR);
     if (fd == -1) {
         sdsfree(name);
         addReplyErrorFormat(c,"Can't connect to target node: %s",

src/config.c

@@ -226,6 +226,11 @@ void loadServerConfigFromString(char *config) {
             }
         } else if (!strcasecmp(argv[0],"maxmemory") && argc == 2) {
             server.maxmemory = memtoll(argv[1],NULL);
+            server.maxmemory_enforced = (double) server.maxmemory / server.maxmemory_frag_guess;
+        } else if (!strcasecmp(argv[0],"rss-aware-maxmemory") && argc==2) {
+            if ((server.rss_aware_maxmemory = yesnotoi(argv[1])) == -1) {
+                err = "argument must be 'yes' or 'no'"; goto loaderr;
+            }
         } else if (!strcasecmp(argv[0],"maxmemory-policy") && argc == 2) {
             if (!strcasecmp(argv[1],"volatile-lru")) {
                 server.maxmemory_policy = REDIS_MAXMEMORY_VOLATILE_LRU;
@@ -631,12 +636,18 @@ void configSetCommand(redisClient *c) {
         if (getLongLongFromObject(o,&ll) == REDIS_ERR ||
             ll < 0) goto badfmt;
         server.maxmemory = ll;
+        server.maxmemory_enforced = (double) server.maxmemory / server.maxmemory_frag_guess;
         if (server.maxmemory) {
             if (server.maxmemory < zmalloc_used_memory()) {
                 redisLog(REDIS_WARNING,"WARNING: the new maxmemory value set via CONFIG SET is smaller than the current memory usage. This will result in keys eviction and/or inability to accept new write commands depending on the maxmemory-policy.");
             }
             freeMemoryIfNeeded();
         }
+    } else if (!strcasecmp(c->argv[2]->ptr,"rss-aware-maxmemory")) {
+        int yn = yesnotoi(o->ptr);
+
+        if (yn == -1) goto badfmt;
+        server.rss_aware_maxmemory = yn;
     } else if (!strcasecmp(c->argv[2]->ptr,"maxclients")) {
         int orig_value = server.maxclients;
 
@@ -1093,6 +1104,8 @@ void configGetCommand(redisClient *c) {
             server.aof_rewrite_incremental_fsync);
     config_get_bool_field("aof-load-truncated",
             server.aof_load_truncated);
+    config_get_bool_field("rss-aware-maxmemory",
+            server.rss_aware_maxmemory);
 
     /* Everything we can't handle with macros follows. */
 

src/config.h

@@ -48,11 +48,13 @@
 #define HAVE_PROC_STAT 1
 #define HAVE_PROC_MAPS 1
 #define HAVE_PROC_SMAPS 1
+#define HAVE_RSS_REPORTING 1
 #endif
 
 /* Test for task_info() */
 #if defined(__APPLE__)
 #define HAVE_TASKINFO 1
+#define HAVE_RSS_REPORTING 1
 #endif
 
 /* Test for backtrace() */

src/debug.c

@@ -336,7 +336,7 @@ void debugCommand(redisClient *c) {
         dictExpand(c->db->dict,keys);
         for (j = 0; j < keys; j++) {
             snprintf(buf,sizeof(buf),"%s:%lu",
-                (c->argc == 3) ? "key" : c->argv[3]->ptr, j);
+                (c->argc == 3) ? "key" : (char*)c->argv[3]->ptr, j);
             key = createStringObject(buf,strlen(buf));
             if (lookupKeyRead(c->db,key) != NULL) {
                 decrRefCount(key);

src/latency.c

@@ -56,6 +56,32 @@ dictType latencyTimeSeriesDictType = {
     dictVanillaFree             /* val destructor */
 };
 
+/* ------------------------- Utility functions ------------------------------ */
+
+#ifdef __linux__
+/* Returns 1 if Transparent Huge Pages support is enabled in the kernel.
+ * Otherwise (or if we are unable to check) 0 is returned. */
+int THPIsEnabled(void) {
+    char buf[1024];
+
+    FILE *fp = fopen("/sys/kernel/mm/transparent_hugepage/enabled","r");
+    if (!fp) return 0;
+    if (fgets(buf,sizeof(buf),fp) == NULL) {
+        fclose(fp);
+        return 0;
+    }
+    fclose(fp);
+    return (strstr(buf,"[never]") == NULL) ? 1 : 0;
+}
+
+/* Report the amount of AnonHugePages in smap, in bytes. If the return
+ * value of the function is non-zero, the process is being targeted by
+ * THP support, and is likely to have memory usage / latency issues. */
+int THPGetAnonHugePagesSize(void) {
+    return zmalloc_get_smap_bytes_by_field("AnonHugePages:");
+}
+#endif
+
 /* ---------------------------- Latency API --------------------------------- */
 
 /* Latency monitor initialization. We just need to create the dictionary
@@ -203,6 +229,7 @@ sds createLatencyReport(void) {
     int advise_hz = 0;              /* Use higher HZ. */
     int advise_large_objects = 0;   /* Deletion of large objects. */
     int advise_relax_fsync_policy = 0; /* appendfsync always is slow. */
+    int advise_disable_thp = 0;     /* AnonHugePages detected. */
     int advices = 0;
 
     /* Return ASAP if the latency engine is disabled and it looks like it
@@ -346,9 +373,15 @@ sds createLatencyReport(void) {
     }
     dictReleaseIterator(di);
 
-    if (eventnum == 0) {
+    /* Add non event based advices. */
+    if (THPGetAnonHugePagesSize() > 0) {
+        advise_disable_thp = 1;
+        advices++;
+    }
+
+    if (eventnum == 0 && advices == 0) {
         report = sdscat(report,"Dave, no latency spike was observed during the lifetime of this Redis instance, not in the slightest bit. I honestly think you ought to sit down calmly, take a stress pill, and think things over.\n");
-    } else if (advices == 0) {
+    } else if (eventnum > 0 && advices == 0) {
         report = sdscat(report,"\nWhile there are latency events logged, I'm not able to suggest any easy fix. Please use the Redis community to get some help, providing this report in your help request.\n");
     } else {
         /* Add all the suggestions accumulated so far. */
@@ -418,6 +451,10 @@ sds createLatencyReport(void) {
         if (advise_large_objects) {
             report = sdscat(report,"- Deleting, expiring or evicting (because of maxmemory policy) large objects is a blocking operation. If you have very large objects that are often deleted, expired, or evicted, try to fragment those objects into multiple smaller objects.\n");
         }
+
+        if (advise_disable_thp) {
+            report = sdscat(report,"- I detected a non zero amount of anonymous huge pages used by your process. This creates very serious latency events in different conditions, especially when Redis is persisting on disk. To disable THP support use the command 'echo never > /sys/kernel/mm/transparent_hugepage/enabled', make sure to also add it into /etc/rc.local so that the command will be executed again after a reboot. Note that even if you have already disabled THP, you still need to restart the Redis process to get rid of the huge pages already created.\n");
+        }
     }
 
     return report;

src/latency.h

@@ -63,6 +63,7 @@ struct latencyStats {
 
 void latencyMonitorInit(void);
 void latencyAddSample(char *event, mstime_t latency);
+int THPIsEnabled(void);
 
 /* Latency monitoring macros. */
 

src/networking.c

@@ -100,6 +100,7 @@ redisClient *createClient(int fd) {
     c->ctime = c->lastinteraction = server.unixtime;
     c->authenticated = 0;
     c->replstate = REDIS_REPL_NONE;
+    c->repl_put_online_on_ack = 0;
     c->reploff = 0;
     c->repl_ack_off = 0;
     c->repl_ack_time = 0;

src/redis-cli.c

@@ -524,7 +524,8 @@ static int cliReadReply(int output_raw_strings) {
         }
         if (config.interactive) {
             /* Filter cases where we should reconnect */
-            if (context->err == REDIS_ERR_IO && errno == ECONNRESET)
+            if (context->err == REDIS_ERR_IO &&
+                (errno == ECONNRESET || errno == EPIPE))
                 return REDIS_ERR;
             if (context->err == REDIS_ERR_EOF)
                 return REDIS_ERR;
@@ -1511,7 +1512,7 @@ static void findBigKeys(void) {
     for(i=0;i<TYPE_NONE; i++) {
         maxkeys[i] = sdsempty();
         if(!maxkeys[i]) {
-            fprintf(stderr, "Failed to allocate memory for largest key names!");
+            fprintf(stderr, "Failed to allocate memory for largest key names!\n");
             exit(1);
         }
     }
@@ -1914,6 +1915,8 @@ int main(int argc, char **argv) {
     argc -= firstarg;
     argv += firstarg;
 
+    signal(SIGPIPE, SIG_IGN);
+
     /* Latency mode */
     if (config.latency_mode) {
         if (cliConnect(0) == REDIS_ERR) exit(1);

src/redis-trib.rb

@@ -50,14 +50,16 @@ end
 class ClusterNode
     def initialize(addr)
         s = addr.split(":")
-        if s.length != 2
-            puts "Invalid node name #{addr}"
-            exit 1
+        if s.length < 2
+           puts "Invalid IP or Port (given as #{addr}) - use IP:Port format"
+           exit 1
         end
+        port = s.pop # removes port from split array
+        ip = s.join(":") # if s.length > 1 here, it's IPv6, so restore address
         @r = nil
         @info = {}
-        @info[:host] = s[0]
-        @info[:port] = s[1]
+        @info[:host] = ip
+        @info[:port] = port
         @info[:slots] = {}
         @info[:migrating] = {}
         @info[:importing] = {}

src/redis.c

@@ -31,6 +31,7 @@
 #include "cluster.h"
 #include "slowlog.h"
 #include "bio.h"
+#include "latency.h"
 
 #include <time.h>
 #include <signal.h>
@@ -1432,8 +1433,11 @@ void initServerConfig(void) {
     server.maxclients = REDIS_MAX_CLIENTS;
     server.bpop_blocked_clients = 0;
     server.maxmemory = REDIS_DEFAULT_MAXMEMORY;
+    server.maxmemory_frag_guess = REDIS_DEFAULT_MAXMEMORY_FRAG_GUESS;
+    server.maxmemory_enforced = (double) REDIS_DEFAULT_MAXMEMORY / server.maxmemory_frag_guess;
     server.maxmemory_policy = REDIS_DEFAULT_MAXMEMORY_POLICY;
     server.maxmemory_samples = REDIS_DEFAULT_MAXMEMORY_SAMPLES;
+    server.rss_aware_maxmemory = REDIS_DEFAULT_RSS_AWARE_MAXMEMORY;
     server.hash_max_ziplist_entries = REDIS_HASH_MAX_ZIPLIST_ENTRIES;
     server.hash_max_ziplist_value = REDIS_HASH_MAX_ZIPLIST_VALUE;
     server.list_max_ziplist_entries = REDIS_LIST_MAX_ZIPLIST_ENTRIES;
@@ -3153,7 +3157,7 @@ void evictionPoolPopulate(dict *sampledict, dict *keydict, struct evictionPoolEn
 }
 
 int freeMemoryIfNeeded(void) {
-    size_t mem_used, mem_tofree, mem_freed;
+    size_t mem_used, mem_tofree, mem_freed, mem_target = server.maxmemory;
     int slaves = listLength(server.slaves);
     mstime_t latency;
 
@@ -3179,14 +3183,72 @@ int freeMemoryIfNeeded(void) {
         mem_used -= aofRewriteBufferSize();
     }
 
+    /* If we use RSS aware maxmemory, update the target memory using
+     * the current fragmentation figure. */
+#ifdef HAVE_RSS_REPORTING
+    if (server.rss_aware_maxmemory &&
+        server.maxmemory_policy != REDIS_MAXMEMORY_NO_EVICTION)
+    {
+        static unsigned long iterations = 0;
+        static unsigned long sampling_stage = 1;
+        static float last_observed_frag = 0;
+
+        /* For some time, we analyze what happens during memory pressure, when
+         * objects are evicted and reallocated. */
+        if (mem_used > server.maxmemory_enforced) {
+            unsigned long sample_cycles = 1000000;
+
+            /* Every sample_cycle cycles we sample the fragmentation, and
+             * compare it with the previos one. If it is no longer raising,
+             * we take it as a guess of the fragmentation with this workload. */
+            if (sampling_stage && iterations < sample_cycles) {
+                iterations++;
+                if (iterations == sample_cycles) {
+                    float current_frag = zmalloc_get_fragmentation_ratio(server.resident_set_size);
+                    if (last_observed_frag == 0) {
+                        /* First sample we get. */
+                        last_observed_frag = current_frag;
+                    } else {
+                        if (current_frag <= last_observed_frag) {
+                            size_t enforced_new;
+
+                            sampling_stage = 0;
+                            server.maxmemory_frag_guess = current_frag;
+                            /* Update the global fragmentation guess and use
+                             * it (also used it for successive
+                             * "CONFIG SET maxmemory" commands). */
+                            if (server.maxmemory_frag_guess < 1)
+                                server.maxmemory_frag_guess = 1;
+                            else if (server.maxmemory_frag_guess > 2)
+                                server.maxmemory_frag_guess = 2;
+
+                            /* Only set the new limit if it is higher than our
+                             * initial guess, otherwise it is futile: RSS will
+                             * not go backward anyway. */
+                            enforced_new = (double) server.maxmemory /
+                                           server.maxmemory_frag_guess;
+                            if (enforced_new > server.maxmemory_enforced)
+                                server.maxmemory_enforced = enforced_new;
+                            redisLog(REDIS_NOTICE,"RSS aware maxmemory, fragmentation looks stable at: %f", server.maxmemory_frag_guess);
+                        }
+                        last_observed_frag = current_frag;
+                    }
+                    iterations = 0;
+                }
+            }
+        }
+        mem_target = server.maxmemory_enforced;
+    }
+#endif
+
     /* Check if we are over the memory limit. */
-    if (mem_used <= server.maxmemory) return REDIS_OK;
+    if (mem_used <= mem_target) return REDIS_OK;
 
     if (server.maxmemory_policy == REDIS_MAXMEMORY_NO_EVICTION)
         return REDIS_ERR; /* We need to free memory, but policy forbids. */
 
     /* Compute how much memory we need to free. */
-    mem_tofree = mem_used - server.maxmemory;
+    mem_tofree = mem_used - mem_target;
     mem_freed = 0;
     latencyStartMonitor(latency);
     while (mem_freed < mem_tofree) {
@@ -3330,10 +3392,13 @@ int linuxOvercommitMemoryValue(void) {
     return atoi(buf);
 }
 
-void linuxOvercommitMemoryWarning(void) {
+void linuxMemoryWarnings(void) {
     if (linuxOvercommitMemoryValue() == 0) {
         redisLog(REDIS_WARNING,"WARNING overcommit_memory is set to 0! Background save may fail under low memory condition. To fix this issue add 'vm.overcommit_memory = 1' to /etc/sysctl.conf and then reboot or run the command 'sysctl vm.overcommit_memory=1' for this to take effect.");
     }
+    if (THPIsEnabled()) {
+        redisLog(REDIS_WARNING,"WARNING you have Transparent Huge Pages (THP) support enabled in your kernel. This will create latency and memory usage issues with Redis. To fix this issue run the command 'echo never > /sys/kernel/mm/transparent_hugepage/enabled' as root, and add it to your /etc/rc.local in order to retain the setting after a reboot. Redis must be restarted after THP is disabled.");
+    }
 }
 #endif /* __linux__ */
 
@@ -3606,7 +3671,7 @@ int main(int argc, char **argv) {
         /* Things not needed when running in Sentinel mode. */
         redisLog(REDIS_WARNING,"Server started, Redis version " REDIS_VERSION);
     #ifdef __linux__
-        linuxOvercommitMemoryWarning();
+        linuxMemoryWarnings();
     #endif
         loadDataFromDisk();
         if (server.cluster_enabled) {

src/redis.h

@@ -121,6 +121,8 @@ typedef long long mstime_t; /* millisecond time type. */
 #define REDIS_DEFAULT_REPL_DISABLE_TCP_NODELAY 0
 #define REDIS_DEFAULT_MAXMEMORY 0
 #define REDIS_DEFAULT_MAXMEMORY_SAMPLES 5
+#define REDIS_DEFAULT_MAXMEMORY_FRAG_GUESS 1.4
+#define REDIS_DEFAULT_RSS_AWARE_MAXMEMORY 0
 #define REDIS_DEFAULT_AOF_FILENAME "appendonly.aof"
 #define REDIS_DEFAULT_AOF_NO_FSYNC_ON_REWRITE 0
 #define REDIS_DEFAULT_AOF_LOAD_TRUNCATED 1
@@ -532,6 +534,7 @@ typedef struct redisClient {
     int flags;              /* REDIS_SLAVE | REDIS_MONITOR | REDIS_MULTI ... */
     int authenticated;      /* when requirepass is non-NULL */
     int replstate;          /* replication state if this is a slave */
+    int repl_put_online_on_ack; /* Install slave write handler on ACK. */
     int repldbfd;           /* replication DB file descriptor */
     off_t repldboff;        /* replication DB file offset */
     off_t repldbsize;       /* replication DB file size */
@@ -838,6 +841,10 @@ struct redisServer {
     unsigned long long maxmemory;   /* Max number of memory bytes to use */
     int maxmemory_policy;           /* Policy for key eviction */
     int maxmemory_samples;          /* Pricision of random sampling */
+    /* RSS aware maxmemory additional state. */
+    int rss_aware_maxmemory;                /* Non zero if enabled. */
+    unsigned long long maxmemory_enforced;  /* Currently enforced maxmemory. */
+    float maxmemory_frag_guess;             /* Guessed fragmentation. */
     /* Blocked clients */
     unsigned int bpop_blocked_clients; /* Number of clients blocked by lists */
     list *unblocked_clients; /* list of clients to unblock before next loop */

src/replication.c

@@ -40,6 +40,7 @@
 void replicationDiscardCachedMaster(void);
 void replicationResurrectCachedMaster(int newfd);
 void replicationSendAck(void);
+void putSlaveOnline(redisClient *slave);
 
 /* --------------------------- Utility functions ---------------------------- */
 
@@ -398,6 +399,7 @@ int masterTryPartialResynchronization(redisClient *c) {
     c->flags |= REDIS_SLAVE;
     c->replstate = REDIS_REPL_ONLINE;
     c->repl_ack_time = server.unixtime;
+    c->repl_put_online_on_ack = 0;
     listAddNodeTail(server.slaves,c);
     /* We can't use the connection buffers since they are used to accumulate
      * new commands at this stage. But we are sure the socket send buffer is
@@ -623,6 +625,11 @@ void replconfCommand(redisClient *c) {
             if (offset > c->repl_ack_off)
                 c->repl_ack_off = offset;
             c->repl_ack_time = server.unixtime;
+            /* If this was a diskless replication, we need to really put
+             * the slave online when the first ACK is received (which
+             * confirms slave is online and ready to get more data). */
+            if (c->repl_put_online_on_ack && c->replstate == REDIS_REPL_ONLINE)
+                putSlaveOnline(c);
             /* Note: this command does not reply anything! */
             return;
         } else if (!strcasecmp(c->argv[j]->ptr,"getack")) {
@@ -652,6 +659,7 @@ void replconfCommand(redisClient *c) {
  * 3) Update the count of good slaves. */
 void putSlaveOnline(redisClient *slave) {
     slave->replstate = REDIS_REPL_ONLINE;
+    slave->repl_put_online_on_ack = 0;
     slave->repl_ack_time = server.unixtime;
     if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
         sendReplyToClient, slave) == AE_ERR) {
@@ -660,6 +668,8 @@ void putSlaveOnline(redisClient *slave) {
         return;
     }
     refreshGoodSlavesCount();
+    redisLog(REDIS_NOTICE,"Synchronization with slave %s succeeded",
+        replicationGetSlaveName(slave));
 }
 
 void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
@@ -713,7 +723,6 @@ void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
         slave->repldbfd = -1;
         aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
         putSlaveOnline(slave);
-        redisLog(REDIS_NOTICE,"Synchronization with slave succeeded (disk)");
     }
 }
 
@@ -752,10 +761,16 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
              * diskless replication, our work is trivial, we can just put
              * the slave online. */
             if (type == REDIS_RDB_CHILD_TYPE_SOCKET) {
-                putSlaveOnline(slave);
                 redisLog(REDIS_NOTICE,
-                    "Synchronization with slave %s succeeded (socket)",
+                    "Streamed RDB transfer with slave %s succeeded (socket). Waiting for REPLCONF ACK from slave to enable streaming",
                         replicationGetSlaveName(slave));
+                /* Note: we wait for a REPLCONF ACK message from slave in
+                 * order to really put it online (install the write handler
+                 * so that the accumulated data can be transfered). However
+                 * we change the replication state ASAP, since our slave
+                 * is technically online now. */
+                slave->replstate = REDIS_REPL_ONLINE;
+                slave->repl_put_online_on_ack = 1;
             } else {
                 if (bgsaveerr != REDIS_OK) {
                     freeClient(slave);
@@ -929,7 +944,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
     int eof_reached = 0;
 
     if (usemark) {
-        /* Update the last bytes array, and check if it matches our delimiter. */
+        /* Update the last bytes array, and check if it matches our delimiter.*/
         if (nread >= REDIS_RUN_ID_SIZE) {
             memcpy(lastbytes,buf+nread-REDIS_RUN_ID_SIZE,REDIS_RUN_ID_SIZE);
         } else {
@@ -1357,7 +1372,8 @@ error:
 int connectWithMaster(void) {
     int fd;
 
-    fd = anetTcpNonBlockConnect(NULL,server.masterhost,server.masterport);
+    fd = anetTcpNonBlockBindConnect(NULL,
+        server.masterhost,server.masterport,REDIS_BIND_ADDR);
     if (fd == -1) {
         redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
             strerror(errno));
@@ -2002,8 +2018,9 @@ void replicationCron(void) {
             if (slave->flags & REDIS_PRE_PSYNC) continue;
             if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout)
             {
-                    redisLog(REDIS_WARNING, "Disconnecting timedout slave: %s",
-                        replicationGetSlaveName(slave));
+                redisLog(REDIS_WARNING, "Disconnecting timedout slave: %s",
+                    replicationGetSlaveName(slave));
+                freeClient(slave);
             }
         }
     }

src/zmalloc.c

@@ -328,27 +328,39 @@ float zmalloc_get_fragmentation_ratio(size_t rss) {
     return (float)rss/zmalloc_used_memory();
 }
 
+/* Get the sum of the specified field (converted form kb to bytes) in
+ * /proc/self/smaps. The field must be specified with trailing ":" as it
+ * apperas in the smaps output.
+ *
+ * Example: zmalloc_get_smap_bytes_by_field("Rss:");
+ */
 #if defined(HAVE_PROC_SMAPS)
-size_t zmalloc_get_private_dirty(void) {
+size_t zmalloc_get_smap_bytes_by_field(char *field) {
     char line[1024];
-    size_t pd = 0;
+    size_t bytes = 0;
     FILE *fp = fopen("/proc/self/smaps","r");
+    int flen = strlen(field);
 
     if (!fp) return 0;
     while(fgets(line,sizeof(line),fp) != NULL) {
-        if (strncmp(line,"Private_Dirty:",14) == 0) {
+        if (strncmp(line,field,flen) == 0) {
             char *p = strchr(line,'k');
             if (p) {
                 *p = '\0';
-                pd += strtol(line+14,NULL,10) * 1024;
+                bytes += strtol(line+flen,NULL,10) * 1024;
             }
         }
     }
     fclose(fp);
-    return pd;
+    return bytes;
 }
 #else
-size_t zmalloc_get_private_dirty(void) {
+size_t zmalloc_get_smap_bytes_by_field(char *field) {
+    REDIS_NOTUSED(field);
     return 0;
 }
 #endif
+
+size_t zmalloc_get_private_dirty(void) {
+    return zmalloc_get_smap_bytes_by_field("Private_Dirty:");
+}

src/zmalloc.h

@@ -76,6 +76,7 @@ void zmalloc_set_oom_handler(void (*oom_handler)(size_t));
 float zmalloc_get_fragmentation_ratio(size_t rss);
 size_t zmalloc_get_rss(void);
 size_t zmalloc_get_private_dirty(void);
+size_t zmalloc_get_smap_bytes_by_field(char *field);
 void zlibc_free(void *ptr);
 
 #ifndef HAVE_MALLOC_SIZE

tests/cluster/tests/04-resharding.tcl

@@ -53,13 +53,16 @@ test "Cluster consistency during live resharding" {
             puts -nonewline "...Starting resharding..."
             flush stdout
             set target [dict get [get_myself [randomInt 5]] id]
-            set tribpid [exec \
+            set tribpid [lindex [exec \
                 ../../../src/redis-trib.rb reshard \
                 --from all \
                 --to $target \
                 --slots 100 \
                 --yes \
-                127.0.0.1:[get_instance_attrib redis 0 port] &]
+                127.0.0.1:[get_instance_attrib redis 0 port] \
+                | [info nameofexecutable] \
+                ../tests/helpers/onlydots.tcl \
+                &] 0]
         }
 
         # Write random data to random list.

tests/cluster/tests/helpers/onlydots.tcl

@@ -0,0 +1,16 @@
+# Read the standard input and only shows dots in the output, filtering out
+# all the other characters. Designed to avoid bufferization so that when
+# we get the output of redis-trib and want to show just the dots, we'll see
+# the dots as soon as redis-trib will output them.
+
+fconfigure stdin -buffering none
+
+while 1 {
+    set c [read stdin 1]
+    if {$c eq {}} {
+        exit 0; # EOF
+    } elseif {$c eq {.}} {
+        puts -nonewline .
+        flush stdout
+    }
+}

utils/releasetools/02_upload_tarball.sh

@@ -1,5 +1,6 @@
 #!/bin/bash
 echo "Uploading..."
 scp /tmp/redis-${1}.tar.gz antirez@antirez.com:/var/virtual/download.redis.io/httpdocs/releases/
-echo "Updating web site..."
+echo "Updating web site... (press any key if it is a stable release, or Ctrl+C)"
+read x
 ssh antirez@antirez.com "cd /var/virtual/download.redis.io/httpdocs; ./update.sh ${1}"