From 30ff59e645964770be49acd6fce84112ad8fbcb9 Mon Sep 17 00:00:00 2001 From: Yossi Gottlieb Date: Tue, 18 Dec 2012 17:06:32 +0200 Subject: [PATCH] Conditional replication. --- README.Garantia | 32 +++++++ redis.conf | 4 + src/aof.c | 10 ++- src/config.c | 4 + src/rdb.c | 14 +++- src/rdb.h | 2 + src/redis.c | 29 ++++++- src/redis.h | 3 + src/replication.c | 208 +++++++++++++++++++++++++++------------------- 9 files changed, 216 insertions(+), 90 deletions(-) create mode 100644 README.Garantia mode change 100644 => 100755 redis.conf mode change 100644 => 100755 src/aof.c mode change 100644 => 100755 src/config.c mode change 100644 => 100755 src/rdb.c mode change 100644 => 100755 src/rdb.h mode change 100644 => 100755 src/redis.c mode change 100644 => 100755 src/redis.h mode change 100644 => 100755 src/replication.c diff --git a/README.Garantia b/README.Garantia new file mode 100644 index 000000000..418a50ab9 --- /dev/null +++ b/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 + 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. + diff --git a/redis.conf b/redis.conf old mode 100644 new mode 100755 index 97aea334b..a678cf2bb --- a/redis.conf +++ b/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. diff --git a/src/aof.c b/src/aof.c old mode 100644 new mode 100755 index ae45e4f0b..dcce6e7de --- a/src/aof.c +++ b/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++) diff --git a/src/config.c b/src/config.c old mode 100644 new mode 100755 index 7e8f8a2ba..0d74b6784 --- a/src/config.c +++ b/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; diff --git a/src/rdb.c b/src/rdb.c old mode 100644 new mode 100755 index dbda7a526..d24f6597d --- a/src/rdb.c +++ b/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; diff --git a/src/rdb.h b/src/rdb.h old mode 100644 new mode 100755 index d0f2aad86..536ce9757 --- a/src/rdb.h +++ b/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 diff --git a/src/redis.c b/src/redis.c old mode 100644 new mode 100755 index d860a7b24..3c9310442 --- a/src/redis.c +++ b/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", diff --git a/src/redis.h b/src/redis.h old mode 100644 new mode 100755 index 9ea716638..c1ebe6e2a --- a/src/redis.h +++ b/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); diff --git a/src/replication.c b/src/replication.c old mode 100644 new mode 100755 index 720cd4c19..ff1c8414b --- a/src/replication.c +++ b/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;