Conditional replication.

README.Garantia

@@ -0,0 +1,32 @@
+This is the Garantia Data custom Redis version, which is based on the publicly
+available Redis 2.6 with various additional features we use internally.
+
+This file provides some high-level documentation of the additional features.
+
+Conditional Replication
+-----------------------
+
+Conditional replication allows slaves and masters to avoid replication if the
+data on both ends is identical.  This is done by maintaining a 'dbversion'
+value which gets updated on every write (a kind of incremental hash). The
+SYNC command is extended to allow slaves to announce their dbversion and
+request replication from scratch only if it mismatches the master's dbversion.
+
+The 'dbversion' value is stored in the RDB, which makes it incompatible with
+stock Redis RDB files.
+
+
+
+New/Modified Redis Commands
+---------------------------
+
+SYNC <dbversion>
+    When sent by a slave, the master will compare the requested dbversion to
+    its own.  If identical, it will respond with "+INSYNC" and skip creation
+    of RDB file.  Instead, it will begin sending the command stream as if
+    the RDB creation and transmission was complete.
+
+    dbversion is a hash-like value that gets updated whenever a write
+    operation is performed and represents a database "state".  It is also
+    stored in RDB files.
+

redis.conf

@@ -196,6 +196,10 @@ slave-read-only yes
 #
 # repl-timeout 60
 
+# Enable the conditional sync mechanism.  This causes a dbversion hash value
+# to be maintained and has a slight affect on performance.
+conditional-sync yes
+
 # The slave priority is an integer number published by Redis in the INFO output.
 # It is used by Redis Sentinel in order to select a slave to promote into a
 # master if the master is no longer working correctly.

src/aof.c

@@ -478,7 +478,8 @@ int loadAppendOnlyFile(char *filename) {
     struct redis_stat sb;
     int old_aof_state = server.aof_state;
     long loops = 0;
-
+    int select_skipped = 0;
+    
     if (fp && redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0) {
         server.aof_current_size = 0;
         fclose(fp);
@@ -548,6 +549,13 @@ int loadAppendOnlyFile(char *filename) {
         /* The fake client should never get blocked */
         redisAssert((fakeClient->flags & REDIS_BLOCKED) == 0);
 
+        /* Compute dbversion */
+        if (!select_skipped && cmd->proc == selectCommand) {
+            select_skipped = 1;
+        } else {
+            if (server.conditional_sync) update_dbversion(fakeClient);
+        }
+        
         /* Clean up. Command code may have changed argv/argc so we use the
          * argv/argc of the client instead of the local variables. */
         for (j = 0; j < fakeClient->argc; j++)

src/config.c

@@ -256,6 +256,10 @@ void loadServerConfigFromString(char *config) {
             if ((server.daemonize = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }
+        } else if (!strcasecmp(argv[0],"conditional-sync") && argc == 2) {
+            if ((server.conditional_sync = yesnotoi(argv[1])) == -1) {
+                err = "argument must be 'yes' or 'no'"; goto loaderr;
+            }
         } else if (!strcasecmp(argv[0],"appendonly") && argc == 2) {
             int yes;
 

src/rdb.c

@@ -649,9 +649,12 @@ int rdbSave(char *filename) {
     rioInitWithFile(&rdb,fp);
     if (server.rdb_checksum)
         rdb.update_cksum = rioGenericUpdateChecksum;
-    snprintf(magic,sizeof(magic),"REDIS%04d",REDIS_RDB_VERSION);
+    snprintf(magic,sizeof(magic),"REDIS%04d",REDIS_RDB_VERSION_GARANTIA);
     if (rdbWriteRaw(&rdb,magic,9) == -1) goto werr;
 
+    /* write dbversion */
+    if (rdbWriteRaw(&rdb,&server.dbversion, sizeof(server.dbversion)) == -1) goto werr;
+    
     for (j = 0; j < server.dbnum; j++) {
         redisDb *db = server.db+j;
         dict *d = db->dict;
@@ -1095,13 +1098,20 @@ int rdbLoad(char *filename) {
         return REDIS_ERR;
     }
     rdbver = atoi(buf+5);
-    if (rdbver < 1 || rdbver > REDIS_RDB_VERSION) {
+    if (rdbver < 1 || 
+        (rdbver < REDIS_RDB_VERSION_GARANTIA_PREFIX && rdbver > REDIS_RDB_VERSION) ||
+        (rdbver > REDIS_RDB_VERSION_GARANTIA_PREFIX && rdbver > REDIS_RDB_VERSION_GARANTIA)) {
         fclose(fp);
         redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
         errno = EINVAL;
         return REDIS_ERR;
     }
 
+    /* read dbversion */
+    if (rdbver > REDIS_RDB_VERSION_GARANTIA_PREFIX) {
+        if (rioRead(&rdb, &server.dbversion, sizeof(server.dbversion)) == 0) goto eoferr;
+    }
+
     startLoading(fp);
     while(1) {
         robj *key, *val;

src/rdb.h

@@ -39,6 +39,8 @@
 /* The current RDB version. When the format changes in a way that is no longer
  * backward compatible this number gets incremented. */
 #define REDIS_RDB_VERSION 6
+#define REDIS_RDB_VERSION_GARANTIA_PREFIX 1000
+#define REDIS_RDB_VERSION_GARANTIA 1006
 
 /* Defines related to the dump file format. To store 32 bits lengths for short
  * keys requires a lot of space, so we check the most significant 2 bits of

src/redis.c

@@ -219,7 +219,7 @@ struct redisCommand redisCommandTable[] = {
     {"multi",multiCommand,1,"rs",0,NULL,0,0,0,0,0},
     {"exec",execCommand,1,"sM",0,NULL,0,0,0,0,0},
     {"discard",discardCommand,1,"rs",0,NULL,0,0,0,0,0},
-    {"sync",syncCommand,1,"ars",0,NULL,0,0,0,0,0},
+    {"sync",syncCommand,-1,"ars",0,NULL,0,0,0,0,0},
     {"replconf",replconfCommand,-1,"ars",0,NULL,0,0,0,0,0},
     {"flushdb",flushdbCommand,1,"w",0,NULL,0,0,0,0,0},
     {"flushall",flushallCommand,1,"w",0,NULL,0,0,0,0,0},
@@ -1101,6 +1101,7 @@ void initServerConfig() {
     server.syslog_ident = zstrdup("redis");
     server.syslog_facility = LOG_LOCAL0;
     server.daemonize = 0;
+    server.conditional_sync = 1;
     server.aof_state = REDIS_AOF_OFF;
     server.aof_fsync = AOF_FSYNC_EVERYSEC;
     server.aof_no_fsync_on_rewrite = 0;
@@ -1312,6 +1313,7 @@ void initServer() {
     server.rdb_save_time_last = -1;
     server.rdb_save_time_start = -1;
     server.dirty = 0;
+    server.dbversion = 0;
     server.stat_numcommands = 0;
     server.stat_numconnections = 0;
     server.stat_expiredkeys = 0;
@@ -1484,6 +1486,26 @@ void alsoPropagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
     redisOpArrayAppend(&server.also_propagate,cmd,dbid,argv,argc,target);
 }
 
+void update_dbversion(redisClient *c)
+{
+    int i;
+
+    /* we don't want to do real hashing here because of performance, so we
+     * try to bump the dbversion in a somewhat request-related manner without
+     * going through full hashing.
+     */
+    server.dbversion += *(unsigned short *) c->argv[0];     /* command */
+    server.dbversion += c->argc;    
+    for (i = 1; i < c->argc; i++) {
+        robj *a = c->argv[i];
+        if (a->type != REDIS_STRING)
+            continue;
+        if (a->encoding == REDIS_ENCODING_RAW && sdslen(a->ptr) > 4) {
+            server.dbversion += *(unsigned int *)a->ptr;
+        }
+    }
+}
+
 /* Call() is the core of Redis execution of a command */
 void call(redisClient *c, int flags) {
     long long dirty, start = ustime(), duration;
@@ -1514,6 +1536,9 @@ void call(redisClient *c, int flags) {
      * per-command statistics that we show in INFO commandstats. */
     if (flags & REDIS_CALL_SLOWLOG)
         slowlogPushEntryIfNeeded(c->argv,c->argc,duration);
+    if (dirty > 0 && server.conditional_sync) {
+        update_dbversion(c);
+    }
     server.slowlog_complexity_params_count = 0; /* Need to zero the count in case we're in a nested call to "call()" */
     if (flags & REDIS_CALL_STATS) {
         c->cmd->microseconds += duration;
@@ -1952,6 +1977,7 @@ sds genRedisInfoString(char *section) {
             "# Persistence\r\n"
             "loading:%d\r\n"
             "rdb_changes_since_last_save:%lld\r\n"
+            "rdb_dbversion:%016llx\r\n"
             "rdb_bgsave_in_progress:%d\r\n"
             "rdb_last_save_time:%ld\r\n"
             "rdb_last_bgsave_status:%s\r\n"
@@ -1965,6 +1991,7 @@ sds genRedisInfoString(char *section) {
             "aof_last_bgrewrite_status:%s\r\n",
             server.loading,
             server.dirty,
+            server.dbversion,
             server.rdb_child_pid != -1,
             server.lastsave,
             (server.lastbgsave_status == REDIS_OK) ? "ok" : "err",

src/redis.h

@@ -563,6 +563,7 @@ struct redisServer {
     size_t client_max_querybuf_len; /* Limit for client query buffer length */
     int dbnum;                      /* Total number of configured DBs */
     int daemonize;                  /* True if running as a daemon */
+    int conditional_sync;           /* Conditional synchronziation support */
     clientBufferLimitsConfig client_obuf_limits[REDIS_CLIENT_LIMIT_NUM_CLASSES];
     /* AOF persistence */
     int aof_state;                  /* REDIS_AOF_(ON|OFF|WAIT_REWRITE) */
@@ -588,6 +589,7 @@ struct redisServer {
     /* RDB persistence */
     long long dirty;                /* Changes to DB from the last save */
     long long dirty_before_bgsave;  /* Used to restore dirty on failed BGSAVE */
+    unsigned long long dbversion;   /* Current unique DB version */
     pid_t rdb_child_pid;            /* PID of RDB saving child */
     struct saveparam *saveparams;   /* Save points array for RDB */
     int saveparamslen;              /* Number of saving points */
@@ -965,6 +967,7 @@ int htNeedsResize(dict *dict);
 void oom(const char *msg);
 void populateCommandTable(void);
 void resetCommandTableStats(void);
+void update_dbversion(redisClient *c);
 
 /* Set data type */
 robj *setTypeCreate(robj *value);

src/replication.c

@@ -115,6 +115,8 @@ void replicationFeedMonitors(redisClient *c, list *monitors, int dictid, robj **
 }
 
 void syncCommand(redisClient *c) {
+    int bgsave_required = 1;
+    
     /* ignore SYNC if aleady slave or in monitor mode */
     if (c->flags & REDIS_SLAVE) return;
 
@@ -135,44 +137,61 @@ void syncCommand(redisClient *c) {
     }
 
     redisLog(REDIS_NOTICE,"Slave ask for synchronization");
-    /* Here we need to check if there is a background saving operation
-     * in progress, or if it is required to start one */
-    if (server.rdb_child_pid != -1) {
-        /* Ok a background save is in progress. Let's check if it is a good
-         * one for replication, i.e. if there is another slave that is
-         * registering differences since the server forked to save */
-        redisClient *slave;
-        listNode *ln;
-        listIter li;
-
-        listRewind(server.slaves,&li);
-        while((ln = listNext(&li))) {
-            slave = ln->value;
-            if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
-        }
-        if (ln) {
-            /* Perfect, the server is already registering differences for
-             * another slave. Set the right state, and copy the buffer. */
-            copyClientOutputBuffer(c,slave);
-            c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
-            redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
-        } else {
-            /* No way, we need to wait for the next BGSAVE in order to
-             * register differences */
-            c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
-            redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
-        }
-    } else {
-        /* Ok we don't have a BGSAVE in progress, let's start one */
-        redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC");
-        if (rdbSaveBackground(server.rdb_filename) != REDIS_OK) {
-            redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
-            addReplyError(c,"Unable to perform background save");
+    if (c->argc == 2) {
+        if (!server.conditional_sync) {
+            addReplyError(c,"Conditional SYNC is not enabled on this server");
             return;
         }
-        c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
+        unsigned long long sync_dbversion = strtoull((const char *)c->argv[1]->ptr, NULL, 16);
+        if (sync_dbversion == server.dbversion) {
+            redisLog(REDIS_NOTICE, "Slave is in sync, BGSAVE is not necessary");
+            c->replstate = REDIS_REPL_ONLINE;
+            bgsave_required = 0;
+            addReplyStatus(c, "INSYNC");
+        }
     }
-    c->repldbfd = -1;
+
+    if (bgsave_required) {    
+        /* Here we need to check if there is a background saving operation
+         * in progress, or if it is required to start one */
+        if (server.rdb_child_pid != -1) {
+            /* Ok a background save is in progress. Let's check if it is a good
+             * one for replication, i.e. if there is another slave that is
+             * registering differences since the server forked to save */
+            redisClient *slave;
+            listNode *ln;
+            listIter li;
+
+            listRewind(server.slaves,&li);
+            while((ln = listNext(&li))) {
+                slave = ln->value;
+                if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
+            }
+            if (ln) {
+                /* Perfect, the server is already registering differences for
+                 * another slave. Set the right state, and copy the buffer. */
+                copyClientOutputBuffer(c,slave);
+                c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
+                redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
+            } else {
+                /* No way, we need to wait for the next BGSAVE in order to
+                 * register differences */
+                c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
+                redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
+            }
+        } else {
+            /* Ok we don't have a BGSAVE in progress, let's start one */
+            redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC");
+            if (rdbSaveBackground(server.rdb_filename) != REDIS_OK) {
+                redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
+                addReplyError(c,"Unable to perform background save");
+                return;
+            }
+            c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
+        }
+        c->repldbfd = -1;
+    }
+    
     c->flags |= REDIS_SLAVE;
     c->slaveseldb = 0;
     listAddNodeTail(server.slaves,c);
@@ -349,6 +368,7 @@ void replicationAbortSyncTransfer(void) {
 void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
     char buf[4096];
     ssize_t nread, readlen;
+    int insync = 0;
     off_t left;
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(privdata);
@@ -375,65 +395,76 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
              * timestamp. */
             server.repl_transfer_lastio = server.unixtime;
             return;
+        } else if (server.conditional_sync && (strcmp(buf, "+INSYNC") == 0)) {
+            redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: already in sync, proceeding.");
+            unlink(server.repl_transfer_tmpfile);
+            aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
+            insync = 1;
         } else if (buf[0] != '$') {
             redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
             goto error;
         }
-        server.repl_transfer_size = strtol(buf+1,NULL,10);
-        redisLog(REDIS_NOTICE,
-            "MASTER <-> SLAVE sync: receiving %ld bytes from master",
-            server.repl_transfer_size);
-        return;
+
+        if (!insync) {
+            server.repl_transfer_size = strtol(buf+1,NULL,10);
+            redisLog(REDIS_NOTICE,
+                "MASTER <-> SLAVE sync: receiving %ld bytes from master",
+                server.repl_transfer_size);
+            return;
+        }
     }
 
     /* Read bulk data */
-    left = server.repl_transfer_size - server.repl_transfer_read;
-    readlen = (left < (signed)sizeof(buf)) ? left : (signed)sizeof(buf);
-    nread = read(fd,buf,readlen);
-    if (nread <= 0) {
-        redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
-            (nread == -1) ? strerror(errno) : "connection lost");
-        replicationAbortSyncTransfer();
-        return;
-    }
-    server.repl_transfer_lastio = server.unixtime;
-    if (write(server.repl_transfer_fd,buf,nread) != nread) {
-        redisLog(REDIS_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchronization: %s", strerror(errno));
-        goto error;
-    }
-    server.repl_transfer_read += nread;
-
-    /* Sync data on disk from time to time, otherwise at the end of the transfer
-     * we may suffer a big delay as the memory buffers are copied into the
-     * actual disk. */
-    if (server.repl_transfer_read >=
-        server.repl_transfer_last_fsync_off + REPL_MAX_WRITTEN_BEFORE_FSYNC)
-    {
-        off_t sync_size = server.repl_transfer_read -
-                          server.repl_transfer_last_fsync_off;
-        rdb_fsync_range(server.repl_transfer_fd,
-            server.repl_transfer_last_fsync_off, sync_size);
-        server.repl_transfer_last_fsync_off += sync_size;
-    }
-
-    /* Check if the transfer is now complete */
-    if (server.repl_transfer_read == server.repl_transfer_size) {
-        if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
-            redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
+    if (!insync) {
+        left = server.repl_transfer_size - server.repl_transfer_read;
+        readlen = (left < (signed)sizeof(buf)) ? left : (signed)sizeof(buf);
+        nread = read(fd,buf,readlen);
+        if (nread <= 0) {
+            redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
+                (nread == -1) ? strerror(errno) : "connection lost");
             replicationAbortSyncTransfer();
             return;
         }
-        redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory");
-        emptyDb();
-        /* Before loading the DB into memory we need to delete the readable
-         * handler, otherwise it will get called recursively since
-         * rdbLoad() will call the event loop to process events from time to
-         * time for non blocking loading. */
-        aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
-        if (rdbLoad(server.rdb_filename) != REDIS_OK) {
-            redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
-            replicationAbortSyncTransfer();
-            return;
+        server.repl_transfer_lastio = server.unixtime;
+        if (write(server.repl_transfer_fd,buf,nread) != nread) {
+            redisLog(REDIS_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchronization: %s", strerror(errno));
+            goto error;
+        }
+        server.repl_transfer_read += nread;
+
+        /* Sync data on disk from time to time, otherwise at the end of the transfer
+         * we may suffer a big delay as the memory buffers are copied into the
+         * actual disk. */
+        if (server.repl_transfer_read >=
+            server.repl_transfer_last_fsync_off + REPL_MAX_WRITTEN_BEFORE_FSYNC)
+        {
+            off_t sync_size = server.repl_transfer_read -
+                              server.repl_transfer_last_fsync_off;
+            rdb_fsync_range(server.repl_transfer_fd,
+                server.repl_transfer_last_fsync_off, sync_size);
+            server.repl_transfer_last_fsync_off += sync_size;
+        }
+    }
+    /* Check if the transfer is now complete */
+    if (insync || (server.repl_transfer_read == server.repl_transfer_size)) {
+        if (!insync) {
+            if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
+                redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
+                replicationAbortSyncTransfer();
+                return;
+            }
+            redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory");
+            emptyDb();
+            /* Before loading the DB into memory we need to delete the readable
+             * handler, otherwise it will get called recursively since
+             * rdbLoad() will call the event loop to process events from time to
+             * time for non blocking loading. */
+            aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
+            if (rdbLoad(server.rdb_filename) != REDIS_OK) {
+                redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
+                replicationAbortSyncTransfer();
+                return;
+            }
         }
         /* Final setup of the connected slave <- master link */
         zfree(server.repl_transfer_tmpfile);
@@ -515,7 +546,7 @@ char *sendSynchronousCommand(int fd, ...) {
 }
 
 void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
-    char tmpfile[256], *err;
+    char tmpfile[256], *err, synccmd[64];
     int dfd, maxtries = 5;
     int sockerr = 0;
     socklen_t errlen = sizeof(sockerr);
@@ -613,7 +644,12 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
     }
 
     /* Issue the SYNC command */
-    if (syncWrite(fd,"SYNC\r\n",6,server.repl_syncio_timeout*1000) == -1) {
+    if (server.conditional_sync) {
+        snprintf(synccmd, sizeof(synccmd)-1, "SYNC %016llx\r\n", server.dbversion);
+    } else {
+        strncpy(synccmd, "SYNC\r\n", sizeof(synccmd)-1);
+    }
+    if (syncWrite(fd,synccmd,strlen(synccmd),server.repl_syncio_timeout*1000) == -1) {
         redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
             strerror(errno));
         goto error;