Redis 3.2.5.

Compiling Redis worked as a side effect of jemalloc target specifying
-ldl as needed linker options, otherwise it is not provided during
linking and dlopen() API will remain unresolved symbols.

00-RELEASENOTES

@@ -10,6 +10,327 @@ 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.2.5     Released Wed Oct 26 09:16:40 CEST 2016
+================================================================================
+
+Upgrade urgency LOW: This release only fixes a compilation issue due to the
+                     missing -ldl at linking time.
+
+zach shipko in commit 4736407:
+ BSDs don't have -ldl
+ 1 file changed, 15 insertions(+), 5 deletions(-)
+
+antirez in commit 9ada818:
+ Fix modules compilation when libc malloc is used.
+ 1 file changed, 2 insertions(+), 2 deletions(-)
+
+================================================================================
+Redis 3.2.4     Released Mon Sep 26 08:58:21 CEST 2016
+================================================================================
+
+Upgrade urgency CRITICAL: Redis 3.2 and unstable contained a security
+                          vulnerability fixed by this release.
+
+Hello Redis Wizards of the Memory Stores Empire,
+
+this is a Redis critical release in order to fix a security issue
+which is documented clearly here:
+
+    https://github.com/antirez/redis/commit/6d9f8e2462fc2c426d48c941edeb78e5df7d2977
+
+Thanks to Cory Duplantis of Cisco Talos for reporting the issue.
+
+IMPACT:
+
+The gist is that using CONFIG SET calls (or by manipulating redis.conf)
+an attacker is able to compromise certain fields of the "server" global
+structure, including the aof filename pointer, that could be made pointing
+to something else. In turn the AOF name is used in different contexts such
+as logging, rename(2) and open(2) syscalls, leading to potential problems.
+
+Please note that since having access to CONFIG SET also means to be able
+to change the AOF filename (and many other things) directly, this issue
+actual real world impact is quite small, so I would not panik: if you
+have CONFIG SET level of access, you can do more and more easily.
+
+AFFECTED VERSIONS:
+
+All Redis 3.2.x versions are affected.
+
+OTHER CHANGES IN THIS RELEASE:
+
+This release also includes other things:
+
+* TCP binding bug fixed when only certain addresses were available for
+a given port.
+
+* A much better crash report that includes part of the Redis binary:
+this will allow to fix bugs even when we just have a crash log and
+no other help from the original poster oft the issue.
+
+* A fix for Redis Cluster redis-trib displaying of info after creating
+a new cluster.
+
+Please check the following list of commits for credits about who did what.
+Thanks to all the contributors and a special thank to Oran Agra for the
+help in this release.
+
+List of commits:
+
+antirez in commit 0539634:
+ Security: CONFIG SET client-output-buffer-limit overflow fixed.
+ 1 file changed, 5 insertions(+), 3 deletions(-)
+
+antirez in commit c01abcd:
+ fix the fix for the TCP binding.
+ 1 file changed, 15 insertions(+), 10 deletions(-)
+
+oranagra in commit a6d0698:
+ fix tcp binding when IPv6 is unsupported
+ 2 files changed, 14 insertions(+), 10 deletions(-)
+
+antirez in commit 22b6c28:
+ debug.c: no need to define _GNU_SOURCE, is defined in fmacros.h.
+ 1 file changed, 1 deletion(-)
+
+antirez in commit 9e9d398:
+ crash log - improve code dump with more info and called symbols.
+ 1 file changed, 59 insertions(+), 20 deletions(-)
+
+oranagra in commit 3745c5d:
+ crash log - add hex dump of function code
+ 1 file changed, 22 insertions(+)
+
+antirez in commit c1cc07b:
+ Sentinel example config: warn about protected mode.
+ 1 file changed, 16 insertions(+), 1 deletion(-)
+
+rojingeorge in commit 011dc9f:
+ Display the nodes summary once the cluster is established using redis-trib.rb
+ 1 file changed, 5 insertions(+)
+
+Guo Xiao in commit f4e3a94:
+ Use the standard predefined identifier __func__ (since C99)
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+================================================================================
+Redis 3.2.3     Released Tue Aug 02 10:55:24 CEST 2016
+================================================================================
+
+Upgrade urgency MODERATE: Fix replication delay and redis-cli security issue.
+
+Hello,
+
+this is a minor release of Redis addressing two bugs:
+
+1. There was an inverted if statement logic problem in replication.c causing
+   a replication delay. This is not an actual problem since things work
+   as expected, but worth to fix ASAP anyway.
+
+2. Redis-cli created the history file with insecure permissions, allowing
+   reding from the file. This was actually a bug in linenoise which is
+   now fixed. The applied fix is from Chris Lamb.
+
+List of commits:
+
+Qu Chen in commit e67ad1d:
+ Fix a bug to delay bgsave while AOF rewrite in progress for replication
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+antirez in commit 7153668:
+ Update linenoise to fix insecure redis-cli history file creation.
+ 1 file changed, 6 insertions(+), 1 deletion(-)
+
+================================================================================
+Redis 3.2.2     Released Thu Jul 28 14:14:54 CEST 2016
+================================================================================
+
+Upgrade urgency MODERATE: A Redis server and a Sentinel crash are now fixed.
+                          GEORADIUS errors in reported entries are fixed.
+
+Hi all! Redis 3.2.2 is finally out with non trivial bugfixes and a few new
+features. Let's start from the bugs:
+
+1. There was a bug in the List type implementation, able to cause the
+   crash of the server under certain (non trivial to replicate) circumstances
+   when the LSET command was used. Now the bug is fixed and a new stress tester
+   that was able to easily trigger the bug was added to the test suite.
+
+2. Redis Sentinel, when monitoring multiple masters, could crash after
+   a Sentinel address update event.
+
+3. Redis Sentinel now checks slaves INFO state more often when disconnected.
+   This is not really a bug fix, but may allow to more easily detect that
+   a slave is able to fail over its master reducing certain delays.
+
+4. It was possible, under a variety of conditions, that the AOF and RDB children
+   process could spawn at the same time. This is known to trash disk I/O,
+   AOF performances, and to ultimately create latency in the Redis server.
+   Normally Redis avoids to have the two writing children at the same time, but
+   there were edge cases discovered by Oran Agra (that also co-authored the
+   fix with me) where the double-fork could happen. In order to fix this bug
+   non trivial changes to the replication code were operated, however it was
+   important to back port this fix into 3.2.2 because the bug could lead to
+   bad latency experiences in certain cases.
+
+5. Many GEORADIUS bugs are now fixed \o/. This started as a failing CI
+   test. I grepped for more clues and there were a number of random failures
+   in the points reported by GEORADIUS. The errors were found to be related
+   to three different bugs (one of these was a bug in the test itself).
+   It's not a critical bug: the effect is to, sometimes, don't report objects
+   that are near the radius, but only with specific sets of coordinates
+   and radius settings. However now the issues are fixed and the error
+   vectors were added as regression tests.
+
+And now the good news (not that bug fixes are not good...), that is, the
+new features!
+
+1. Now slaves support the slave-announce-ip and slave-announce-port options.
+   Using these features a slave can be reported by the master `INFO` output
+   and `ROLE` command as having arbitrary IP and port. This allows to have
+   Sentinel deployments when working with containers or NAT-ed environments
+   more easily.
+
+2. The RDB check utlity is now part of Redis and uses the same RDB code that
+   Redis uses in order to load the dataset in memory, so a given version
+   of Redis is always able to check the RDB it produced... without another
+   external check tool which is supposed to be taken in sync with the
+   rdb.c implementation. This in turn also means that the new RDB checking
+   is able to spot more complex bugs, since it really loads the dataset
+   instead of just skipping bytes.
+
+About the redis-check-dump utility, now it is also able to show certain
+information about the RDB file, like the version that produced it, when
+it was produced, and so forth. Example:
+
+    ./redis-check-rdb dump.rdb
+    [offset 0] Checking RDB file dump.rdb
+    [offset 32] AUX FIELD redis-ver = '3.2.2'
+    [offset 46] AUX FIELD redis-bits = '64'
+    [offset 58] AUX FIELD ctime = '1469710178'
+    [offset 73] AUX FIELD used-mem = '1186528'
+
+The ability to check how much memory was used by the instance that produced
+the RDB file can be especially useful.
+
+For credits and the full change log, see the list of commits below, that was
+modified in order to be a bit less verbose compared to the last releases.
+
+Enjoy!
+Salvatore
+
+antirez in commit 0a45fbc:
+ Ability of slave to announce arbitrary ip/port to master.
+ 6 files changed, 129 insertions(+), 17 deletions(-)
+
+antirez in commit c3982c0:
+ redis-benchmark: new option to show server errors on stdout.
+ 1 file changed, 17 insertions(+), 1 deletion(-)
+
+antirez in commit fdafe23:
+ Multiple GEORADIUS bugs fixed.
+ 6 files changed, 177 insertions(+), 21 deletions(-)
+
+antirez in commit a1bfe22:
+ Replication: when possible start RDB saving ASAP.
+ 1 file changed, 8 insertions(+), 2 deletions(-)
+
+antirez in commit 7ca69af:
+ Sentinel: new test unit 07 that tests master down conditions.
+ 1 file changed, 68 insertions(+)
+
+antirez in commit 5b5e652:
+ Sentinel: check Slave INFO state more often when disconnected.
+ 2 files changed, 10 insertions(+), 3 deletions(-)
+
+antirez in commit 21cffc2:
+ Avoid simultaneous RDB and AOF child process.
+ 4 files changed, 51 insertions(+), 8 deletions(-)
+
+antirez in commit 017378e:
+ Replication: start BGSAVE for replication always in replicationCron().
+ 1 file changed, 15 insertions(+), 12 deletions(-)
+
+antirez in commit 940be9a:
+ Regression test for issue #3333.
+ 1 file changed, 6 insertions(+)
+
+antirez in commit 21736b4:
+ getLongLongFromObject: use string2ll() instead of strict_strtoll().
+ 1 file changed, 1 insertion(+), 6 deletions(-)
+
+antirez in commit 0b748e9:
+ redis-cli: check SELECT reply type just in state updated.
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+sskorgal in commit 1158386:
+ Fix for redis_cli printing default DB when select command fails.
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+antirez in commit 026f9fc:
+ Sentinel: fix cross-master Sentinel address update.
+ 1 file changed, 9 insertions(+), 2 deletions(-)
+
+antirez in commit 11523b3:
+ CONFIG GET is now no longer case sensitive.
+ 1 file changed, 12 insertions(+), 12 deletions(-)
+
+antirez in commit f5a7f4f:
+ Fix test for new RDB checksum failure message.
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+antirez in commit 4c6ff74:
+ Make tcp-keepalive default to 300 in internal conf.
+ 1 file changed, 1 insertion(+), 1 deletion(-)
+
+antirez in commit 27dbec2:
+ In Redis RDB check: more details in error reportings.
+ 1 file changed, 24 insertions(+), 13 deletions(-)
+
+antirez in commit 41f3004:
+ In Redis RDB check: log decompression errors.
+ 2 files changed, 23 insertions(+), 2 deletions(-)
+
+antirez in commit 278fe3e:
+ In Redis RDB check: log object type on error.
+ 1 file changed, 27 insertions(+), 2 deletions(-)
+
+antirez in commit a117dfa:
+ Added a trivial program to randomly corrupt RDB files in /utils.
+ 1 file changed, 44 insertions(+)
+
+antirez in commit f5110c3:
+ In Redis RDB check: minor output message changes.
+ 1 file changed, 4 insertions(+), 1 deletion(-)
+
+antirez in commit 35b18bf:
+ In Redis RDB check: better error reporting.
+ 4 files changed, 71 insertions(+), 14 deletions(-)
+
+antirez in commit f578f08:
+ In Redis RDB check: initial POC.
+ 2 files changed, 189 insertions(+), 647 deletions(-)
+
+tielei in commit 7f1e1ca:
+ A string with 21 chars is not representable as a 64-bit integer.
+ 1 file changed, 2 insertions(+), 2 deletions(-)
+
+antirez in commit 7a3a595:
+ Test: new randomized stress tester for #3343 alike bugs.
+ 1 file changed, 27 insertions(+), 2 deletions(-)
+
+antirez in commit c75ca10:
+ Stress tester WIP.
+ 1 file changed, 3 insertions(+)
+
+antirez in commit 2c3fcf8:
+ Regression test for issue #3343 exact min crash sequence.
+ 1 file changed, 16 insertions(+)
+
+antirez in commit 7041967:
+ Fix quicklistReplaceAtIndex() by updating the quicklist ziplist size.
+ 1 file changed, 1 insertion(+)
 
 ================================================================================
 Redis 3.2.1     Released Fri Jun 17 15:01:56 CEST 2016

deps/geohash-int/geohash.c

@@ -1,7 +1,7 @@
 /*
  * Copyright (c) 2013-2014, yinqiwen <yinqiwen@gmail.com>
  * Copyright (c) 2014, Matt Stancliff <matt@genges.com>.
- * Copyright (c) 2015, Salvatore Sanfilippo <antirez@gmail.com>.
+ * Copyright (c) 2015-2016, Salvatore Sanfilippo <antirez@gmail.com>.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -118,7 +118,7 @@ void geohashGetCoordRange(GeoHashRange *long_range, GeoHashRange *lat_range) {
     lat_range->min = GEO_LAT_MIN;
 }
 
-int geohashEncode(GeoHashRange *long_range, GeoHashRange *lat_range,
+int geohashEncode(const GeoHashRange *long_range, const GeoHashRange *lat_range,
                   double longitude, double latitude, uint8_t step,
                   GeoHashBits *hash) {
     /* Check basic arguments sanity. */

deps/geohash-int/geohash.h

@@ -95,7 +95,7 @@ typedef struct {
  * -1:failed
  */
 void geohashGetCoordRange(GeoHashRange *long_range, GeoHashRange *lat_range);
-int geohashEncode(GeoHashRange *long_range, GeoHashRange *lat_range,
+int geohashEncode(const GeoHashRange *long_range, const GeoHashRange *lat_range,
                   double longitude, double latitude, uint8_t step,
                   GeoHashBits *hash);
 int geohashEncodeType(double longitude, double latitude,

deps/geohash-int/geohash_helper.c

@@ -1,7 +1,7 @@
 /*
  * Copyright (c) 2013-2014, yinqiwen <yinqiwen@gmail.com>
  * Copyright (c) 2014, Matt Stancliff <matt@genges.com>.
- * Copyright (c) 2015, Salvatore Sanfilippo <antirez@gmail.com>.
+ * Copyright (c) 2015-2016, Salvatore Sanfilippo <antirez@gmail.com>.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -35,6 +35,7 @@
  */
 
 #include "geohash_helper.h"
+#include <math.h>
 
 #define D_R (M_PI / 180.0)
 #define R_MAJOR 6378137.0
@@ -54,8 +55,8 @@ const double MERCATOR_MIN = -20037726.37;
 static inline double deg_rad(double ang) { return ang * D_R; }
 static inline double rad_deg(double ang) { return ang / D_R; }
 
-/* You must *ONLY* estimate steps when you are encoding.
- * If you are decoding, always decode to GEO_STEP_MAX (26). */
+/* This function is used in order to estimate the step (bits precision)
+ * of the 9 search area boxes during radius queries. */
 uint8_t geohashEstimateStepsByRadius(double range_meters, double lat) {
     if (range_meters == 0) return 26;
     int step = 1;
@@ -63,12 +64,15 @@ uint8_t geohashEstimateStepsByRadius(double range_meters, double lat) {
         range_meters *= 2;
         step++;
     }
-    step -= 2; /* Make sure range is included in the worst case. */
+    step -= 2; /* Make sure range is included in most of the base cases. */
+
     /* Wider range torwards the poles... Note: it is possible to do better
      * than this approximation by computing the distance between meridians
      * at this latitude, but this does the trick for now. */
-    if (lat > 67 || lat < -67) step--;
-    if (lat > 80 || lat < -80) step--;
+    if (lat > 66 || lat < -66) {
+        step--;
+        if (lat > 80 || lat < -80) step--;
+    }
 
     /* Frame to valid range. */
     if (step < 1) step = 1;
@@ -108,12 +112,14 @@ int geohashBoundingBox(double longitude, double latitude, double radius_meters,
     return 1;
 }
 
+/* Return a set of areas (center + 8) that are able to cover a range query
+ * for the specified position and radius. */
 GeoHashRadius geohashGetAreasByRadius(double longitude, double latitude, double radius_meters) {
     GeoHashRange long_range, lat_range;
-    GeoHashRadius radius = { { 0 } };
-    GeoHashBits hash = { 0 };
-    GeoHashNeighbors neighbors = { { 0 } };
-    GeoHashArea area = { { 0 } };
+    GeoHashRadius radius;
+    GeoHashBits hash;
+    GeoHashNeighbors neighbors;
+    GeoHashArea area;
     double min_lon, max_lon, min_lat, max_lat;
     double bounds[4];
     int steps;
@@ -126,12 +132,43 @@ GeoHashRadius geohashGetAreasByRadius(double longitude, double latitude, double
 
     steps = geohashEstimateStepsByRadius(radius_meters,latitude);
 
-    geohashGetCoordRange(&long_range, &lat_range);
-    geohashEncode(&long_range, &lat_range, longitude, latitude, steps, &hash);
-    geohashNeighbors(&hash, &neighbors);
-    geohashGetCoordRange(&long_range, &lat_range);
-    geohashDecode(long_range, lat_range, hash, &area);
+    geohashGetCoordRange(&long_range,&lat_range);
+    geohashEncode(&long_range,&lat_range,longitude,latitude,steps,&hash);
+    geohashNeighbors(&hash,&neighbors);
+    geohashDecode(long_range,lat_range,hash,&area);
 
+    /* Check if the step is enough at the limits of the covered area.
+     * Sometimes when the search area is near an edge of the
+     * area, the estimated step is not small enough, since one of the
+     * north / south / west / east square is too near to the search area
+     * to cover everything. */
+    int decrease_step = 0;
+    {
+        GeoHashArea north, south, east, west;
+
+        geohashDecode(long_range, lat_range, neighbors.north, &north);
+        geohashDecode(long_range, lat_range, neighbors.south, &south);
+        geohashDecode(long_range, lat_range, neighbors.east, &east);
+        geohashDecode(long_range, lat_range, neighbors.west, &west);
+
+        if (geohashGetDistance(longitude,latitude,longitude,north.latitude.max)
+            < radius_meters) decrease_step = 1;
+        if (geohashGetDistance(longitude,latitude,longitude,south.latitude.min)
+            < radius_meters) decrease_step = 1;
+        if (geohashGetDistance(longitude,latitude,east.longitude.max,latitude)
+            < radius_meters) decrease_step = 1;
+        if (geohashGetDistance(longitude,latitude,west.longitude.min,latitude)
+            < radius_meters) decrease_step = 1;
+    }
+
+    if (decrease_step) {
+        steps--;
+        geohashEncode(&long_range,&lat_range,longitude,latitude,steps,&hash);
+        geohashNeighbors(&hash,&neighbors);
+        geohashDecode(long_range,lat_range,hash,&area);
+    }
+
+    /* Exclude the search areas that are useless. */
     if (area.latitude.min < min_lat) {
         GZERO(neighbors.south);
         GZERO(neighbors.south_west);

deps/linenoise/linenoise.c

@@ -111,6 +111,7 @@
 #include <string.h>
 #include <stdlib.h>
 #include <ctype.h>
+#include <sys/stat.h>
 #include <sys/types.h>
 #include <sys/ioctl.h>
 #include <unistd.h>
@@ -1160,10 +1161,14 @@ int linenoiseHistorySetMaxLen(int len) {
 /* Save the history in the specified file. On success 0 is returned
  * otherwise -1 is returned. */
 int linenoiseHistorySave(const char *filename) {
-    FILE *fp = fopen(filename,"w");
+    mode_t old_umask = umask(S_IXUSR|S_IRWXG|S_IRWXO);
+    FILE *fp;
     int j;
 
+    fp = fopen(filename,"w");
+    umask(old_umask);
     if (fp == NULL) return -1;
+    chmod(filename,S_IRUSR|S_IWUSR);
     for (j = 0; j < history_len; j++)
         fprintf(fp,"%s\n",history[j]);
     fclose(fp);

redis.conf

@@ -435,6 +435,35 @@ slave-priority 100
 # By default min-slaves-to-write is set to 0 (feature disabled) and
 # min-slaves-max-lag is set to 10.
 
+# A Redis master is able to list the address and port of the attached
+# slaves in different ways. For example the "INFO replication" section
+# offers this information, which is used, among other tools, by
+# Redis Sentinel in order to discover slave instances.
+# Another place where this info is available is in the output of the
+# "ROLE" command of a masteer.
+#
+# The listed IP and address normally reported by a slave is obtained
+# in the following way:
+#
+#   IP: The address is auto detected by checking the peer address
+#   of the socket used by the slave to connect with the master.
+#
+#   Port: The port is communicated by the slave during the replication
+#   handshake, and is normally the port that the slave is using to
+#   list for connections.
+#
+# However when port forwarding or Network Address Translation (NAT) is
+# used, the slave may be actually reachable via different IP and port
+# pairs. The following two options can be used by a slave in order to
+# report to its master a specific set of IP and port, so that both INFO
+# and ROLE will report those values.
+#
+# There is no need to use both the options if you need to override just
+# the port or the IP address.
+#
+# slave-announce-ip 5.5.5.5
+# slave-announce-port 1234
+
 ################################## SECURITY ###################################
 
 # Require clients to issue AUTH <PASSWORD> before processing any other

sentinel.conf

@@ -1,5 +1,21 @@
 # Example sentinel.conf
 
+# *** IMPORTANT ***
+#
+# By default Sentinel will not be reachable from interfaces different than
+# localhost, either use the 'bind' directive to bind to a list of network
+# interfaces, or disable protected mode with "protected-mode no" by
+# adding it to this configuration file.
+#
+# Before doing that MAKE SURE the instance is protected from the outside
+# world via firewalling or other means.
+#
+# For example you may use one of the following:
+#
+# bind 127.0.0.1 192.168.1.1
+#
+# protected-mode no
+
 # port <sentinel-port>
 # The port that this sentinel instance will run on
 port 26379
@@ -178,4 +194,3 @@ sentinel failover-timeout mymaster 180000
 #
 # sentinel client-reconfig-script mymaster /var/redis/reconfig.sh
 
-

src/Makefile

@@ -65,17 +65,27 @@ ifeq ($(uname_S),SunOS)
 	FINAL_LIBS+= -ldl -lnsl -lsocket -lresolv -lpthread -lrt
 else
 ifeq ($(uname_S),Darwin)
-	# Darwin (nothing to do)
+	# Darwin
+	FINAL_LIBS+= -ldl
 else
 ifeq ($(uname_S),AIX)
         # AIX
         FINAL_LDFLAGS+= -Wl,-bexpall
-        FINAL_LIBS+= -pthread -lcrypt -lbsd
-
+        FINAL_LIBS+=-ldl -pthread -lcrypt -lbsd
+else
+ifeq ($(uname_S),OpenBSD)
+	# OpenBSD
+	FINAL_LIBS+= -lpthread
+else
+ifeq ($(uname_S),FreeBSD)
+	# FreeBSD
+	FINAL_LIBS+= -lpthread
 else
 	# All the other OSes (notably Linux)
 	FINAL_LDFLAGS+= -rdynamic
-	FINAL_LIBS+= -pthread
+	FINAL_LIBS+=-ldl -pthread
+endif
+endif
 endif
 endif
 endif
@@ -95,7 +105,7 @@ endif
 ifeq ($(MALLOC),jemalloc)
 	DEPENDENCY_TARGETS+= jemalloc
 	FINAL_CFLAGS+= -DUSE_JEMALLOC -I../deps/jemalloc/include
-	FINAL_LIBS+= ../deps/jemalloc/lib/libjemalloc.a -ldl
+	FINAL_LIBS+= ../deps/jemalloc/lib/libjemalloc.a
 endif
 
 REDIS_CC=$(QUIET_CC)$(CC) $(FINAL_CFLAGS)

src/anet.c

@@ -486,7 +486,7 @@ static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backl
         goto end;
     }
     if (p == NULL) {
-        anetSetError(err, "unable to bind socket");
+        anetSetError(err, "unable to bind socket, errno: %d", errno);
         goto error;
     }
 

src/aof.c

@@ -251,7 +251,10 @@ int startAppendOnly(void) {
             strerror(errno));
         return C_ERR;
     }
-    if (rewriteAppendOnlyFileBackground() == C_ERR) {
+    if (server.rdb_child_pid != -1) {
+        server.aof_rewrite_scheduled = 1;
+        serverLog(LL_WARNING,"AOF was enabled but there is already a child process saving an RDB file on disk. An AOF background was scheduled to start when possible.");
+    } else if (rewriteAppendOnlyFileBackground() == C_ERR) {
         close(server.aof_fd);
         serverLog(LL_WARNING,"Redis needs to enable the AOF but can't trigger a background AOF rewrite operation. Check the above logs for more info about the error.");
         return C_ERR;
@@ -1261,7 +1264,7 @@ int rewriteAppendOnlyFileBackground(void) {
     pid_t childpid;
     long long start;
 
-    if (server.aof_child_pid != -1) return C_ERR;
+    if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR;
     if (aofCreatePipes() != C_OK) return C_ERR;
     start = ustime();
     if ((childpid = fork()) == 0) {

src/config.c

@@ -549,8 +549,9 @@ void loadServerConfigFromString(char *config) {
             unsigned long long hard, soft;
             int soft_seconds;
 
-            if (class == -1) {
-                err = "Unrecognized client limit class";
+            if (class == -1 || class == CLIENT_TYPE_MASTER) {
+                err = "Unrecognized client limit class: the user specified "
+                "an invalid one, or 'master' which has no buffer limits.";
                 goto loaderr;
             }
             hard = memtoll(argv[2],NULL);
@@ -570,6 +571,16 @@ void loadServerConfigFromString(char *config) {
             }
         } else if (!strcasecmp(argv[0],"slave-priority") && argc == 2) {
             server.slave_priority = atoi(argv[1]);
+        } else if (!strcasecmp(argv[0],"slave-announce-ip") && argc == 2) {
+            zfree(server.slave_announce_ip);
+            server.slave_announce_ip = zstrdup(argv[1]);
+        } else if (!strcasecmp(argv[0],"slave-announce-port") && argc == 2) {
+            server.slave_announce_port = atoi(argv[1]);
+            if (server.slave_announce_port < 0 ||
+                server.slave_announce_port > 65535)
+            {
+                err = "Invalid port"; goto loaderr;
+            }
         } else if (!strcasecmp(argv[0],"min-slaves-to-write") && argc == 2) {
             server.repl_min_slaves_to_write = atoi(argv[1]);
             if (server.repl_min_slaves_to_write < 0) {
@@ -824,7 +835,8 @@ void configSetCommand(client *c) {
             long val;
 
             if ((j % 4) == 0) {
-                if (getClientTypeByName(v[j]) == -1) {
+                int class = getClientTypeByName(v[j]);
+                if (class == -1 || class == CLIENT_TYPE_MASTER) {
                     sdsfreesplitres(v,vlen);
                     goto badfmt;
                 }
@@ -857,6 +869,9 @@ void configSetCommand(client *c) {
 
         if (flags == -1) goto badfmt;
         server.notify_keyspace_events = flags;
+    } config_set_special_field("slave-announce-ip") {
+        zfree(server.slave_announce_ip);
+        server.slave_announce_ip = ((char*)o->ptr)[0] ? zstrdup(o->ptr) : NULL;
 
     /* Boolean fields.
      * config_set_bool_field(name,var). */
@@ -933,6 +948,8 @@ void configSetCommand(client *c) {
       "repl-diskless-sync-delay",server.repl_diskless_sync_delay,0,LLONG_MAX) {
     } config_set_numerical_field(
       "slave-priority",server.slave_priority,0,LLONG_MAX) {
+    } config_set_numerical_field(
+      "slave-announce-port",server.slave_announce_port,0,65535) {
     } config_set_numerical_field(
       "min-slaves-to-write",server.repl_min_slaves_to_write,0,LLONG_MAX) {
         refreshGoodSlavesCount();
@@ -1001,7 +1018,7 @@ badfmt: /* Bad format errors */
  *----------------------------------------------------------------------------*/
 
 #define config_get_string_field(_name,_var) do { \
-    if (stringmatch(pattern,_name,0)) { \
+    if (stringmatch(pattern,_name,1)) { \
         addReplyBulkCString(c,_name); \
         addReplyBulkCString(c,_var ? _var : ""); \
         matches++; \
@@ -1009,7 +1026,7 @@ badfmt: /* Bad format errors */
 } while(0);
 
 #define config_get_bool_field(_name,_var) do { \
-    if (stringmatch(pattern,_name,0)) { \
+    if (stringmatch(pattern,_name,1)) { \
         addReplyBulkCString(c,_name); \
         addReplyBulkCString(c,_var ? "yes" : "no"); \
         matches++; \
@@ -1017,7 +1034,7 @@ badfmt: /* Bad format errors */
 } while(0);
 
 #define config_get_numerical_field(_name,_var) do { \
-    if (stringmatch(pattern,_name,0)) { \
+    if (stringmatch(pattern,_name,1)) { \
         ll2string(buf,sizeof(buf),_var); \
         addReplyBulkCString(c,_name); \
         addReplyBulkCString(c,buf); \
@@ -1026,7 +1043,7 @@ badfmt: /* Bad format errors */
 } while(0);
 
 #define config_get_enum_field(_name,_var,_enumvar) do { \
-    if (stringmatch(pattern,_name,0)) { \
+    if (stringmatch(pattern,_name,1)) { \
         addReplyBulkCString(c,_name); \
         addReplyBulkCString(c,configEnumGetNameOrUnknown(_enumvar,_var)); \
         matches++; \
@@ -1048,6 +1065,7 @@ void configGetCommand(client *c) {
     config_get_string_field("unixsocket",server.unixsocket);
     config_get_string_field("logfile",server.logfile);
     config_get_string_field("pidfile",server.pidfile);
+    config_get_string_field("slave-announce-ip",server.slave_announce_ip);
 
     /* Numerical values */
     config_get_numerical_field("maxmemory",server.maxmemory);
@@ -1090,6 +1108,7 @@ void configGetCommand(client *c) {
     config_get_numerical_field("maxclients",server.maxclients);
     config_get_numerical_field("watchdog-period",server.watchdog_period);
     config_get_numerical_field("slave-priority",server.slave_priority);
+    config_get_numerical_field("slave-announce-port",server.slave_announce_port);
     config_get_numerical_field("min-slaves-to-write",server.repl_min_slaves_to_write);
     config_get_numerical_field("min-slaves-max-lag",server.repl_min_slaves_max_lag);
     config_get_numerical_field("hz",server.hz);
@@ -1138,12 +1157,12 @@ void configGetCommand(client *c) {
 
     /* Everything we can't handle with macros follows. */
 
-    if (stringmatch(pattern,"appendonly",0)) {
+    if (stringmatch(pattern,"appendonly",1)) {
         addReplyBulkCString(c,"appendonly");
         addReplyBulkCString(c,server.aof_state == AOF_OFF ? "no" : "yes");
         matches++;
     }
-    if (stringmatch(pattern,"dir",0)) {
+    if (stringmatch(pattern,"dir",1)) {
         char buf[1024];
 
         if (getcwd(buf,sizeof(buf)) == NULL)
@@ -1153,7 +1172,7 @@ void configGetCommand(client *c) {
         addReplyBulkCString(c,buf);
         matches++;
     }
-    if (stringmatch(pattern,"save",0)) {
+    if (stringmatch(pattern,"save",1)) {
         sds buf = sdsempty();
         int j;
 
@@ -1169,7 +1188,7 @@ void configGetCommand(client *c) {
         sdsfree(buf);
         matches++;
     }
-    if (stringmatch(pattern,"client-output-buffer-limit",0)) {
+    if (stringmatch(pattern,"client-output-buffer-limit",1)) {
         sds buf = sdsempty();
         int j;
 
@@ -1187,14 +1206,14 @@ void configGetCommand(client *c) {
         sdsfree(buf);
         matches++;
     }
-    if (stringmatch(pattern,"unixsocketperm",0)) {
+    if (stringmatch(pattern,"unixsocketperm",1)) {
         char buf[32];
         snprintf(buf,sizeof(buf),"%o",server.unixsocketperm);
         addReplyBulkCString(c,"unixsocketperm");
         addReplyBulkCString(c,buf);
         matches++;
     }
-    if (stringmatch(pattern,"slaveof",0)) {
+    if (stringmatch(pattern,"slaveof",1)) {
         char buf[256];
 
         addReplyBulkCString(c,"slaveof");
@@ -1206,7 +1225,7 @@ void configGetCommand(client *c) {
         addReplyBulkCString(c,buf);
         matches++;
     }
-    if (stringmatch(pattern,"notify-keyspace-events",0)) {
+    if (stringmatch(pattern,"notify-keyspace-events",1)) {
         robj *flagsobj = createObject(OBJ_STRING,
             keyspaceEventsFlagsToString(server.notify_keyspace_events));
 
@@ -1215,7 +1234,7 @@ void configGetCommand(client *c) {
         decrRefCount(flagsobj);
         matches++;
     }
-    if (stringmatch(pattern,"bind",0)) {
+    if (stringmatch(pattern,"bind",1)) {
         sds aux = sdsjoin(server.bindaddr,server.bindaddr_count," ");
 
         addReplyBulkCString(c,"bind");
@@ -1768,6 +1787,7 @@ int rewriteConfig(char *path) {
     rewriteConfigOctalOption(state,"unixsocketperm",server.unixsocketperm,CONFIG_DEFAULT_UNIX_SOCKET_PERM);
     rewriteConfigNumericalOption(state,"timeout",server.maxidletime,CONFIG_DEFAULT_CLIENT_TIMEOUT);
     rewriteConfigNumericalOption(state,"tcp-keepalive",server.tcpkeepalive,CONFIG_DEFAULT_TCP_KEEPALIVE);
+    rewriteConfigNumericalOption(state,"slave-announce-port",server.slave_announce_port,CONFIG_DEFAULT_SLAVE_ANNOUNCE_PORT);
     rewriteConfigEnumOption(state,"loglevel",server.verbosity,loglevel_enum,CONFIG_DEFAULT_VERBOSITY);
     rewriteConfigStringOption(state,"logfile",server.logfile,CONFIG_DEFAULT_LOGFILE);
     rewriteConfigYesNoOption(state,"syslog-enabled",server.syslog_enabled,CONFIG_DEFAULT_SYSLOG_ENABLED);
@@ -1781,6 +1801,7 @@ int rewriteConfig(char *path) {
     rewriteConfigStringOption(state,"dbfilename",server.rdb_filename,CONFIG_DEFAULT_RDB_FILENAME);
     rewriteConfigDirOption(state);
     rewriteConfigSlaveofOption(state);
+    rewriteConfigStringOption(state,"slave-announce-ip",server.slave_announce_ip,CONFIG_DEFAULT_SLAVE_ANNOUNCE_IP);
     rewriteConfigStringOption(state,"masterauth",server.masterauth,NULL);
     rewriteConfigYesNoOption(state,"slave-serve-stale-data",server.repl_serve_stale_data,CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA);
     rewriteConfigYesNoOption(state,"slave-read-only",server.repl_slave_ro,CONFIG_DEFAULT_SLAVE_READ_ONLY);

src/debug.c

@@ -39,6 +39,8 @@
 #include <ucontext.h>
 #include <fcntl.h>
 #include "bio.h"
+#include <unistd.h>
+#include <dlfcn.h>
 #endif /* HAVE_BACKTRACE */
 
 #ifdef __CYGWIN__
@@ -671,6 +673,8 @@ static void *getMcontextEip(ucontext_t *uc) {
     return (void*) uc->uc_mcontext.gregs[16]; /* Linux 64 */
     #elif defined(__ia64__) /* Linux IA64 */
     return (void*) uc->uc_mcontext.sc_ip;
+    #elif defined(__arm__) /* Linux ARM */
+    return (void*) uc->uc_mcontext.arm_pc;
     #endif
 #else
     return NULL;
@@ -972,6 +976,32 @@ int memtest_test_linux_anonymous_maps(void) {
 }
 #endif
 
+/* Scans the (assumed) x86 code starting at addr, for a max of `len`
+ * bytes, searching for E8 (callq) opcodes, and dumping the symbols
+ * and the call offset if they appear to be valid. */
+void dumpX86Calls(void *addr, size_t len) {
+    size_t j;
+    unsigned char *p = addr;
+    Dl_info info;
+    /* Hash table to best-effort avoid printing the same symbol
+     * multiple times. */
+    unsigned long ht[256] = {0};
+
+    if (len < 5) return;
+    for (j = 0; j < len-4; j++) {
+        if (p[j] != 0xE8) continue; /* Not an E8 CALL opcode. */
+        unsigned long target = (unsigned long)addr+j+5;
+        target += *((int32_t*)(p+j+1));
+        if (dladdr((void*)target, &info) != 0 && info.dli_sname != NULL) {
+            if (ht[target&0xff] != target) {
+                printf("Function at 0x%lx is %s\n",target,info.dli_sname);
+                ht[target&0xff] = target;
+            }
+            j += 4; /* Skip the 32 bit immediate. */
+        }
+    }
+}
+
 void sigsegvHandler(int sig, siginfo_t *info, void *secret) {
     ucontext_t *uc = (ucontext_t*) secret;
     void *eip = getMcontextEip(uc);
@@ -1022,19 +1052,49 @@ void sigsegvHandler(int sig, siginfo_t *info, void *secret) {
     bioKillThreads();
     if (memtest_test_linux_anonymous_maps()) {
         serverLogRaw(LL_WARNING|LL_RAW,
-            "!!! MEMORY ERROR DETECTED! Check your memory ASAP !!!");
+            "!!! MEMORY ERROR DETECTED! Check your memory ASAP !!!\n");
     } else {
         serverLogRaw(LL_WARNING|LL_RAW,
-            "Fast memory test PASSED, however your memory can still be broken. Please run a memory test for several hours if possible.");
+            "Fast memory test PASSED, however your memory can still be broken. Please run a memory test for several hours if possible.\n");
     }
 #endif
 
+    if (eip != NULL) {
+        Dl_info info;
+        if (dladdr(eip, &info) != 0) {
+            serverLog(LL_WARNING|LL_RAW,
+                "\n------ DUMPING CODE AROUND EIP ------\n"
+                "Symbol: %s (base: %p)\n"
+                "Module: %s (base %p)\n"
+                "$ xxd -r -p /tmp/dump.hex /tmp/dump.bin\n"
+                "$ objdump --adjust-vma=%p -D -b binary -m i386:x86-64 /tmp/dump.bin\n"
+                "------\n",
+                info.dli_sname, info.dli_saddr, info.dli_fname, info.dli_fbase,
+                info.dli_saddr);
+            size_t len = (long)eip - (long)info.dli_saddr;
+            unsigned long sz = sysconf(_SC_PAGESIZE);
+            if (len < 1<<13) { /* we don't have functions over 8k (verified) */
+                /* Find the address of the next page, which is our "safety"
+                 * limit when dumping. Then try to dump just 128 bytes more
+                 * than EIP if there is room, or stop sooner. */
+                unsigned long next = ((unsigned long)eip + sz) & ~(sz-1);
+                unsigned long end = (unsigned long)eip + 128;
+                if (end > next) end = next;
+                len = end - (unsigned long)info.dli_saddr;
+                serverLogHexDump(LL_WARNING, "dump of function",
+                    info.dli_saddr ,len);
+                dumpX86Calls(info.dli_saddr,len);
+            }
+        }
+    }
+
     serverLogRaw(LL_WARNING|LL_RAW,
 "\n=== REDIS BUG REPORT END. Make sure to include from START to END. ===\n\n"
 "       Please report the crash by opening an issue on github:\n\n"
 "           http://github.com/antirez/redis/issues\n\n"
 "  Suspect RAM error? Use redis-server --test-memory to verify it.\n\n"
 );
+
     /* free(messages); Don't call free() with possibly corrupted memory. */
     if (server.daemonize && server.supervised == 0) unlink(server.pidfile);
 
@@ -1055,7 +1115,7 @@ void serverLogHexDump(int level, char *descr, void *value, size_t len) {
     unsigned char *v = value;
     char charset[] = "0123456789abcdef";
 
-    serverLog(level,"%s (hexdump):", descr);
+    serverLog(level,"%s (hexdump of %zu bytes):", descr, len);
     b = buf;
     while(len) {
         b[0] = charset[(*v)>>4];

src/debugmacro.h

@@ -0,0 +1,41 @@
+/* This file contains debugging macros to be used when investigating issues.
+ *
+ * -----------------------------------------------------------------------------
+ *
+ * Copyright (c) 2016, Salvatore Sanfilippo <antirez at gmail dot com>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include <stdio.h>
+#define D(...)                                                               \
+    do {                                                                     \
+        FILE *fp = fopen("/tmp/log.txt","a");                                \
+        fprintf(fp,"%s:%s:%d:\t", __FILE__, __func__, __LINE__);             \
+        fprintf(fp,__VA_ARGS__);                                             \
+        fprintf(fp,"\n");                                                    \
+        fclose(fp);                                                          \
+    } while (0);

src/geo.c

@@ -1,6 +1,6 @@
 /*
  * Copyright (c) 2014, Matt Stancliff <matt@genges.com>.
- * Copyright (c) 2015, Salvatore Sanfilippo <antirez@gmail.com>.
+ * Copyright (c) 2015-2016, Salvatore Sanfilippo <antirez@gmail.com>.
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -30,6 +30,7 @@
 
 #include "geo.h"
 #include "geohash_helper.h"
+#include "debugmacro.h"
 
 /* Things exported from t_zset.c only for geo.c, since it is the only other
  * part of Redis that requires close zset introspection. */

src/networking.c

@@ -105,6 +105,7 @@ client *createClient(int fd) {
     c->repl_ack_off = 0;
     c->repl_ack_time = 0;
     c->slave_listening_port = 0;
+    c->slave_ip[0] = '\0';
     c->slave_capa = SLAVE_CAPA_NONE;
     c->reply = listCreate();
     c->reply_bytes = 0;

src/object.c

@@ -380,10 +380,10 @@ robj *tryObjectEncoding(robj *o) {
      if (o->refcount > 1) return o;
 
     /* Check if we can represent this string as a long integer.
-     * Note that we are sure that a string larger than 21 chars is not
+     * Note that we are sure that a string larger than 20 chars is not
      * representable as a 32 nor 64 bit integer. */
     len = sdslen(s);
-    if (len <= 21 && string2l(s,len,&value)) {
+    if (len <= 20 && string2l(s,len,&value)) {
         /* This object is encodable as a long. Try to use a shared object.
          * Note that we avoid using shared integers when maxmemory is used
          * because every object needs to have a private LRU field for the LRU
@@ -616,18 +616,13 @@ int getLongDoubleFromObjectOrReply(client *c, robj *o, long double *target, cons
 
 int getLongLongFromObject(robj *o, long long *target) {
     long long value;
-    char *eptr;
 
     if (o == NULL) {
         value = 0;
     } else {
         serverAssertWithInfo(NULL,o,o->type == OBJ_STRING);
         if (sdsEncodedObject(o)) {
-            errno = 0;
-            value = strtoll(o->ptr, &eptr, 10);
-            if (isspace(((char*)o->ptr)[0]) || eptr[0] != '\0' ||
-                errno == ERANGE)
-                return C_ERR;
+            if (string2ll(o->ptr,sdslen(o->ptr),&value) == 0) return C_ERR;
         } else if (o->encoding == OBJ_ENCODING_INT) {
             value = (long)o->ptr;
         } else {

src/quicklist.c

@@ -671,6 +671,7 @@ int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
         /* quicklistIndex provides an uncompressed node */
         entry.node->zl = ziplistDelete(entry.node->zl, &entry.zi);
         entry.node->zl = ziplistInsert(entry.node->zl, entry.zi, data, sz);
+        quicklistNodeUpdateSz(entry.node);
         quicklistCompress(quicklist, entry.node);
         return 1;
     } else {

src/rdb.c

@@ -45,13 +45,30 @@
 #define RDB_LOAD_ENC    (1<<0)
 #define RDB_LOAD_PLAIN  (1<<1)
 
-#define rdbExitReportCorruptRDB(reason) rdbCheckThenExit(reason, __LINE__);
+#define rdbExitReportCorruptRDB(...) rdbCheckThenExit(__LINE__,__VA_ARGS__)
 
-void rdbCheckThenExit(char *reason, int where) {
-    serverLog(LL_WARNING, "Corrupt RDB detected at rdb.c:%d (%s). "
-        "Running 'redis-check-rdb %s'",
-        where, reason, server.rdb_filename);
-    redis_check_rdb(server.rdb_filename);
+extern int rdbCheckMode;
+void rdbCheckError(const char *fmt, ...);
+void rdbCheckSetError(const char *fmt, ...);
+
+void rdbCheckThenExit(int linenum, char *reason, ...) {
+    va_list ap;
+    char msg[1024];
+    int len;
+
+    len = snprintf(msg,sizeof(msg),
+        "Internal error in RDB reading function at rdb.c:%d -> ", linenum);
+    va_start(ap,reason);
+    vsnprintf(msg+len,sizeof(msg)-len,reason,ap);
+    va_end(ap);
+
+    if (!rdbCheckMode) {
+        serverLog(LL_WARNING, "%s", msg);
+        char *argv[2] = {"",server.rdb_filename};
+        redis_check_rdb_main(2,argv);
+    } else {
+        rdbCheckError("%s",msg);
+    }
     exit(1);
 }
 
@@ -142,10 +159,14 @@ uint32_t rdbLoadLen(rio *rdb, int *isencoded) {
         /* Read a 14 bit len. */
         if (rioRead(rdb,buf+1,1) == 0) return RDB_LENERR;
         return ((buf[0]&0x3F)<<8)|buf[1];
-    } else {
+    } else if (type == RDB_32BITLEN) {
         /* Read a 32 bit len. */
         if (rioRead(rdb,&len,4) == 0) return RDB_LENERR;
         return ntohl(len);
+    } else {
+        rdbExitReportCorruptRDB(
+            "Unknown length encoding %d in rdbLoadLen()",type);
+        return -1; /* Never reached. */
     }
 }
 
@@ -199,7 +220,7 @@ void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags) {
         val = (int32_t)v;
     } else {
         val = 0; /* anti-warning */
-        rdbExitReportCorruptRDB("Unknown RDB integer encoding type");
+        rdbExitReportCorruptRDB("Unknown RDB integer encoding type %d",enctype);
     }
     if (plain) {
         char buf[LONG_STR_SIZE], *p;
@@ -298,7 +319,10 @@ void *rdbLoadLzfStringObject(rio *rdb, int flags) {
 
     /* Load the compressed representation and uncompress it to target. */
     if (rioRead(rdb,c,clen) == 0) goto err;
-    if (lzf_decompress(c,clen,val,len) == 0) goto err;
+    if (lzf_decompress(c,clen,val,len) == 0) {
+        if (rdbCheckMode) rdbCheckSetError("Invalid LZF compressed string");
+        goto err;
+    }
     zfree(c);
 
     if (plain)
@@ -406,7 +430,7 @@ void *rdbGenericLoadStringObject(rio *rdb, int flags) {
         case RDB_ENC_LZF:
             return rdbLoadLzfStringObject(rdb,flags);
         default:
-            rdbExitReportCorruptRDB("Unknown RDB encoding type");
+            rdbExitReportCorruptRDB("Unknown RDB string encoding type %d",len);
         }
     }
 
@@ -895,7 +919,7 @@ int rdbSaveBackground(char *filename) {
     pid_t childpid;
     long long start;
 
-    if (server.rdb_child_pid != -1) return C_ERR;
+    if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR;
 
     server.dirty_before_bgsave = server.dirty;
     server.lastbgsave_try = time(NULL);
@@ -1192,11 +1216,11 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
                     hashTypeConvert(o, OBJ_ENCODING_HT);
                 break;
             default:
-                rdbExitReportCorruptRDB("Unknown encoding");
+                rdbExitReportCorruptRDB("Unknown RDB encoding type %d",rdbtype);
                 break;
         }
     } else {
-        rdbExitReportCorruptRDB("Unknown object type");
+        rdbExitReportCorruptRDB("Unknown RDB encoding type %d",rdbtype);
     }
     return o;
 }
@@ -1562,7 +1586,7 @@ int rdbSaveToSlavesSockets(void) {
     long long start;
     int pipefds[2];
 
-    if (server.rdb_child_pid != -1) return C_ERR;
+    if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR;
 
     /* Before to fork, create a pipe that will be used in order to
      * send back to the parent the IDs of the slaves that successfully
@@ -1717,11 +1741,33 @@ void saveCommand(client *c) {
     }
 }
 
+/* BGSAVE [SCHEDULE] */
 void bgsaveCommand(client *c) {
+    int schedule = 0;
+
+    /* The SCHEDULE option changes the behavior of BGSAVE when an AOF rewrite
+     * is in progress. Instead of returning an error a BGSAVE gets scheduled. */
+    if (c->argc > 1) {
+        if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"schedule")) {
+            schedule = 1;
+        } else {
+            addReply(c,shared.syntaxerr);
+            return;
+        }
+    }
+
     if (server.rdb_child_pid != -1) {
         addReplyError(c,"Background save already in progress");
     } else if (server.aof_child_pid != -1) {
-        addReplyError(c,"Can't BGSAVE while AOF log rewriting is in progress");
+        if (schedule) {
+            server.rdb_bgsave_scheduled = 1;
+            addReplyStatus(c,"Background saving scheduled");
+        } else {
+            addReplyError(c,
+                "An AOF log rewriting in progress: can't BGSAVE right now. "
+                "Use BGSAVE SCHEDULE in order to schedule a BGSAVE whenver "
+                "possible.");
+        }
     } else if (rdbSaveBackground(server.rdb_filename) == C_OK) {
         addReplyStatus(c,"Background saving started");
     } else {

src/redis-benchmark.c

@@ -65,6 +65,7 @@ static struct config {
     int randomkeys_keyspacelen;
     int keepalive;
     int pipeline;
+    int showerrors;
     long long start;
     long long totlatency;
     long long *latency;
@@ -212,6 +213,16 @@ static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
                     exit(1);
                 }
 
+                if (config.showerrors) {
+                    static time_t lasterr_time = 0;
+                    time_t now = time(NULL);
+                    redisReply *r = reply;
+                    if (r->type == REDIS_REPLY_ERROR && lasterr_time != now) {
+                        lasterr_time = now;
+                        printf("Error from server: %s\n", r->str);
+                    }
+                }
+
                 freeReplyObject(reply);
                 /* This is an OK for prefix commands such as auth and select.*/
                 if (c->prefix_pending > 0) {
@@ -227,7 +238,7 @@ static void readHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
                             c->randptr[j] -= c->prefixlen;
                         c->prefixlen = 0;
                     }
-                    continue;                
+                    continue;
                 }
 
                 if (config.requests_finished < config.requests)
@@ -518,6 +529,8 @@ int parseOptions(int argc, const char **argv) {
             config.loop = 1;
         } else if (!strcmp(argv[i],"-I")) {
             config.idlemode = 1;
+        } else if (!strcmp(argv[i],"-e")) {
+            config.showerrors = 1;
         } else if (!strcmp(argv[i],"-t")) {
             if (lastarg) goto invalid;
             /* We get the list of tests to run as a string in the form
@@ -569,6 +582,8 @@ usage:
 "  is executed. Default tests use this to hit random keys in the\n"
 "  specified range.\n"
 " -P <numreq>        Pipeline <numreq> requests. Default 1 (no pipeline).\n"
+" -e                 If server replies with errors, show them on stdout.\n"
+"                    (no more than 1 error per second is displayed)\n"
 " -q                 Quiet. Just show query/sec values\n"
 " --csv              Output in CSV format\n"
 " -l                 Loop. Run the tests forever\n"
@@ -649,6 +664,7 @@ int main(int argc, const char **argv) {
     config.keepalive = 1;
     config.datasize = 3;
     config.pipeline = 1;
+    config.showerrors = 0;
     config.randomkeys = 0;
     config.randomkeys_keyspacelen = 0;
     config.quiet = 0;

src/redis-check-rdb.c

@@ -1,6 +1,5 @@
 /*
- * Copyright (c) 2009-2012, Pieter Noordhuis <pcnoordhuis at gmail dot com>
- * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
+ * Copyright (c) 2016, Salvatore Sanfilippo <antirez at gmail dot com>
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
@@ -28,683 +27,319 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
-
 #include "server.h"
 #include "rdb.h"
-#include <stdlib.h>
-#include <stdio.h>
-#include <unistd.h>
-#include <fcntl.h>
-#include <sys/stat.h>
-#include <sys/mman.h>
-#include "lzf.h"
-#include "crc64.h"
 
-#define ERROR(...) { \
-    serverLog(LL_WARNING, __VA_ARGS__); \
-    exit(1); \
+#include <stdarg.h>
+
+void createSharedObjects(void);
+void rdbLoadProgressCallback(rio *r, const void *buf, size_t len);
+long long rdbLoadMillisecondTime(rio *rdb);
+int rdbCheckMode = 0;
+
+struct {
+    rio *rio;
+    robj *key;                      /* Current key we are reading. */
+    int key_type;                   /* Current key type if != -1. */
+    unsigned long keys;             /* Number of keys processed. */
+    unsigned long expires;          /* Number of keys with an expire. */
+    unsigned long already_expired;  /* Number of keys already expired. */
+    int doing;                      /* The state while reading the RDB. */
+    int error_set;                  /* True if error is populated. */
+    char error[1024];
+} rdbstate;
+
+/* At every loading step try to remember what we were about to do, so that
+ * we can log this information when an error is encountered. */
+#define RDB_CHECK_DOING_START 0
+#define RDB_CHECK_DOING_READ_TYPE 1
+#define RDB_CHECK_DOING_READ_EXPIRE 2
+#define RDB_CHECK_DOING_READ_KEY 3
+#define RDB_CHECK_DOING_READ_OBJECT_VALUE 4
+#define RDB_CHECK_DOING_CHECK_SUM 5
+#define RDB_CHECK_DOING_READ_LEN 6
+#define RDB_CHECK_DOING_READ_AUX 7
+
+char *rdb_check_doing_string[] = {
+    "start",
+    "read-type",
+    "read-expire",
+    "read-key",
+    "read-object-value",
+    "check-sum",
+    "read-len",
+    "read-aux"
+};
+
+char *rdb_type_string[] = {
+    "string",
+    "list-linked",
+    "set-hashtable",
+    "zset-v1",
+    "hash-hashtable",
+    "zset-v2",
+    "module-value",
+    "","",
+    "hash-zipmap",
+    "list-ziplist",
+    "set-intset",
+    "zset-ziplist",
+    "hash-ziplist",
+    "quicklist"
+};
+
+/* Show a few stats collected into 'rdbstate' */
+void rdbShowGenericInfo(void) {
+    printf("[info] %lu keys read\n", rdbstate.keys);
+    printf("[info] %lu expires\n", rdbstate.expires);
+    printf("[info] %lu already expired\n", rdbstate.already_expired);
 }
 
-/* data type to hold offset in file and size */
-typedef struct {
-    void *data;
-    size_t size;
-    size_t offset;
-} pos;
+/* Called on RDB errors. Provides details about the RDB and the offset
+ * we were when the error was detected. */
+void rdbCheckError(const char *fmt, ...) {
+    char msg[1024];
+    va_list ap;
 
-static unsigned char level = 0;
-static pos positions[16];
+    va_start(ap, fmt);
+    vsnprintf(msg, sizeof(msg), fmt, ap);
+    va_end(ap);
 
-#define CURR_OFFSET (positions[level].offset)
-
-/* Hold a stack of errors */
-typedef struct {
-    char error[16][1024];
-    size_t offset[16];
-    size_t level;
-} errors_t;
-static errors_t errors;
-
-#define SHIFT_ERROR(provided_offset, ...) { \
-    sprintf(errors.error[errors.level], __VA_ARGS__); \
-    errors.offset[errors.level] = provided_offset; \
-    errors.level++; \
+    printf("--- RDB ERROR DETECTED ---\n");
+    printf("[offset %llu] %s\n",
+        (unsigned long long) (rdbstate.rio ?
+            rdbstate.rio->processed_bytes : 0), msg);
+    printf("[additional info] While doing: %s\n",
+        rdb_check_doing_string[rdbstate.doing]);
+    if (rdbstate.key)
+        printf("[additional info] Reading key '%s'\n",
+            (char*)rdbstate.key->ptr);
+    if (rdbstate.key_type != -1)
+        printf("[additional info] Reading type %d (%s)\n",
+            rdbstate.key_type,
+            ((unsigned)rdbstate.key_type <
+             sizeof(rdb_type_string)/sizeof(char*)) ?
+                rdb_type_string[rdbstate.key_type] : "unknown");
+    rdbShowGenericInfo();
 }
 
-/* Data type to hold opcode with optional key name an success status */
-typedef struct {
-    char* key;
-    int type;
-    char success;
-} entry;
+/* Print informations during RDB checking. */
+void rdbCheckInfo(const char *fmt, ...) {
+    char msg[1024];
+    va_list ap;
 
-#define MAX_TYPES_NUM 256
-#define MAX_TYPE_NAME_LEN 16
-/* store string types for output */
-static char types[MAX_TYPES_NUM][MAX_TYPE_NAME_LEN];
+    va_start(ap, fmt);
+    vsnprintf(msg, sizeof(msg), fmt, ap);
+    va_end(ap);
 
-/* Return true if 't' is a valid object type. */
-static int rdbCheckType(unsigned char t) {
-    /* In case a new object type is added, update the following
-     * condition as necessary. */
-    return
-        (t >= RDB_TYPE_HASH_ZIPMAP && t <= RDB_TYPE_HASH_ZIPLIST) ||
-        t <= RDB_TYPE_HASH ||
-        t >= RDB_OPCODE_EXPIRETIME_MS;
+    printf("[offset %llu] %s\n",
+        (unsigned long long) (rdbstate.rio ?
+            rdbstate.rio->processed_bytes : 0), msg);
 }
 
-/* when number of bytes to read is negative, do a peek */
-static int readBytes(void *target, long num) {
-    char peek = (num < 0) ? 1 : 0;
-    num = (num < 0) ? -num : num;
+/* Used inside rdb.c in order to log specific errors happening inside
+ * the RDB loading internals. */
+void rdbCheckSetError(const char *fmt, ...) {
+    va_list ap;
 
-    pos p = positions[level];
-    if (p.offset + num > p.size) {
-        return 0;
-    } else {
-        memcpy(target, (void*)((size_t)p.data + p.offset), num);
-        if (!peek) positions[level].offset += num;
-    }
-    return 1;
+    va_start(ap, fmt);
+    vsnprintf(rdbstate.error, sizeof(rdbstate.error), fmt, ap);
+    va_end(ap);
+    rdbstate.error_set = 1;
 }
 
-int processHeader(void) {
-    char buf[10] = "_________";
-    int dump_version;
+/* During RDB check we setup a special signal handler for memory violations
+ * and similar conditions, so that we can log the offending part of the RDB
+ * if the crash is due to broken content. */
+void rdbCheckHandleCrash(int sig, siginfo_t *info, void *secret) {
+    UNUSED(sig);
+    UNUSED(info);
+    UNUSED(secret);
 
-    if (!readBytes(buf, 9)) {
-        ERROR("Cannot read header");
-    }
-
-    /* expect the first 5 bytes to equal REDIS */
-    if (memcmp(buf,"REDIS",5) != 0) {
-        ERROR("Wrong signature in header");
-    }
-
-    dump_version = (int)strtol(buf + 5, NULL, 10);
-    if (dump_version < 1 || dump_version > 6) {
-        ERROR("Unknown RDB format version: %d", dump_version);
-    }
-    return dump_version;
+    rdbCheckError("Server crash checking the specified RDB file!");
+    exit(1);
 }
 
-static int loadType(entry *e) {
-    uint32_t offset = CURR_OFFSET;
+void rdbCheckSetupSignals(void) {
+    struct sigaction act;
 
-    /* this byte needs to qualify as type */
-    unsigned char t;
-    if (readBytes(&t, 1)) {
-        if (rdbCheckType(t)) {
-            e->type = t;
-            return 1;
-        } else {
-            SHIFT_ERROR(offset, "Unknown type (0x%02x)", t);
-        }
-    } else {
-        SHIFT_ERROR(offset, "Could not read type");
-    }
-
-    /* failure */
-    return 0;
-}
-
-static int peekType() {
-    unsigned char t;
-    if (readBytes(&t, -1) && (rdbCheckType(t)))
-        return t;
-    return -1;
-}
-
-/* discard time, just consume the bytes */
-static int processTime(int type) {
-    uint32_t offset = CURR_OFFSET;
-    unsigned char t[8];
-    int timelen = (type == RDB_OPCODE_EXPIRETIME_MS) ? 8 : 4;
-
-    if (readBytes(t,timelen)) {
-        return 1;
-    } else {
-        SHIFT_ERROR(offset, "Could not read time");
-    }
-
-    /* failure */
-    return 0;
-}
-
-static uint32_t loadLength(int *isencoded) {
-    unsigned char buf[2];
-    uint32_t len;
-    int type;
-
-    if (isencoded) *isencoded = 0;
-    if (!readBytes(buf, 1)) return RDB_LENERR;
-    type = (buf[0] & 0xC0) >> 6;
-    if (type == RDB_6BITLEN) {
-        /* Read a 6 bit len */
-        return buf[0] & 0x3F;
-    } else if (type == RDB_ENCVAL) {
-        /* Read a 6 bit len encoding type */
-        if (isencoded) *isencoded = 1;
-        return buf[0] & 0x3F;
-    } else if (type == RDB_14BITLEN) {
-        /* Read a 14 bit len */
-        if (!readBytes(buf+1,1)) return RDB_LENERR;
-        return ((buf[0] & 0x3F) << 8) | buf[1];
-    } else {
-        /* Read a 32 bit len */
-        if (!readBytes(&len, 4)) return RDB_LENERR;
-        return (unsigned int)ntohl(len);
-    }
-}
-
-static char *loadIntegerObject(int enctype) {
-    uint32_t offset = CURR_OFFSET;
-    unsigned char enc[4];
-    long long val;
-
-    if (enctype == RDB_ENC_INT8) {
-        uint8_t v;
-        if (!readBytes(enc, 1)) return NULL;
-        v = enc[0];
-        val = (int8_t)v;
-    } else if (enctype == RDB_ENC_INT16) {
-        uint16_t v;
-        if (!readBytes(enc, 2)) return NULL;
-        v = enc[0]|(enc[1]<<8);
-        val = (int16_t)v;
-    } else if (enctype == RDB_ENC_INT32) {
-        uint32_t v;
-        if (!readBytes(enc, 4)) return NULL;
-        v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24);
-        val = (int32_t)v;
-    } else {
-        SHIFT_ERROR(offset, "Unknown integer encoding (0x%02x)", enctype);
-        return NULL;
-    }
-
-    /* convert val into string */
-    char *buf;
-    buf = zmalloc(sizeof(char) * 128);
-    sprintf(buf, "%lld", val);
-    return buf;
-}
-
-static char* loadLzfStringObject() {
-    unsigned int slen, clen;
-    char *c, *s;
-
-    if ((clen = loadLength(NULL)) == RDB_LENERR) return NULL;
-    if ((slen = loadLength(NULL)) == RDB_LENERR) return NULL;
-
-    c = zmalloc(clen);
-    if (!readBytes(c, clen)) {
-        zfree(c);
-        return NULL;
-    }
-
-    s = zmalloc(slen+1);
-    if (lzf_decompress(c,clen,s,slen) == 0) {
-        zfree(c); zfree(s);
-        return NULL;
-    }
-
-    zfree(c);
-    return s;
-}
-
-/* returns NULL when not processable, char* when valid */
-static char* loadStringObject() {
-    uint32_t offset = CURR_OFFSET;
-    int isencoded;
-    uint32_t len;
-
-    len = loadLength(&isencoded);
-    if (isencoded) {
-        switch(len) {
-        case RDB_ENC_INT8:
-        case RDB_ENC_INT16:
-        case RDB_ENC_INT32:
-            return loadIntegerObject(len);
-        case RDB_ENC_LZF:
-            return loadLzfStringObject();
-        default:
-            /* unknown encoding */
-            SHIFT_ERROR(offset, "Unknown string encoding (0x%02x)", len);
-            return NULL;
-        }
-    }
-
-    if (len == RDB_LENERR) return NULL;
-
-    char *buf = zmalloc(sizeof(char) * (len+1));
-    if (buf == NULL) return NULL;
-    buf[len] = '\0';
-    if (!readBytes(buf, len)) {
-        zfree(buf);
-        return NULL;
-    }
-    return buf;
-}
-
-static int processStringObject(char** store) {
-    unsigned long offset = CURR_OFFSET;
-    char *key = loadStringObject();
-    if (key == NULL) {
-        SHIFT_ERROR(offset, "Error reading string object");
-        zfree(key);
-        return 0;
-    }
-
-    if (store != NULL) {
-        *store = key;
-    } else {
-        zfree(key);
-    }
-    return 1;
-}
-
-static double* loadDoubleValue() {
-    char buf[256];
-    unsigned char len;
-    double* val;
-
-    if (!readBytes(&len,1)) return NULL;
-
-    val = zmalloc(sizeof(double));
-    switch(len) {
-    case 255: *val = R_NegInf;  return val;
-    case 254: *val = R_PosInf;  return val;
-    case 253: *val = R_Nan;     return val;
-    default:
-        if (!readBytes(buf, len)) {
-            zfree(val);
-            return NULL;
-        }
-        buf[len] = '\0';
-        sscanf(buf, "%lg", val);
-        return val;
-    }
-}
-
-static int processDoubleValue(double** store) {
-    unsigned long offset = CURR_OFFSET;
-    double *val = loadDoubleValue();
-    if (val == NULL) {
-        SHIFT_ERROR(offset, "Error reading double value");
-        zfree(val);
-        return 0;
-    }
-
-    if (store != NULL) {
-        *store = val;
-    } else {
-        zfree(val);
-    }
-    return 1;
-}
-
-static int loadPair(entry *e) {
-    uint32_t offset = CURR_OFFSET;
-    uint32_t i;
-
-    /* read key first */
-    char *key;
-    if (processStringObject(&key)) {
-        e->key = key;
-    } else {
-        SHIFT_ERROR(offset, "Error reading entry key");
-        return 0;
-    }
-
-    uint32_t length = 0;
-    if (e->type == RDB_TYPE_LIST ||
-        e->type == RDB_TYPE_SET  ||
-        e->type == RDB_TYPE_ZSET ||
-        e->type == RDB_TYPE_HASH) {
-        if ((length = loadLength(NULL)) == RDB_LENERR) {
-            SHIFT_ERROR(offset, "Error reading %s length", types[e->type]);
-            return 0;
-        }
-    }
-
-    switch(e->type) {
-    case RDB_TYPE_STRING:
-    case RDB_TYPE_HASH_ZIPMAP:
-    case RDB_TYPE_LIST_ZIPLIST:
-    case RDB_TYPE_SET_INTSET:
-    case RDB_TYPE_ZSET_ZIPLIST:
-    case RDB_TYPE_HASH_ZIPLIST:
-        if (!processStringObject(NULL)) {
-            SHIFT_ERROR(offset, "Error reading entry value");
-            return 0;
-        }
-    break;
-    case RDB_TYPE_LIST:
-    case RDB_TYPE_SET:
-        for (i = 0; i < length; i++) {
-            offset = CURR_OFFSET;
-            if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element at index %d (length: %d)", i, length);
-                return 0;
-            }
-        }
-    break;
-    case RDB_TYPE_ZSET:
-        for (i = 0; i < length; i++) {
-            offset = CURR_OFFSET;
-            if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element key at index %d (length: %d)", i, length);
-                return 0;
-            }
-            offset = CURR_OFFSET;
-            if (!processDoubleValue(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element value at index %d (length: %d)", i, length);
-                return 0;
-            }
-        }
-    break;
-    case RDB_TYPE_HASH:
-        for (i = 0; i < length; i++) {
-            offset = CURR_OFFSET;
-            if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element key at index %d (length: %d)", i, length);
-                return 0;
-            }
-            offset = CURR_OFFSET;
-            if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element value at index %d (length: %d)", i, length);
-                return 0;
-            }
-        }
-    break;
-    default:
-        SHIFT_ERROR(offset, "Type not implemented");
-        return 0;
-    }
-    /* because we're done, we assume success */
-    e->success = 1;
-    return 1;
-}
-
-static entry loadEntry() {
-    entry e = { NULL, -1, 0 };
-    uint32_t length, offset[4];
-
-    /* reset error container */
-    errors.level = 0;
-
-    offset[0] = CURR_OFFSET;
-    if (!loadType(&e)) {
-        return e;
-    }
-
-    offset[1] = CURR_OFFSET;
-    if (e.type == RDB_OPCODE_SELECTDB) {
-        if ((length = loadLength(NULL)) == RDB_LENERR) {
-            SHIFT_ERROR(offset[1], "Error reading database number");
-            return e;
-        }
-        if (length > 63) {
-            SHIFT_ERROR(offset[1], "Database number out of range (%d)", length);
-            return e;
-        }
-    } else if (e.type == RDB_OPCODE_EOF) {
-        if (positions[level].offset < positions[level].size) {
-            SHIFT_ERROR(offset[0], "Unexpected EOF");
-        } else {
-            e.success = 1;
-        }
-        return e;
-    } else {
-        /* optionally consume expire */
-        if (e.type == RDB_OPCODE_EXPIRETIME ||
-            e.type == RDB_OPCODE_EXPIRETIME_MS) {
-            if (!processTime(e.type)) return e;
-            if (!loadType(&e)) return e;
-        }
-
-        offset[1] = CURR_OFFSET;
-        if (!loadPair(&e)) {
-            SHIFT_ERROR(offset[1], "Error for type %s", types[e.type]);
-            return e;
-        }
-    }
-
-    /* all entries are followed by a valid type:
-     * e.g. a new entry, SELECTDB, EXPIRE, EOF */
-    offset[2] = CURR_OFFSET;
-    if (peekType() == -1) {
-        SHIFT_ERROR(offset[2], "Followed by invalid type");
-        SHIFT_ERROR(offset[0], "Error for type %s", types[e.type]);
-        e.success = 0;
-    } else {
-        e.success = 1;
-    }
-
-    return e;
-}
-
-static void printCentered(int indent, int width, char* body) {
-    char head[256], tail[256];
-    memset(head, '\0', 256);
-    memset(tail, '\0', 256);
-
-    memset(head, '=', indent);
-    memset(tail, '=', width - 2 - indent - strlen(body));
-    serverLog(LL_WARNING, "%s %s %s", head, body, tail);
-}
-
-static void printValid(uint64_t ops, uint64_t bytes) {
-    char body[80];
-    sprintf(body, "Processed %llu valid opcodes (in %llu bytes)",
-        (unsigned long long) ops, (unsigned long long) bytes);
-    printCentered(4, 80, body);
-}
-
-static void printSkipped(uint64_t bytes, uint64_t offset) {
-    char body[80];
-    sprintf(body, "Skipped %llu bytes (resuming at 0x%08llx)",
-        (unsigned long long) bytes, (unsigned long long) offset);
-    printCentered(4, 80, body);
-}
-
-static void printErrorStack(entry *e) {
-    unsigned int i;
-    char body[64];
-
-    if (e->type == -1) {
-        sprintf(body, "Error trace");
-    } else if (e->type >= 253) {
-        sprintf(body, "Error trace (%s)", types[e->type]);
-    } else if (!e->key) {
-        sprintf(body, "Error trace (%s: (unknown))", types[e->type]);
-    } else {
-        char tmp[41];
-        strncpy(tmp, e->key, 40);
-
-        /* display truncation at the last 3 chars */
-        if (strlen(e->key) > 40) {
-            memset(&tmp[37], '.', 3);
-        }
-
-        /* display unprintable characters as ? */
-        for (i = 0; i < strlen(tmp); i++) {
-            if (tmp[i] <= 32) tmp[i] = '?';
-        }
-        sprintf(body, "Error trace (%s: %s)", types[e->type], tmp);
-    }
-
-    printCentered(4, 80, body);
-
-    /* display error stack */
-    for (i = 0; i < errors.level; i++) {
-        serverLog(LL_WARNING, "0x%08lx - %s",
-            (unsigned long) errors.offset[i], errors.error[i]);
-    }
-}
-
-void process(void) {
-    uint64_t num_errors = 0, num_valid_ops = 0, num_valid_bytes = 0;
-    entry entry = { NULL, -1, 0 };
-    int dump_version = processHeader();
-
-    /* Exclude the final checksum for RDB >= 5. Will be checked at the end. */
-    if (dump_version >= 5) {
-        if (positions[0].size < 8) {
-            serverLog(LL_WARNING, "RDB version >= 5 but no room for checksum.");
-            exit(1);
-        }
-        positions[0].size -= 8;
-    }
-
-    level = 1;
-    while(positions[0].offset < positions[0].size) {
-        positions[1] = positions[0];
-
-        entry = loadEntry();
-        if (!entry.success) {
-            printValid(num_valid_ops, num_valid_bytes);
-            printErrorStack(&entry);
-            num_errors++;
-            num_valid_ops = 0;
-            num_valid_bytes = 0;
-
-            /* search for next valid entry */
-            uint64_t offset = positions[0].offset + 1;
-            int i = 0;
-
-            while (!entry.success && offset < positions[0].size) {
-                positions[1].offset = offset;
-
-                /* find 3 consecutive valid entries */
-                for (i = 0; i < 3; i++) {
-                    entry = loadEntry();
-                    if (!entry.success) break;
-                }
-                /* check if we found 3 consecutive valid entries */
-                if (i < 3) {
-                    offset++;
-                }
-            }
-
-            /* print how many bytes we have skipped to find a new valid opcode */
-            if (offset < positions[0].size) {
-                printSkipped(offset - positions[0].offset, offset);
-            }
-
-            positions[0].offset = offset;
-        } else {
-            num_valid_ops++;
-            num_valid_bytes += positions[1].offset - positions[0].offset;
-
-            /* advance position */
-            positions[0] = positions[1];
-        }
-        zfree(entry.key);
-    }
-
-    /* because there is another potential error,
-     * print how many valid ops we have processed */
-    printValid(num_valid_ops, num_valid_bytes);
-
-    /* expect an eof */
-    if (entry.type != RDB_OPCODE_EOF) {
-        /* last byte should be EOF, add error */
-        errors.level = 0;
-        SHIFT_ERROR(positions[0].offset, "Expected EOF, got %s", types[entry.type]);
-
-        /* this is an EOF error so reset type */
-        entry.type = -1;
-        printErrorStack(&entry);
-
-        num_errors++;
-    }
-
-    /* Verify checksum */
-    if (dump_version >= 5) {
-        uint64_t crc = crc64(0,positions[0].data,positions[0].size);
-        uint64_t crc2;
-        unsigned char *p = (unsigned char*)positions[0].data+positions[0].size;
-        crc2 = ((uint64_t)p[0] << 0) |
-               ((uint64_t)p[1] << 8) |
-               ((uint64_t)p[2] << 16) |
-               ((uint64_t)p[3] << 24) |
-               ((uint64_t)p[4] << 32) |
-               ((uint64_t)p[5] << 40) |
-               ((uint64_t)p[6] << 48) |
-               ((uint64_t)p[7] << 56);
-        if (crc != crc2) {
-            SHIFT_ERROR(positions[0].offset, "RDB CRC64 does not match.");
-        } else {
-            serverLog(LL_WARNING, "CRC64 checksum is OK");
-        }
-    }
-
-    /* print summary on errors */
-    if (num_errors) {
-        serverLog(LL_WARNING, "Total unprocessable opcodes: %llu",
-            (unsigned long long) num_errors);
-    }
+    sigemptyset(&act.sa_mask);
+    act.sa_flags = SA_NODEFER | SA_RESETHAND | SA_SIGINFO;
+    act.sa_sigaction = rdbCheckHandleCrash;
+    sigaction(SIGSEGV, &act, NULL);
+    sigaction(SIGBUS, &act, NULL);
+    sigaction(SIGFPE, &act, NULL);
+    sigaction(SIGILL, &act, NULL);
 }
 
+/* Check the specified RDB file. */
 int redis_check_rdb(char *rdbfilename) {
-    int fd;
-    off_t size;
-    struct stat stat;
-    void *data;
+    uint64_t dbid;
+    int type, rdbver;
+    char buf[1024];
+    long long expiretime, now = mstime();
+    FILE *fp;
+    rio rdb;
 
-    fd = open(rdbfilename, O_RDONLY);
-    if (fd < 1) {
-        ERROR("Cannot open file: %s", rdbfilename);
+    if ((fp = fopen(rdbfilename,"r")) == NULL) return C_ERR;
+
+    rioInitWithFile(&rdb,fp);
+    rdbstate.rio = &rdb;
+    rdb.update_cksum = rdbLoadProgressCallback;
+    if (rioRead(&rdb,buf,9) == 0) goto eoferr;
+    buf[9] = '\0';
+    if (memcmp(buf,"REDIS",5) != 0) {
+        rdbCheckError("Wrong signature trying to load DB from file");
+        return 1;
     }
-    if (fstat(fd, &stat) == -1) {
-        ERROR("Cannot stat: %s", rdbfilename);
-    } else {
-        size = stat.st_size;
+    rdbver = atoi(buf+5);
+    if (rdbver < 1 || rdbver > RDB_VERSION) {
+        rdbCheckError("Can't handle RDB format version %d",rdbver);
+        return 1;
     }
 
-    if (sizeof(size_t) == sizeof(int32_t) && size >= INT_MAX) {
-        ERROR("Cannot check dump files >2GB on a 32-bit platform");
+    startLoading(fp);
+    while(1) {
+        robj *key, *val;
+        expiretime = -1;
+
+        /* Read type. */
+        rdbstate.doing = RDB_CHECK_DOING_READ_TYPE;
+        if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
+
+        /* Handle special types. */
+        if (type == RDB_OPCODE_EXPIRETIME) {
+            rdbstate.doing = RDB_CHECK_DOING_READ_EXPIRE;
+            /* EXPIRETIME: load an expire associated with the next key
+             * to load. Note that after loading an expire we need to
+             * load the actual type, and continue. */
+            if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr;
+            /* We read the time so we need to read the object type again. */
+            rdbstate.doing = RDB_CHECK_DOING_READ_TYPE;
+            if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
+            /* the EXPIRETIME opcode specifies time in seconds, so convert
+             * into milliseconds. */
+            expiretime *= 1000;
+        } else if (type == RDB_OPCODE_EXPIRETIME_MS) {
+            /* EXPIRETIME_MS: milliseconds precision expire times introduced
+             * with RDB v3. Like EXPIRETIME but no with more precision. */
+            rdbstate.doing = RDB_CHECK_DOING_READ_EXPIRE;
+            if ((expiretime = rdbLoadMillisecondTime(&rdb)) == -1) goto eoferr;
+            /* We read the time so we need to read the object type again. */
+            rdbstate.doing = RDB_CHECK_DOING_READ_TYPE;
+            if ((type = rdbLoadType(&rdb)) == -1) goto eoferr;
+        } else if (type == RDB_OPCODE_EOF) {
+            /* EOF: End of file, exit the main loop. */
+            break;
+        } else if (type == RDB_OPCODE_SELECTDB) {
+            /* SELECTDB: Select the specified database. */
+            rdbstate.doing = RDB_CHECK_DOING_READ_LEN;
+            if ((dbid = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
+                goto eoferr;
+            rdbCheckInfo("Selecting DB ID %d", dbid);
+            continue; /* Read type again. */
+        } else if (type == RDB_OPCODE_RESIZEDB) {
+            /* RESIZEDB: Hint about the size of the keys in the currently
+             * selected data base, in order to avoid useless rehashing. */
+            uint64_t db_size, expires_size;
+            rdbstate.doing = RDB_CHECK_DOING_READ_LEN;
+            if ((db_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
+                goto eoferr;
+            if ((expires_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
+                goto eoferr;
+            continue; /* Read type again. */
+        } else if (type == RDB_OPCODE_AUX) {
+            /* AUX: generic string-string fields. Use to add state to RDB
+             * which is backward compatible. Implementations of RDB loading
+             * are requierd to skip AUX fields they don't understand.
+             *
+             * An AUX field is composed of two strings: key and value. */
+            robj *auxkey, *auxval;
+            rdbstate.doing = RDB_CHECK_DOING_READ_AUX;
+            if ((auxkey = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
+            if ((auxval = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
+
+            rdbCheckInfo("AUX FIELD %s = '%s'",
+                (char*)auxkey->ptr, (char*)auxval->ptr);
+            decrRefCount(auxkey);
+            decrRefCount(auxval);
+            continue; /* Read type again. */
+        } else {
+            if (!rdbIsObjectType(type)) {
+                rdbCheckError("Invalid object type: %d", type);
+                return 1;
+            }
+            rdbstate.key_type = type;
+        }
+
+        /* Read key */
+        rdbstate.doing = RDB_CHECK_DOING_READ_KEY;
+        if ((key = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
+        rdbstate.key = key;
+        rdbstate.keys++;
+        /* Read value */
+        rdbstate.doing = RDB_CHECK_DOING_READ_OBJECT_VALUE;
+        if ((val = rdbLoadObject(type,&rdb)) == NULL) goto eoferr;
+        /* Check if the key already expired. This function is used when loading
+         * an RDB file from disk, either at startup, or when an RDB was
+         * received from the master. In the latter case, the master is
+         * responsible for key expiry. If we would expire keys here, the
+         * snapshot taken by the master may not be reflected on the slave. */
+        if (server.masterhost == NULL && expiretime != -1 && expiretime < now)
+            rdbstate.already_expired++;
+        if (expiretime != -1) rdbstate.expires++;
+        rdbstate.key = NULL;
+        decrRefCount(key);
+        decrRefCount(val);
+        rdbstate.key_type = -1;
+    }
+    /* Verify the checksum if RDB version is >= 5 */
+    if (rdbver >= 5 && server.rdb_checksum) {
+        uint64_t cksum, expected = rdb.cksum;
+
+        rdbstate.doing = RDB_CHECK_DOING_CHECK_SUM;
+        if (rioRead(&rdb,&cksum,8) == 0) goto eoferr;
+        memrev64ifbe(&cksum);
+        if (cksum == 0) {
+            rdbCheckInfo("RDB file was saved with checksum disabled: no check performed.");
+        } else if (cksum != expected) {
+            rdbCheckError("RDB CRC error");
+        } else {
+            rdbCheckInfo("Checksum OK");
+        }
     }
 
-    data = mmap(NULL, size, PROT_READ, MAP_SHARED, fd, 0);
-    if (data == MAP_FAILED) {
-        ERROR("Cannot mmap: %s", rdbfilename);
-    }
-
-    /* Initialize static vars */
-    positions[0].data = data;
-    positions[0].size = size;
-    positions[0].offset = 0;
-    errors.level = 0;
-
-    /* Object types */
-    sprintf(types[RDB_TYPE_STRING], "STRING");
-    sprintf(types[RDB_TYPE_LIST], "LIST");
-    sprintf(types[RDB_TYPE_SET], "SET");
-    sprintf(types[RDB_TYPE_ZSET], "ZSET");
-    sprintf(types[RDB_TYPE_HASH], "HASH");
-
-    /* Object types only used for dumping to disk */
-    sprintf(types[RDB_OPCODE_EXPIRETIME], "EXPIRETIME");
-    sprintf(types[RDB_OPCODE_SELECTDB], "SELECTDB");
-    sprintf(types[RDB_OPCODE_EOF], "EOF");
-
-    process();
-
-    munmap(data, size);
-    close(fd);
+    fclose(fp);
     return 0;
+
+eoferr: /* unexpected end of file is handled here with a fatal exit */
+    if (rdbstate.error_set) {
+        rdbCheckError(rdbstate.error);
+    } else {
+        rdbCheckError("Unexpected EOF reading RDB file");
+    }
+    return 1;
 }
 
 /* RDB check main: called form redis.c when Redis is executed with the
- * redis-check-rdb alias. */
-int redis_check_rdb_main(char **argv, int argc) {
+ * redis-check-rdb alias.
+ *
+ * The function never returns, but exits with the status code according
+ * to success (RDB is sane) or error (RDB is corrupted). */
+int redis_check_rdb_main(int argc, char **argv) {
     if (argc != 2) {
         fprintf(stderr, "Usage: %s <rdb-file-name>\n", argv[0]);
         exit(1);
     }
-    serverLog(LL_WARNING, "Checking RDB file %s", argv[1]);
-    exit(redis_check_rdb(argv[1]));
-    return 0;
+    createSharedObjects(); /* Needed for loading. */
+    server.loading_process_events_interval_bytes = 0;
+    rdbCheckMode = 1;
+    rdbCheckInfo("Checking RDB file %s", argv[1]);
+    rdbCheckSetupSignals();
+    int retval = redis_check_rdb(argv[1]);
+    if (retval == 0) {
+        rdbCheckInfo("\\o/ RDB looks OK! \\o/");
+        rdbShowGenericInfo();
+    }
+    exit(retval);
 }

src/redis-cli.c

@@ -161,7 +161,7 @@ static void cliRefreshPrompt(void) {
         len = anetFormatAddr(config.prompt, sizeof(config.prompt),
                            config.hostip, config.hostport);
     /* Add [dbnum] if needed */
-    if (config.dbnum != 0 && config.last_cmd_type != REDIS_REPLY_ERROR)
+    if (config.dbnum != 0)
         len += snprintf(config.prompt+len,sizeof(config.prompt)-len,"[%d]",
             config.dbnum);
     snprintf(config.prompt+len,sizeof(config.prompt)-len,"> ");
@@ -916,7 +916,7 @@ static int cliSendCommand(int argc, char **argv, int repeat) {
             return REDIS_ERR;
         } else {
             /* Store database number when SELECT was successfully executed. */
-            if (!strcasecmp(command,"select") && argc == 2) {
+            if (!strcasecmp(command,"select") && argc == 2 && config.last_cmd_type != REDIS_REPLY_ERROR) {
                 config.dbnum = atoi(argv[1]);
                 cliRefreshPrompt();
             } else if (!strcasecmp(command,"auth") && argc == 2) {

src/redis-trib.rb

@@ -1305,6 +1305,11 @@ class RedisTrib
         sleep 1
         wait_cluster_join
         flush_nodes_config # Useful for the replicas
+        # Reset the node information, so that when the
+        # final summary is listed in check_cluster about the newly created cluster
+        # all the nodes would get properly listed as slaves or masters
+        reset_nodes
+        load_cluster_info_from_node(argv[0])
         check_cluster
     end
 

src/replication.c

@@ -47,7 +47,7 @@ int cancelReplicationHandshake(void);
 
 /* Return the pointer to a string representing the slave ip:listening_port
  * pair. Mostly useful for logging, since we want to log a slave using its
- * IP address and it's listening port which is more clear for the user, for
+ * IP address and its listening port which is more clear for the user, for
  * example: "Closing connection with slave 10.1.2.3:6380". */
 char *replicationGetSlaveName(client *c) {
     static char buf[NET_PEER_ID_LEN];
@@ -55,7 +55,12 @@ char *replicationGetSlaveName(client *c) {
 
     ip[0] = '\0';
     buf[0] = '\0';
-    if (anetPeerToString(c->fd,ip,sizeof(ip),NULL) != -1) {
+    if (c->slave_ip[0] != '\0' ||
+        anetPeerToString(c->fd,ip,sizeof(ip),NULL) != -1)
+    {
+        /* Note that the 'ip' buffer is always larger than 'c->slave_ip' */
+        if (c->slave_ip[0] != '\0') memcpy(ip,c->slave_ip,sizeof(c->slave_ip));
+
         if (c->slave_listening_port)
             anetFormatAddr(buf,sizeof(buf),ip,c->slave_listening_port);
         else
@@ -647,7 +652,7 @@ void syncCommand(client *c) {
         } else {
             /* No way, we need to wait for the next BGSAVE in order to
              * register differences. */
-            serverLog(LL_NOTICE,"Waiting for next BGSAVE for SYNC");
+            serverLog(LL_NOTICE,"Can't attach the slave to the current BGSAVE. Waiting for next BGSAVE for SYNC");
         }
 
     /* CASE 2: BGSAVE is in progress, with socket target. */
@@ -657,7 +662,7 @@ void syncCommand(client *c) {
         /* 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. */
-        serverLog(LL_NOTICE,"Waiting for next BGSAVE for SYNC");
+        serverLog(LL_NOTICE,"Current BGSAVE has socket target. Waiting for next BGSAVE for SYNC");
 
     /* CASE 3: There is no BGSAVE is progress. */
     } else {
@@ -666,12 +671,18 @@ void syncCommand(client *c) {
              * replicationCron() since we want to delay its start a
              * few seconds to wait for more slaves to arrive. */
             if (server.repl_diskless_sync_delay)
-                serverLog(LL_NOTICE,"Delay next BGSAVE for SYNC");
+                serverLog(LL_NOTICE,"Delay next BGSAVE for diskless SYNC");
         } else {
             /* 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) != C_OK) return;
+            if (server.aof_child_pid == -1) {
+                startBgsaveForReplication(c->slave_capa);
+            } else {
+                serverLog(LL_NOTICE,
+                    "No BGSAVE in progress, but an AOF rewrite is active. "
+                    "BGSAVE for replication delayed");
+            }
         }
     }
 
@@ -711,6 +722,15 @@ void replconfCommand(client *c) {
                     &port,NULL) != C_OK))
                 return;
             c->slave_listening_port = port;
+        } else if (!strcasecmp(c->argv[j]->ptr,"ip-address")) {
+            sds ip = c->argv[j+1]->ptr;
+            if (sdslen(ip) < sizeof(c->slave_ip)) {
+                memcpy(c->slave_ip,ip,sdslen(ip)+1);
+            } else {
+                addReplyErrorFormat(c,"REPLCONF ip-address provided by "
+                    "slave instance is too long: %zd bytes", sdslen(ip));
+                return;
+            }
         } else if (!strcasecmp(c->argv[j]->ptr,"capa")) {
             /* Ignore capabilities not understood by this master. */
             if (!strcasecmp(c->argv[j+1]->ptr,"eof"))
@@ -1453,7 +1473,8 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
     /* Set the slave port, so that Master's INFO command can list the
      * slave listening port correctly. */
     if (server.repl_state == REPL_STATE_SEND_PORT) {
-        sds port = sdsfromlonglong(server.port);
+        sds port = sdsfromlonglong(server.slave_announce_port ?
+            server.slave_announce_port : server.port);
         err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF",
                 "listening-port",port, NULL);
         sdsfree(port);
@@ -1473,6 +1494,37 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
                                 "REPLCONF listening-port: %s", err);
         }
         sdsfree(err);
+        server.repl_state = REPL_STATE_SEND_IP;
+    }
+
+    /* Skip REPLCONF ip-address if there is no slave-announce-ip option set. */
+    if (server.repl_state == REPL_STATE_SEND_IP &&
+        server.slave_announce_ip == NULL)
+    {
+            server.repl_state = REPL_STATE_SEND_CAPA;
+    }
+
+    /* Set the slave ip, so that Master's INFO command can list the
+     * slave IP address port correctly in case of port forwarding or NAT. */
+    if (server.repl_state == REPL_STATE_SEND_IP) {
+        err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"REPLCONF",
+                "ip-address",server.slave_announce_ip, NULL);
+        if (err) goto write_error;
+        sdsfree(err);
+        server.repl_state = REPL_STATE_RECEIVE_IP;
+        return;
+    }
+
+    /* Receive REPLCONF ip-address reply. */
+    if (server.repl_state == REPL_STATE_RECEIVE_IP) {
+        err = sendSynchronousCommand(SYNC_CMD_READ,fd,NULL);
+        /* Ignore the error if any, not all the Redis versions support
+         * REPLCONF listening-port. */
+        if (err[0] == '-') {
+            serverLog(LL_NOTICE,"(Non critical) Master does not understand "
+                                "REPLCONF ip-address: %s", err);
+        }
+        sdsfree(err);
         server.repl_state = REPL_STATE_SEND_CAPA;
     }
 
@@ -1778,12 +1830,16 @@ void roleCommand(client *c) {
         listRewind(server.slaves,&li);
         while((ln = listNext(&li))) {
             client *slave = ln->value;
-            char ip[NET_IP_STR_LEN];
+            char ip[NET_IP_STR_LEN], *slaveip = slave->slave_ip;
 
-            if (anetPeerToString(slave->fd,ip,sizeof(ip),NULL) == -1) continue;
+            if (slaveip[0] == '\0') {
+                if (anetPeerToString(slave->fd,ip,sizeof(ip),NULL) == -1)
+                    continue;
+                slaveip = ip;
+            }
             if (slave->replstate != SLAVE_STATE_ONLINE) continue;
             addReplyMultiBulkLen(c,3);
-            addReplyBulkCString(c,ip);
+            addReplyBulkCString(c,slaveip);
             addReplyBulkLongLong(c,slave->slave_listening_port);
             addReplyBulkLongLong(c,slave->repl_ack_off);
             slaves++;
@@ -2304,13 +2360,12 @@ void replicationCron(void) {
         replicationScriptCacheFlush();
     }
 
-    /* If we are using diskless replication and there are slaves waiting
-     * in WAIT_BGSAVE_START state, check if enough seconds elapsed and
-     * start a BGSAVE.
+    /* Start a BGSAVE good for replication if we have slaves in
+     * WAIT_BGSAVE_START state.
      *
-     * This code is also useful to trigger a BGSAVE if the diskless
-     * replication was turned off with CONFIG SET, while there were already
-     * slaves in WAIT_BGSAVE_START state. */
+     * In case of diskless replication, we make sure to wait the specified
+     * number of seconds (according to configuration) so that other slaves
+     * have the time to arrive before we start streaming. */
     if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) {
         time_t idle, max_idle = 0;
         int slaves_waiting = 0;
@@ -2330,9 +2385,13 @@ void replicationCron(void) {
             }
         }
 
-        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 (slaves_waiting &&
+            (!server.repl_diskless_sync ||
+             max_idle > server.repl_diskless_sync_delay))
+        {
+            /* Start the BGSAVE. The called function may start a
+             * BGSAVE with socket target or disk target depending on the
+             * configuration and slaves capabilities. */
             startBgsaveForReplication(mincapa);
         }
     }

src/sentinel.c

@@ -1062,11 +1062,18 @@ int sentinelUpdateSentinelAddressInAllMasters(sentinelRedisInstance *ri) {
         sentinelRedisInstance *master = dictGetVal(de), *match;
         match = getSentinelRedisInstanceByAddrAndRunID(master->sentinels,
                                                        NULL,0,ri->runid);
-        if (match->link->disconnected == 0) {
+        /* If there is no match, this master does not know about this
+         * Sentinel, try with the next one. */
+        if (match == NULL) continue;
+
+        /* Disconnect the old links if connected. */
+        if (match->link->cc != NULL)
             instanceLinkCloseConnection(match->link,match->link->cc);
+        if (match->link->pc != NULL)
             instanceLinkCloseConnection(match->link,match->link->pc);
-        }
+
         if (match == ri) continue; /* Address already updated for it. */
+
         /* Update the address of the matching Sentinel by copying the address
          * of the Sentinel object that received the address update. */
         releaseSentinelAddr(match->addr);
@@ -2572,9 +2579,15 @@ void sentinelSendPeriodicCommands(sentinelRedisInstance *ri) {
     /* If this is a slave of a master in O_DOWN condition we start sending
      * it INFO every second, instead of the usual SENTINEL_INFO_PERIOD
      * period. In this state we want to closely monitor slaves in case they
-     * are turned into masters by another Sentinel, or by the sysadmin. */
+     * are turned into masters by another Sentinel, or by the sysadmin.
+     *
+     * Similarly we monitor the INFO output more often if the slave reports
+     * to be disconnected from the master, so that we can have a fresh
+     * disconnection time figure. */
     if ((ri->flags & SRI_SLAVE) &&
-        (ri->master->flags & (SRI_O_DOWN|SRI_FAILOVER_IN_PROGRESS))) {
+        ((ri->master->flags & (SRI_O_DOWN|SRI_FAILOVER_IN_PROGRESS)) ||
+         (ri->master_link_down_time != 0)))
+    {
         info_period = 1000;
     } else {
         info_period = SENTINEL_INFO_PERIOD;

src/server.c

@@ -231,7 +231,7 @@ struct redisCommand redisCommandTable[] = {
     {"ping",pingCommand,-1,"tF",0,NULL,0,0,0,0,0},
     {"echo",echoCommand,2,"F",0,NULL,0,0,0,0,0},
     {"save",saveCommand,1,"as",0,NULL,0,0,0,0,0},
-    {"bgsave",bgsaveCommand,1,"a",0,NULL,0,0,0,0,0},
+    {"bgsave",bgsaveCommand,-1,"a",0,NULL,0,0,0,0,0},
     {"bgrewriteaof",bgrewriteaofCommand,1,"a",0,NULL,0,0,0,0,0},
     {"shutdown",shutdownCommand,-1,"alt",0,NULL,0,0,0,0,0},
     {"lastsave",lastsaveCommand,1,"RF",0,NULL,0,0,0,0,0},
@@ -1267,8 +1267,8 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     /* Clear the paused clients flag if needed. */
     clientsArePaused(); /* Don't check return value, just use the side effect. */
 
-    /* Replication cron function -- used to reconnect to master and
-     * to detect transfer failures. */
+    /* Replication cron function -- used to reconnect to master,
+     * detect transfer failures, start background RDB transfers and so forth. */
     run_with_period(1000) replicationCron();
 
     /* Run the Redis Cluster cron. */
@@ -1286,6 +1286,22 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
         migrateCloseTimedoutSockets();
     }
 
+    /* Start a scheduled BGSAVE if the corresponding flag is set. This is
+     * useful when we are forced to postpone a BGSAVE because an AOF
+     * rewrite is in progress.
+     *
+     * Note: this code must be after the replicationCron() call above so
+     * make sure when refactoring this file to keep this order. This is useful
+     * because we want to give priority to RDB savings for replication. */
+    if (server.rdb_child_pid == -1 && server.aof_child_pid == -1 &&
+        server.rdb_bgsave_scheduled &&
+        (server.unixtime-server.lastbgsave_try > CONFIG_BGSAVE_RETRY_DELAY ||
+         server.lastbgsave_status == C_OK))
+    {
+        if (rdbSaveBackground(server.rdb_filename) == C_OK)
+            server.rdb_bgsave_scheduled = 0;
+    }
+
     server.cronloops++;
     return 1000/server.hz;
 }
@@ -1538,6 +1554,8 @@ void initServerConfig(void) {
     server.repl_diskless_sync = CONFIG_DEFAULT_REPL_DISKLESS_SYNC;
     server.repl_diskless_sync_delay = CONFIG_DEFAULT_REPL_DISKLESS_SYNC_DELAY;
     server.slave_priority = CONFIG_DEFAULT_SLAVE_PRIORITY;
+    server.slave_announce_ip = CONFIG_DEFAULT_SLAVE_ANNOUNCE_IP;
+    server.slave_announce_port = CONFIG_DEFAULT_SLAVE_ANNOUNCE_PORT;
     server.master_repl_offset = 0;
 
     /* Replication partial resync backlog */
@@ -1758,6 +1776,7 @@ int listenToPort(int port, int *fds, int *count) {
     if (server.bindaddr_count == 0) server.bindaddr[0] = NULL;
     for (j = 0; j < server.bindaddr_count || j == 0; j++) {
         if (server.bindaddr[j] == NULL) {
+            int unsupported = 0;
             /* Bind * for both IPv6 and IPv4, we enter here only if
              * server.bindaddr_count == 0. */
             fds[*count] = anetTcp6Server(server.neterr,port,NULL,
@@ -1765,19 +1784,27 @@ int listenToPort(int port, int *fds, int *count) {
             if (fds[*count] != ANET_ERR) {
                 anetNonBlock(NULL,fds[*count]);
                 (*count)++;
+            } else if (errno == EAFNOSUPPORT) {
+                unsupported++;
+                serverLog(LL_WARNING,"Not listening to IPv6: unsupproted");
+            }
 
+            if (*count == 1 || unsupported) {
                 /* Bind the IPv4 address as well. */
                 fds[*count] = anetTcpServer(server.neterr,port,NULL,
                     server.tcp_backlog);
                 if (fds[*count] != ANET_ERR) {
                     anetNonBlock(NULL,fds[*count]);
                     (*count)++;
+                } else if (errno == EAFNOSUPPORT) {
+                    unsupported++;
+                    serverLog(LL_WARNING,"Not listening to IPv4: unsupproted");
                 }
             }
             /* Exit the loop if we were able to bind * on IPv4 and IPv6,
              * otherwise fds[*count] will be ANET_ERR and we'll print an
              * error and return to the caller with an error. */
-            if (*count == 2) break;
+            if (*count + unsupported == 2) break;
         } else if (strchr(server.bindaddr[j],':')) {
             /* Bind IPv6 address. */
             fds[*count] = anetTcp6Server(server.neterr,port,server.bindaddr[j],
@@ -1904,6 +1931,7 @@ void initServer(void) {
     server.rdb_child_pid = -1;
     server.aof_child_pid = -1;
     server.rdb_child_type = RDB_CHILD_TYPE_NONE;
+    server.rdb_bgsave_scheduled = 0;
     aofRewriteBufferReset();
     server.aof_buf = sdsempty();
     server.lastsave = time(NULL); /* At startup we consider the DB saved. */
@@ -3174,11 +3202,15 @@ sds genRedisInfoString(char *section) {
             while((ln = listNext(&li))) {
                 client *slave = listNodeValue(ln);
                 char *state = NULL;
-                char ip[NET_IP_STR_LEN];
+                char ip[NET_IP_STR_LEN], *slaveip = slave->slave_ip;
                 int port;
                 long lag = 0;
 
-                if (anetPeerToString(slave->fd,ip,sizeof(ip),&port) == -1) continue;
+                if (slaveip[0] == '\0') {
+                    if (anetPeerToString(slave->fd,ip,sizeof(ip),&port) == -1)
+                        continue;
+                    slaveip = ip;
+                }
                 switch(slave->replstate) {
                 case SLAVE_STATE_WAIT_BGSAVE_START:
                 case SLAVE_STATE_WAIT_BGSAVE_END:
@@ -3198,7 +3230,7 @@ sds genRedisInfoString(char *section) {
                 info = sdscatprintf(info,
                     "slave%d:ip=%s,port=%d,state=%s,"
                     "offset=%lld,lag=%ld\r\n",
-                    slaveid,ip,slave->slave_listening_port,state,
+                    slaveid,slaveip,slave->slave_listening_port,state,
                     slave->repl_ack_off, lag);
                 slaveid++;
             }
@@ -3968,7 +4000,7 @@ int main(int argc, char **argv) {
      * the program main. However the program is part of the Redis executable
      * so that we can easily execute an RDB check on loading errors. */
     if (strstr(argv[0],"redis-check-rdb") != NULL)
-        exit(redis_check_rdb_main(argv,argc));
+        redis_check_rdb_main(argc,argv);
 
     if (argc >= 2) {
         j = 1; /* First option to parse in argv[] */

src/server.h

@@ -110,7 +110,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #define CONFIG_DEFAULT_CLUSTER_CONFIG_FILE "nodes.conf"
 #define CONFIG_DEFAULT_DAEMONIZE 0
 #define CONFIG_DEFAULT_UNIX_SOCKET_PERM 0
-#define CONFIG_DEFAULT_TCP_KEEPALIVE 0
+#define CONFIG_DEFAULT_TCP_KEEPALIVE 300
 #define CONFIG_DEFAULT_PROTECTED_MODE 1
 #define CONFIG_DEFAULT_LOGFILE ""
 #define CONFIG_DEFAULT_SYSLOG_ENABLED 0
@@ -122,6 +122,8 @@ typedef long long mstime_t; /* millisecond time type. */
 #define CONFIG_DEFAULT_REPL_DISKLESS_SYNC_DELAY 5
 #define CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA 1
 #define CONFIG_DEFAULT_SLAVE_READ_ONLY 1
+#define CONFIG_DEFAULT_SLAVE_ANNOUNCE_IP NULL
+#define CONFIG_DEFAULT_SLAVE_ANNOUNCE_PORT 0
 #define CONFIG_DEFAULT_REPL_DISABLE_TCP_NODELAY 0
 #define CONFIG_DEFAULT_MAXMEMORY 0
 #define CONFIG_DEFAULT_MAXMEMORY_SAMPLES 5
@@ -302,13 +304,15 @@ typedef long long mstime_t; /* millisecond time type. */
 #define REPL_STATE_RECEIVE_AUTH 5 /* Wait for AUTH reply */
 #define REPL_STATE_SEND_PORT 6 /* Send REPLCONF listening-port */
 #define REPL_STATE_RECEIVE_PORT 7 /* Wait for REPLCONF reply */
-#define REPL_STATE_SEND_CAPA 8 /* Send REPLCONF capa */
-#define REPL_STATE_RECEIVE_CAPA 9 /* Wait for REPLCONF reply */
-#define REPL_STATE_SEND_PSYNC 10 /* Send PSYNC */
-#define REPL_STATE_RECEIVE_PSYNC 11 /* Wait for PSYNC reply */
+#define REPL_STATE_SEND_IP 8 /* Send REPLCONF ip-address */
+#define REPL_STATE_RECEIVE_IP 9 /* Wait for REPLCONF reply */
+#define REPL_STATE_SEND_CAPA 10 /* Send REPLCONF capa */
+#define REPL_STATE_RECEIVE_CAPA 11 /* Wait for REPLCONF reply */
+#define REPL_STATE_SEND_PSYNC 12 /* Send PSYNC */
+#define REPL_STATE_RECEIVE_PSYNC 13 /* Wait for PSYNC reply */
 /* --- End of handshake states --- */
-#define REPL_STATE_TRANSFER 12 /* Receiving .rdb from master */
-#define REPL_STATE_CONNECTED 13 /* Connected to master */
+#define REPL_STATE_TRANSFER 14 /* Receiving .rdb from master */
+#define REPL_STATE_CONNECTED 15 /* Connected to 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
@@ -594,7 +598,8 @@ typedef struct client {
                                        copying this slave output buffer
                                        should use. */
     char replrunid[CONFIG_RUN_ID_SIZE+1]; /* Master run id if is a master. */
-    int slave_listening_port; /* As configured with: SLAVECONF listening-port */
+    int slave_listening_port; /* As configured with: REPLCONF listening-port */
+    char slave_ip[NET_IP_STR_LEN]; /* Optionally given by REPLCONF ip-address */
     int slave_capa;         /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
     multiState mstate;      /* MULTI/EXEC state */
     int btype;              /* Type of blocking op if CLIENT_BLOCKED. */
@@ -837,6 +842,7 @@ struct redisServer {
     time_t lastbgsave_try;          /* Unix time of last attempted bgsave */
     time_t rdb_save_time_last;      /* Time used by last RDB save run. */
     time_t rdb_save_time_start;     /* Current RDB save start time. */
+    int rdb_bgsave_scheduled;       /* BGSAVE when possible if true. */
     int rdb_child_type;             /* Type of save by active child. */
     int lastbgsave_status;          /* C_OK or C_ERR */
     int stop_writes_on_bgsave_err;  /* Don't allow writes if can't BGSAVE */
@@ -889,7 +895,9 @@ struct redisServer {
     time_t repl_down_since; /* Unix time at which link with master went down */
     int repl_disable_tcp_nodelay;   /* Disable TCP_NODELAY after SYNC? */
     int slave_priority;             /* Reported in INFO and used by Sentinel. */
-    char repl_master_runid[CONFIG_RUN_ID_SIZE+1];  /* Master run id for PSYNC. */
+    int slave_announce_port;        /* Give the master this listening port. */
+    char *slave_announce_ip;        /* Give the master this ip address. */
+    char repl_master_runid[CONFIG_RUN_ID_SIZE+1];  /* Master run id for PSYNC.*/
     long long repl_master_initial_offset;         /* Master PSYNC offset. */
     /* Replication script cache. */
     dict *repl_scriptcache_dict;        /* SHA1 all slaves are aware of. */
@@ -1446,7 +1454,7 @@ void sentinelIsRunning(void);
 
 /* redis-check-rdb */
 int redis_check_rdb(char *rdbfilename);
-int redis_check_rdb_main(char **argv, int argc);
+int redis_check_rdb_main(int argc, char **argv);
 
 /* Scripting */
 void scriptingInit(int setup);

src/version.h

@@ -1 +1 @@
-#define REDIS_VERSION "3.2.1"
+#define REDIS_VERSION "3.2.5"

tests/integration/rdb.tcl

@@ -89,7 +89,7 @@ close $fd
 start_server_and_kill_it [list "dir" $server_path] {
     test {Server should not start if RDB is corrupted} {
         wait_for_condition 50 100 {
-            [string match {*RDB checksum*} \
+            [string match {*CRC error*} \
                 [exec tail -n10 < [dict get $srv stdout]]]
         } else {
             fail "Server started even if RDB was corrupted!"

tests/sentinel/tests/05-manual.tcl

@@ -6,7 +6,8 @@ test "Manual failover works" {
     set old_port [RI $master_id tcp_port]
     set addr [S 0 SENTINEL GET-MASTER-ADDR-BY-NAME mymaster]
     assert {[lindex $addr 1] == $old_port}
-    S 0 SENTINEL FAILOVER mymaster
+    catch {S 0 SENTINEL FAILOVER mymaster} reply
+    assert {$reply eq "OK"}
     foreach_sentinel_id id {
         wait_for_condition 1000 50 {
             [lindex [S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster] 1] != $old_port

tests/sentinel/tests/07-down-conditions.tcl

@@ -0,0 +1,68 @@
+# Test conditions where an instance is considered to be down
+
+source "../tests/includes/init-tests.tcl"
+
+proc ensure_master_up {} {
+    wait_for_condition 1000 50 {
+        [dict get [S 4 sentinel master mymaster] flags] eq "master"
+    } else {
+        fail "Master flags are not just 'master'"
+    }
+}
+
+proc ensure_master_down {} {
+    wait_for_condition 1000 50 {
+        [string match *down* \
+            [dict get [S 4 sentinel master mymaster] flags]]
+    } else {
+        fail "Master is not flagged SDOWN"
+    }
+}
+
+test "Crash the majority of Sentinels to prevent failovers for this unit" {
+    for {set id 0} {$id < $quorum} {incr id} {
+        kill_instance sentinel $id
+    }
+}
+
+test "SDOWN is triggered by non-responding but not crashed instance" {
+    lassign [S 4 SENTINEL GET-MASTER-ADDR-BY-NAME mymaster] host port
+    ensure_master_up
+    exec ../../../src/redis-cli -h $host -p $port debug sleep 10 > /dev/null &
+    ensure_master_down
+    ensure_master_up
+}
+
+test "SDOWN is triggered by crashed instance" {
+    lassign [S 4 SENTINEL GET-MASTER-ADDR-BY-NAME mymaster] host port
+    ensure_master_up
+    kill_instance redis 0
+    ensure_master_down
+    restart_instance redis 0
+    ensure_master_up
+}
+
+test "SDOWN is triggered by masters advertising as slaves" {
+    ensure_master_up
+    R 0 slaveof 127.0.0.1 34567
+    ensure_master_down
+    R 0 slaveof no one
+    ensure_master_up
+}
+
+test "SDOWN is triggered by misconfigured instance repling with errors" {
+    ensure_master_up
+    set orig_dir [lindex [R 0 config get dir] 1]
+    set orig_save [lindex [R 0 config get save] 1]
+    # Set dir to / and filename to "tmp" to make sure it will fail.
+    R 0 config set dir /
+    R 0 config set dbfilename tmp
+    R 0 config set save "1000000 1000000"
+    R 0 bgsave
+    ensure_master_down
+    R 0 config set save $orig_save
+    R 0 config set dir $orig_dir
+    R 0 config set dbfilename dump.rdb
+    R 0 bgsave
+    ensure_master_up
+}

tests/unit/expire.tcl

@@ -198,4 +198,10 @@ start_server {tags {"expire"}} {
         r set foo b
         lsort [r keys *]
     } {a e foo s t}
+
+    test {EXPIRE with empty string as TTL should report an error} {
+        r set foo bar
+        catch {r expire foo ""} e
+        set e
+    } {*not an integer*}
 }

tests/unit/geo.tcl

@@ -1,4 +1,4 @@
-# Helper functins to simulate search-in-radius in the Tcl side in order to
+# Helper functions to simulate search-in-radius in the Tcl side in order to
 # verify the Redis implementation with a fuzzy test.
 proc geo_degrad deg {expr {$deg*atan(1)*8/360}}
 
@@ -23,6 +23,44 @@ proc geo_random_point {lonvar latvar} {
     set lat [expr {-70 + rand()*140}]
 }
 
+# Return elements non common to both the lists.
+# This code is from http://wiki.tcl.tk/15489
+proc compare_lists {List1 List2} {
+   set DiffList {}
+   foreach Item $List1 {
+      if {[lsearch -exact $List2 $Item] == -1} {
+         lappend DiffList $Item
+      }
+   }
+   foreach Item $List2 {
+      if {[lsearch -exact $List1 $Item] == -1} {
+         if {[lsearch -exact $DiffList $Item] == -1} {
+            lappend DiffList $Item
+         }
+      }
+   }
+   return $DiffList
+}
+
+# The following list represents sets of random seed, search position
+# and radius that caused bugs in the past. It is used by the randomized
+# test later as a starting point. When the regression vectors are scanned
+# the code reverts to using random data.
+#
+# The format is: seed km lon lat
+set regression_vectors {
+    {1412 156 149.29737817929004 15.95807862745508}
+    {441574 143 59.235461856813856 66.269555127373678}
+    {160645 187 -101.88575239939883 49.061997951502917}
+    {750269 154 -90.187939661642517 66.615930412251487}
+    {342880 145 163.03472387745728 64.012747720821181}
+    {729955 143 137.86663517256579 63.986745399416776}
+    {939895 151 59.149620271823181 65.204186651485145}
+    {1412 156 149.29737817929004 15.95807862745508}
+    {564862 149 84.062063109158544 -65.685403922426232}
+}
+set rv_idx 0
+
 start_server {tags {"geo"}} {
     test {GEOADD create} {
         r geoadd nyc -73.9454966 40.747533 "lic market"
@@ -183,16 +221,25 @@ start_server {tags {"geo"}} {
     }
 
     test {GEOADD + GEORANGE randomized test} {
-        set attempt 10
+        set attempt 20
         while {[incr attempt -1]} {
+            set rv [lindex $regression_vectors $rv_idx]
+            incr rv_idx
+
             unset -nocomplain debuginfo
             set srand_seed [randomInt 1000000]
+            if {$rv ne {}} {set srand_seed [lindex $rv 0]}
             lappend debuginfo "srand_seed is $srand_seed"
             expr {srand($srand_seed)} ; # If you need a reproducible run
             r del mypoints
             set radius_km [expr {[randomInt 200]+10}]
+            if {$rv ne {}} {set radius_km [lindex $rv 1]}
             set radius_m [expr {$radius_km*1000}]
             geo_random_point search_lon search_lat
+            if {$rv ne {}} {
+                set search_lon [lindex $rv 2]
+                set search_lat [lindex $rv 3]
+            }
             lappend debuginfo "Search area: $search_lon,$search_lat $radius_km km"
             set tcl_result {}
             set argv {}
@@ -208,10 +255,40 @@ start_server {tags {"geo"}} {
             set res [lsort [r georadius mypoints $search_lon $search_lat $radius_km km]]
             set res2 [lsort $tcl_result]
             set test_result OK
+
             if {$res != $res2} {
+                set rounding_errors 0
+                set diff [compare_lists $res $res2]
+                foreach place $diff {
+                    set mydist [geo_distance $lon $lat $search_lon $search_lat]
+                    set mydist [expr $mydist/1000]
+                    if {($mydist / $radius_km) > 0.999} {incr rounding_errors}
+                }
+                # Make sure this is a real error and not a rounidng issue.
+                if {[llength $diff] == $rounding_errors} {
+                    set res $res2; # Error silenced
+                }
+            }
+
+            if {$res != $res2} {
+                set diff [compare_lists $res $res2]
+                puts "*** Possible problem in GEO radius query ***"
                 puts "Redis: $res"
                 puts "Tcl  : $res2"
+                puts "Diff : $diff"
                 puts [join $debuginfo "\n"]
+                foreach place $diff {
+                    if {[lsearch -exact $res2 $place] != -1} {
+                        set where "(only in Tcl)"
+                    } else {
+                        set where "(only in Redis)"
+                    }
+                    lassign [lindex [r geopos mypoints $place] 0] lon lat
+                    set mydist [geo_distance $lon $lat $search_lon $search_lat]
+                    set mydist [expr $mydist/1000]
+                    puts "$place -> [r geopos mypoints $place] $mydist $where"
+                    if {($mydist / $radius_km) > 0.999} {incr rounding_errors}
+                }
                 set test_result FAIL
             }
             unset -nocomplain debuginfo

tests/unit/type/list-3.tcl

@@ -13,6 +13,50 @@ start_server {
         assert_equal [r lindex l 1] [lindex $mylist 1]
     }
 
+    test {Regression for quicklist #3343 bug} {
+        r del mylist
+        r lpush mylist 401
+        r lpush mylist 392
+        r rpush mylist [string repeat x 5105]"799"
+        r lset mylist -1 [string repeat x 1014]"702"
+        r lpop mylist
+        r lset mylist -1 [string repeat x 4149]"852"
+        r linsert mylist before 401 [string repeat x 9927]"12"
+        r lrange mylist 0 -1
+        r ping ; # It's enough if the server is still alive
+    } {PONG}
+
+    test {Stress tester for #3343-alike bugs} {
+        r del key
+        for {set j 0} {$j < 10000} {incr j} {
+            set op [randomInt 6]
+            set small_signed_count [expr 5-[randomInt 10]]
+            if {[randomInt 2] == 0} {
+                set ele [randomInt 1000]
+            } else {
+                set ele [string repeat x [randomInt 10000]][randomInt 1000]
+            }
+            switch $op {
+                0 {r lpush key $ele}
+                1 {r rpush key $ele}
+                2 {r lpop key}
+                3 {r rpop key}
+                4 {
+                    catch {r lset key $small_signed_count $ele}
+                }
+                5 {
+                    set otherele [randomInt 1000]
+                    if {[randomInt 2] == 0} {
+                        set where before
+                    } else {
+                        set where after
+                    }
+                    r linsert key $where $otherele $ele
+                }
+            }
+        }
+    }
+
     tags {slow} {
         test {ziplist implementation: value encoding and backlink} {
             if {$::accurate} {set iterations 100} else {set iterations 10}

utils/corrupt_rdb.c

@@ -0,0 +1,44 @@
+/* Trivia program to corrupt an RDB file in order to check the RDB check
+ * program behavior and effectiveness.
+ *
+ * Copyright (C) 2016 Salvatore Sanfilippo.
+ * This software is released in the 3-clause BSD license. */
+
+#include <stdio.h>
+#include <fcntl.h>
+#include <sys/stat.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <time.h>
+
+int main(int argc, char **argv) {
+    struct stat stat;
+    int fd, cycles;
+
+    if (argc != 3) {
+        fprintf(stderr,"Usage: <filename> <cycles>\n");
+        exit(1);
+    }
+
+    srand(time(NULL));
+    cycles = atoi(argv[2]);
+    fd = open("dump.rdb",O_RDWR);
+    if (fd == -1) {
+        perror("open");
+        exit(1);
+    }
+    fstat(fd,&stat);
+
+    while(cycles--) {
+        unsigned char buf[32];
+        unsigned long offset = rand()%stat.st_size;
+        int writelen = 1+rand()%31;
+        int j;
+
+        for (j = 0; j < writelen; j++) buf[j] = (char)rand();
+        lseek(fd,offset,SEEK_SET);
+        printf("Writing %d bytes at offset %lu\n", writelen, offset);
+        write(fd,buf,writelen);
+    }
+    return 0;
+}