Commit 2669fb83 authored by antirez's avatar antirez
Browse files

PSYNC2: different improvements to Redis replication.

The gist of the changes is that now, partial resynchronizations between
slaves and masters (without the need of a full resync with RDB transfer
and so forth), work in a number of cases when it was impossible
in the past. For instance:

1. When a slave is promoted to mastrer, the slaves of the old master can
partially resynchronize with the new master.

2. Chained slalves (slaves of slaves) can be moved to replicate to other
slaves or the master itsef, without requiring a full resync.

3. The master itself, after being turned into a slave, is able to
partially resynchronize with the new master, when it joins replication
again.

In order to obtain this, the following main changes were operated:

* Slaves also take a replication backlog, not just masters.

* Same stream replication for all the slaves and sub slaves. The
replication stream is identical from the top level master to its slaves
and is also the same from the slaves to their sub-slaves and so forth.
This means that if a slave is later promoted to master, it has the
same replication backlong, and can partially resynchronize with its
slaves (that were previously slaves of the old master).

* A given replication history is no longer identified by the `runid` of
a Redis node. There is instead a `replication ID` which changes every
time the instance has a new history no longer coherent with the past
one. So, for example, slaves publish the same replication history of
their master, however when they are turned into masters, they publish
a new replication ID, but still remember the old ID, so that they are
able to partially resynchronize with slaves of the old master (up to a
given offset).

* The replication protocol was slightly modified so that a new extended
+CONTINUE reply from the master is able to inform the slave of a
replication ID change.

* REPLCONF CAPA is used in order to notify masters that a slave is able
to understand the new +CONTINUE reply.

* The RDB file was extended with an auxiliary field that is able to
select a given DB after loading in the slave, so that the slave can
continue receiving the replication stream from the point it was
disconnected without requiring the master to insert "SELECT" statements.
This is useful in order to guarantee the "same stream" property, because
the slave must be able to accumulate an identical backlog.

* Slave pings to sub-slaves are now sent in a special form, when the
top-level master is disconnected, in order to don't interfer with the
replication stream. We just use out of band "\n" bytes as in other parts
of the Redis protocol.

An old design document is available here:

https://gist.github.com/antirez/ae068f95c0d084891305

However the implementation is not identical to the description because
during the work to implement it, different changes were needed in order
to make things working well.
parent 18d32c7e
...@@ -402,6 +402,10 @@ repl-disable-tcp-nodelay no ...@@ -402,6 +402,10 @@ repl-disable-tcp-nodelay no
# need to elapse, starting from the time the last slave disconnected, for # need to elapse, starting from the time the last slave disconnected, for
# the backlog buffer to be freed. # the backlog buffer to be freed.
# #
# Note that slaves never free the backlog for timeout, since they may be
# promoted to masters later, and should be able to correctly "partially
# resynchronize" with the slaves: hence they should always accumulate backlog.
#
# A value of 0 means to never release the backlog. # A value of 0 means to never release the backlog.
# #
# repl-backlog-ttl 3600 # repl-backlog-ttl 3600
......
...@@ -653,7 +653,7 @@ int loadAppendOnlyFile(char *filename) { ...@@ -653,7 +653,7 @@ int loadAppendOnlyFile(char *filename) {
serverLog(LL_NOTICE,"Reading RDB preamble from AOF file..."); serverLog(LL_NOTICE,"Reading RDB preamble from AOF file...");
if (fseek(fp,0,SEEK_SET) == -1) goto readerr; if (fseek(fp,0,SEEK_SET) == -1) goto readerr;
rioInitWithFile(&rdb,fp); rioInitWithFile(&rdb,fp);
if (rdbLoadRio(&rdb) != C_OK) { if (rdbLoadRio(&rdb,NULL) != C_OK) {
serverLog(LL_WARNING,"Error reading the RDB preamble of the AOF file, AOF loading aborted"); serverLog(LL_WARNING,"Error reading the RDB preamble of the AOF file, AOF loading aborted");
goto readerr; goto readerr;
} else { } else {
...@@ -1152,7 +1152,7 @@ int rewriteAppendOnlyFile(char *filename) { ...@@ -1152,7 +1152,7 @@ int rewriteAppendOnlyFile(char *filename) {
if (server.aof_use_rdb_preamble) { if (server.aof_use_rdb_preamble) {
int error; int error;
if (rdbSaveRio(&aof,&error,RDB_SAVE_AOF_PREAMBLE) == C_ERR) { if (rdbSaveRio(&aof,&error,RDB_SAVE_AOF_PREAMBLE,NULL) == C_ERR) {
errno = error; errno = error;
goto werr; goto werr;
} }
......
...@@ -413,7 +413,7 @@ void flushallCommand(client *c) { ...@@ -413,7 +413,7 @@ void flushallCommand(client *c) {
/* Normally rdbSave() will reset dirty, but we don't want this here /* Normally rdbSave() will reset dirty, but we don't want this here
* as otherwise FLUSHALL will not be replicated nor put into the AOF. */ * as otherwise FLUSHALL will not be replicated nor put into the AOF. */
int saved_dirty = server.dirty; int saved_dirty = server.dirty;
rdbSave(server.rdb_filename); rdbSave(server.rdb_filename,NULL);
server.dirty = saved_dirty; server.dirty = saved_dirty;
} }
server.dirty++; server.dirty++;
......
...@@ -320,12 +320,12 @@ void debugCommand(client *c) { ...@@ -320,12 +320,12 @@ void debugCommand(client *c) {
if (c->argc >= 3) c->argv[2] = tryObjectEncoding(c->argv[2]); if (c->argc >= 3) c->argv[2] = tryObjectEncoding(c->argv[2]);
serverAssertWithInfo(c,c->argv[0],1 == 2); serverAssertWithInfo(c,c->argv[0],1 == 2);
} else if (!strcasecmp(c->argv[1]->ptr,"reload")) { } else if (!strcasecmp(c->argv[1]->ptr,"reload")) {
if (rdbSave(server.rdb_filename) != C_OK) { if (rdbSave(server.rdb_filename,NULL) != C_OK) {
addReply(c,shared.err); addReply(c,shared.err);
return; return;
} }
emptyDb(-1,EMPTYDB_NO_FLAGS,NULL); emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
if (rdbLoad(server.rdb_filename) != C_OK) { if (rdbLoad(server.rdb_filename,NULL) != C_OK) {
addReplyError(c,"Error trying to load the RDB dump"); addReplyError(c,"Error trying to load the RDB dump");
return; return;
} }
......
...@@ -352,6 +352,14 @@ void addReplySds(client *c, sds s) { ...@@ -352,6 +352,14 @@ void addReplySds(client *c, sds s) {
} }
} }
/* This low level function just adds whatever protocol you send it to the
* client buffer, trying the static buffer initially, and using the string
* of objects if not possible.
*
* It is efficient because does not create an SDS object nor an Redis object
* if not needed. The object will only be created by calling
* _addReplyStringToList() if we fail to extend the existing tail object
* in the list of objects. */
void addReplyString(client *c, const char *s, size_t len) { void addReplyString(client *c, const char *s, size_t len) {
if (prepareClientToWrite(c) != C_OK) return; if (prepareClientToWrite(c) != C_OK) return;
if (_addReplyToBuffer(c,s,len) != C_OK) if (_addReplyToBuffer(c,s,len) != C_OK)
...@@ -1022,7 +1030,7 @@ int processInlineBuffer(client *c) { ...@@ -1022,7 +1030,7 @@ int processInlineBuffer(client *c) {
char *newline; char *newline;
int argc, j; int argc, j;
sds *argv, aux; sds *argv, aux;
size_t querylen; size_t querylen, protolen;
/* Search for end of line */ /* Search for end of line */
newline = strchr(c->querybuf,'\n'); newline = strchr(c->querybuf,'\n');
...@@ -1035,6 +1043,7 @@ int processInlineBuffer(client *c) { ...@@ -1035,6 +1043,7 @@ int processInlineBuffer(client *c) {
} }
return C_ERR; return C_ERR;
} }
protolen = (newline - c->querybuf)+1; /* Total protocol bytes of command. */
/* Handle the \r\n case. */ /* Handle the \r\n case. */
if (newline && newline != c->querybuf && *(newline-1) == '\r') if (newline && newline != c->querybuf && *(newline-1) == '\r')
...@@ -1057,6 +1066,15 @@ int processInlineBuffer(client *c) { ...@@ -1057,6 +1066,15 @@ int processInlineBuffer(client *c) {
if (querylen == 0 && c->flags & CLIENT_SLAVE) if (querylen == 0 && c->flags & CLIENT_SLAVE)
c->repl_ack_time = server.unixtime; c->repl_ack_time = server.unixtime;
/* Newline from masters can be used to prevent timeouts, but should
* not affect the replication offset since they are always sent
* "out of band" directly writing to the socket and without passing
* from the output buffers. */
if (querylen == 0 && c->flags & CLIENT_MASTER) {
c->reploff -= protolen;
while (protolen--) chopReplicationBacklog();
}
/* Leave data after the first line of the query in the buffer */ /* Leave data after the first line of the query in the buffer */
sdsrange(c->querybuf,querylen+2,-1); sdsrange(c->querybuf,querylen+2,-1);
...@@ -1321,7 +1339,11 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) { ...@@ -1321,7 +1339,11 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
sdsIncrLen(c->querybuf,nread); sdsIncrLen(c->querybuf,nread);
c->lastinteraction = server.unixtime; c->lastinteraction = server.unixtime;
if (c->flags & CLIENT_MASTER) c->reploff += nread; if (c->flags & CLIENT_MASTER) {
c->reploff += nread;
replicationFeedSlavesFromMasterStream(server.slaves,
c->querybuf+qblen,nread);
}
server.stat_net_input_bytes += nread; server.stat_net_input_bytes += nread;
if (sdslen(c->querybuf) > server.client_max_querybuf_len) { if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty(); sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();
......
...@@ -835,7 +835,7 @@ int rdbSaveAuxFieldStrInt(rio *rdb, char *key, long long val) { ...@@ -835,7 +835,7 @@ int rdbSaveAuxFieldStrInt(rio *rdb, char *key, long long val) {
} }
/* Save a few default AUX fields with information about the RDB generated. */ /* Save a few default AUX fields with information about the RDB generated. */
int rdbSaveInfoAuxFields(rio *rdb, int flags) { int rdbSaveInfoAuxFields(rio *rdb, int flags, rdbSaveInfo *rsi) {
int redis_bits = (sizeof(void*) == 8) ? 64 : 32; int redis_bits = (sizeof(void*) == 8) ? 64 : 32;
int aof_preamble = (flags & RDB_SAVE_AOF_PREAMBLE) != 0; int aof_preamble = (flags & RDB_SAVE_AOF_PREAMBLE) != 0;
...@@ -844,6 +844,16 @@ int rdbSaveInfoAuxFields(rio *rdb, int flags) { ...@@ -844,6 +844,16 @@ int rdbSaveInfoAuxFields(rio *rdb, int flags) {
if (rdbSaveAuxFieldStrInt(rdb,"redis-bits",redis_bits) == -1) return -1; if (rdbSaveAuxFieldStrInt(rdb,"redis-bits",redis_bits) == -1) return -1;
if (rdbSaveAuxFieldStrInt(rdb,"ctime",time(NULL)) == -1) return -1; if (rdbSaveAuxFieldStrInt(rdb,"ctime",time(NULL)) == -1) return -1;
if (rdbSaveAuxFieldStrInt(rdb,"used-mem",zmalloc_used_memory()) == -1) return -1; if (rdbSaveAuxFieldStrInt(rdb,"used-mem",zmalloc_used_memory()) == -1) return -1;
/* Handle saving options that generate aux fields. */
if (rsi) {
if (rsi->repl_stream_db &&
rdbSaveAuxFieldStrInt(rdb,"repl-stream-db",rsi->repl_stream_db)
== -1)
{
return -1;
}
}
if (rdbSaveAuxFieldStrInt(rdb,"aof-preamble",aof_preamble) == -1) return -1; if (rdbSaveAuxFieldStrInt(rdb,"aof-preamble",aof_preamble) == -1) return -1;
return 1; return 1;
} }
...@@ -856,7 +866,7 @@ int rdbSaveInfoAuxFields(rio *rdb, int flags) { ...@@ -856,7 +866,7 @@ int rdbSaveInfoAuxFields(rio *rdb, int flags) {
* When the function returns C_ERR and if 'error' is not NULL, the * When the function returns C_ERR and if 'error' is not NULL, the
* integer pointed by 'error' is set to the value of errno just after the I/O * integer pointed by 'error' is set to the value of errno just after the I/O
* error. */ * error. */
int rdbSaveRio(rio *rdb, int *error, int flags) { int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
dictIterator *di = NULL; dictIterator *di = NULL;
dictEntry *de; dictEntry *de;
char magic[10]; char magic[10];
...@@ -869,7 +879,7 @@ int rdbSaveRio(rio *rdb, int *error, int flags) { ...@@ -869,7 +879,7 @@ int rdbSaveRio(rio *rdb, int *error, int flags) {
rdb->update_cksum = rioGenericUpdateChecksum; rdb->update_cksum = rioGenericUpdateChecksum;
snprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION); snprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION);
if (rdbWriteRaw(rdb,magic,9) == -1) goto werr; if (rdbWriteRaw(rdb,magic,9) == -1) goto werr;
if (rdbSaveInfoAuxFields(rdb,flags) == -1) goto werr; if (rdbSaveInfoAuxFields(rdb,flags,rsi) == -1) goto werr;
for (j = 0; j < server.dbnum; j++) { for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j; redisDb *db = server.db+j;
...@@ -945,7 +955,7 @@ werr: ...@@ -945,7 +955,7 @@ werr:
* While the suffix is the 40 bytes hex string we announced in the prefix. * While the suffix is the 40 bytes hex string we announced in the prefix.
* This way processes receiving the payload can understand when it ends * This way processes receiving the payload can understand when it ends
* without doing any processing of the content. */ * without doing any processing of the content. */
int rdbSaveRioWithEOFMark(rio *rdb, int *error) { int rdbSaveRioWithEOFMark(rio *rdb, int *error, rdbSaveInfo *rsi) {
char eofmark[RDB_EOF_MARK_SIZE]; char eofmark[RDB_EOF_MARK_SIZE];
getRandomHexChars(eofmark,RDB_EOF_MARK_SIZE); getRandomHexChars(eofmark,RDB_EOF_MARK_SIZE);
...@@ -953,7 +963,7 @@ int rdbSaveRioWithEOFMark(rio *rdb, int *error) { ...@@ -953,7 +963,7 @@ int rdbSaveRioWithEOFMark(rio *rdb, int *error) {
if (rioWrite(rdb,"$EOF:",5) == 0) goto werr; if (rioWrite(rdb,"$EOF:",5) == 0) goto werr;
if (rioWrite(rdb,eofmark,RDB_EOF_MARK_SIZE) == 0) goto werr; if (rioWrite(rdb,eofmark,RDB_EOF_MARK_SIZE) == 0) goto werr;
if (rioWrite(rdb,"\r\n",2) == 0) goto werr; if (rioWrite(rdb,"\r\n",2) == 0) goto werr;
if (rdbSaveRio(rdb,error,RDB_SAVE_NONE) == C_ERR) goto werr; if (rdbSaveRio(rdb,error,RDB_SAVE_NONE,rsi) == C_ERR) goto werr;
if (rioWrite(rdb,eofmark,RDB_EOF_MARK_SIZE) == 0) goto werr; if (rioWrite(rdb,eofmark,RDB_EOF_MARK_SIZE) == 0) goto werr;
return C_OK; return C_OK;
...@@ -964,7 +974,7 @@ werr: /* Write error. */ ...@@ -964,7 +974,7 @@ werr: /* Write error. */
} }
/* Save the DB on disk. Return C_ERR on error, C_OK on success. */ /* Save the DB on disk. Return C_ERR on error, C_OK on success. */
int rdbSave(char *filename) { int rdbSave(char *filename, rdbSaveInfo *rsi) {
char tmpfile[256]; char tmpfile[256];
char cwd[MAXPATHLEN]; /* Current working dir path for error messages. */ char cwd[MAXPATHLEN]; /* Current working dir path for error messages. */
FILE *fp; FILE *fp;
...@@ -985,7 +995,7 @@ int rdbSave(char *filename) { ...@@ -985,7 +995,7 @@ int rdbSave(char *filename) {
} }
rioInitWithFile(&rdb,fp); rioInitWithFile(&rdb,fp);
if (rdbSaveRio(&rdb,&error,RDB_SAVE_NONE) == C_ERR) { if (rdbSaveRio(&rdb,&error,RDB_SAVE_NONE,rsi) == C_ERR) {
errno = error; errno = error;
goto werr; goto werr;
} }
...@@ -1023,7 +1033,7 @@ werr: ...@@ -1023,7 +1033,7 @@ werr:
return C_ERR; return C_ERR;
} }
int rdbSaveBackground(char *filename) { int rdbSaveBackground(char *filename, rdbSaveInfo *rsi) {
pid_t childpid; pid_t childpid;
long long start; long long start;
...@@ -1040,7 +1050,7 @@ int rdbSaveBackground(char *filename) { ...@@ -1040,7 +1050,7 @@ int rdbSaveBackground(char *filename) {
/* Child */ /* Child */
closeListeningSockets(0); closeListeningSockets(0);
redisSetProcTitle("redis-rdb-bgsave"); redisSetProcTitle("redis-rdb-bgsave");
retval = rdbSave(filename); retval = rdbSave(filename,rsi);
if (retval == C_OK) { if (retval == C_OK) {
size_t private_dirty = zmalloc_get_private_dirty(-1); size_t private_dirty = zmalloc_get_private_dirty(-1);
...@@ -1410,7 +1420,7 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) { ...@@ -1410,7 +1420,7 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
/* Load an RDB file from the rio stream 'rdb'. On success C_OK is returned, /* Load an RDB file from the rio stream 'rdb'. On success C_OK is returned,
* otherwise C_ERR is returned and 'errno' is set accordingly. */ * otherwise C_ERR is returned and 'errno' is set accordingly. */
int rdbLoadRio(rio *rdb) { int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi) {
uint64_t dbid; uint64_t dbid;
int type, rdbver; int type, rdbver;
redisDb *db = server.db+0; redisDb *db = server.db+0;
...@@ -1501,6 +1511,8 @@ int rdbLoadRio(rio *rdb) { ...@@ -1501,6 +1511,8 @@ int rdbLoadRio(rio *rdb) {
serverLog(LL_NOTICE,"RDB '%s': %s", serverLog(LL_NOTICE,"RDB '%s': %s",
(char*)auxkey->ptr, (char*)auxkey->ptr,
(char*)auxval->ptr); (char*)auxval->ptr);
} else if (!strcasecmp(auxkey->ptr,"repl-stream-db")) {
if (rsi) rsi->repl_stream_db = atoi(auxval->ptr);
} else { } else {
/* We ignore fields we don't understand, as by AUX field /* We ignore fields we don't understand, as by AUX field
* contract. */ * contract. */
...@@ -1559,8 +1571,11 @@ eoferr: /* unexpected end of file is handled here with a fatal exit */ ...@@ -1559,8 +1571,11 @@ eoferr: /* unexpected end of file is handled here with a fatal exit */
/* Like rdbLoadRio() but takes a filename instead of a rio stream. The /* Like rdbLoadRio() but takes a filename instead of a rio stream. The
* filename is open for reading and a rio stream object created in order * filename is open for reading and a rio stream object created in order
* to do the actual loading. Moreover the ETA displayed in the INFO * to do the actual loading. Moreover the ETA displayed in the INFO
* output is initialized and finalized. */ * output is initialized and finalized.
int rdbLoad(char *filename) { *
* If you pass an 'rsi' structure initialied with RDB_SAVE_OPTION_INIT, the
* loading code will fiil the information fields in the structure. */
int rdbLoad(char *filename, rdbSaveInfo *rsi) {
FILE *fp; FILE *fp;
rio rdb; rio rdb;
int retval; int retval;
...@@ -1568,7 +1583,7 @@ int rdbLoad(char *filename) { ...@@ -1568,7 +1583,7 @@ int rdbLoad(char *filename) {
if ((fp = fopen(filename,"r")) == NULL) return C_ERR; if ((fp = fopen(filename,"r")) == NULL) return C_ERR;
startLoading(fp); startLoading(fp);
rioInitWithFile(&rdb,fp); rioInitWithFile(&rdb,fp);
retval = rdbLoadRio(&rdb); retval = rdbLoadRio(&rdb,rsi);
fclose(fp); fclose(fp);
stopLoading(); stopLoading();
return retval; return retval;
...@@ -1721,7 +1736,7 @@ void backgroundSaveDoneHandler(int exitcode, int bysignal) { ...@@ -1721,7 +1736,7 @@ void backgroundSaveDoneHandler(int exitcode, int bysignal) {
/* Spawn an RDB child that writes the RDB to the sockets of the slaves /* Spawn an RDB child that writes the RDB to the sockets of the slaves
* that are currently in SLAVE_STATE_WAIT_BGSAVE_START state. */ * that are currently in SLAVE_STATE_WAIT_BGSAVE_START state. */
int rdbSaveToSlavesSockets(void) { int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
int *fds; int *fds;
uint64_t *clientids; uint64_t *clientids;
int numfds; int numfds;
...@@ -1779,7 +1794,7 @@ int rdbSaveToSlavesSockets(void) { ...@@ -1779,7 +1794,7 @@ int rdbSaveToSlavesSockets(void) {
closeListeningSockets(0); closeListeningSockets(0);
redisSetProcTitle("redis-rdb-to-slaves"); redisSetProcTitle("redis-rdb-to-slaves");
retval = rdbSaveRioWithEOFMark(&slave_sockets,NULL); retval = rdbSaveRioWithEOFMark(&slave_sockets,NULL,rsi);
if (retval == C_OK && rioFlush(&slave_sockets) == 0) if (retval == C_OK && rioFlush(&slave_sockets) == 0)
retval = C_ERR; retval = C_ERR;
...@@ -1884,7 +1899,7 @@ void saveCommand(client *c) { ...@@ -1884,7 +1899,7 @@ void saveCommand(client *c) {
addReplyError(c,"Background save already in progress"); addReplyError(c,"Background save already in progress");
return; return;
} }
if (rdbSave(server.rdb_filename) == C_OK) { if (rdbSave(server.rdb_filename,NULL) == C_OK) {
addReply(c,shared.ok); addReply(c,shared.ok);
} else { } else {
addReply(c,shared.err); addReply(c,shared.err);
...@@ -1918,7 +1933,7 @@ void bgsaveCommand(client *c) { ...@@ -1918,7 +1933,7 @@ void bgsaveCommand(client *c) {
"Use BGSAVE SCHEDULE in order to schedule a BGSAVE whenver " "Use BGSAVE SCHEDULE in order to schedule a BGSAVE whenver "
"possible."); "possible.");
} }
} else if (rdbSaveBackground(server.rdb_filename) == C_OK) { } else if (rdbSaveBackground(server.rdb_filename,NULL) == C_OK) {
addReplyStatus(c,"Background saving started"); addReplyStatus(c,"Background saving started");
} else { } else {
addReply(c,shared.err); addReply(c,shared.err);
......
...@@ -118,11 +118,11 @@ uint64_t rdbLoadLen(rio *rdb, int *isencoded); ...@@ -118,11 +118,11 @@ uint64_t rdbLoadLen(rio *rdb, int *isencoded);
int rdbLoadLenByRef(rio *rdb, int *isencoded, uint64_t *lenptr); int rdbLoadLenByRef(rio *rdb, int *isencoded, uint64_t *lenptr);
int rdbSaveObjectType(rio *rdb, robj *o); int rdbSaveObjectType(rio *rdb, robj *o);
int rdbLoadObjectType(rio *rdb); int rdbLoadObjectType(rio *rdb);
int rdbLoad(char *filename); int rdbLoad(char *filename, rdbSaveInfo *rsi);
int rdbSaveBackground(char *filename); int rdbSaveBackground(char *filename, rdbSaveInfo *rsi);
int rdbSaveToSlavesSockets(void); int rdbSaveToSlavesSockets(rdbSaveInfo *rsi);
void rdbRemoveTempFile(pid_t childpid); void rdbRemoveTempFile(pid_t childpid);
int rdbSave(char *filename); int rdbSave(char *filename, rdbSaveInfo *rsi);
ssize_t rdbSaveObject(rio *rdb, robj *o); ssize_t rdbSaveObject(rio *rdb, robj *o);
size_t rdbSavedObjectLen(robj *o); size_t rdbSavedObjectLen(robj *o);
robj *rdbLoadObject(int type, rio *rdb); robj *rdbLoadObject(int type, rio *rdb);
...@@ -136,6 +136,6 @@ int rdbSaveBinaryDoubleValue(rio *rdb, double val); ...@@ -136,6 +136,6 @@ int rdbSaveBinaryDoubleValue(rio *rdb, double val);
int rdbLoadBinaryDoubleValue(rio *rdb, double *val); int rdbLoadBinaryDoubleValue(rio *rdb, double *val);
int rdbSaveBinaryFloatValue(rio *rdb, float val); int rdbSaveBinaryFloatValue(rio *rdb, float val);
int rdbLoadBinaryFloatValue(rio *rdb, float *val); int rdbLoadBinaryFloatValue(rio *rdb, float *val);
int rdbLoadRio(rio *rdb); int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi);
#endif #endif
This diff is collapsed.
...@@ -1079,7 +1079,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { ...@@ -1079,7 +1079,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
{ {
serverLog(LL_NOTICE,"%d changes in %d seconds. Saving...", serverLog(LL_NOTICE,"%d changes in %d seconds. Saving...",
sp->changes, (int)sp->seconds); sp->changes, (int)sp->seconds);
rdbSaveBackground(server.rdb_filename); rdbSaveBackground(server.rdb_filename,NULL);
break; break;
} }
} }
...@@ -1151,7 +1151,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { ...@@ -1151,7 +1151,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
(server.unixtime-server.lastbgsave_try > CONFIG_BGSAVE_RETRY_DELAY || (server.unixtime-server.lastbgsave_try > CONFIG_BGSAVE_RETRY_DELAY ||
server.lastbgsave_status == C_OK)) server.lastbgsave_status == C_OK))
{ {
if (rdbSaveBackground(server.rdb_filename) == C_OK) if (rdbSaveBackground(server.rdb_filename,NULL) == C_OK)
server.rdb_bgsave_scheduled = 0; server.rdb_bgsave_scheduled = 0;
} }
...@@ -1309,10 +1309,11 @@ void initServerConfig(void) { ...@@ -1309,10 +1309,11 @@ void initServerConfig(void) {
int j; int j;
getRandomHexChars(server.runid,CONFIG_RUN_ID_SIZE); getRandomHexChars(server.runid,CONFIG_RUN_ID_SIZE);
server.runid[CONFIG_RUN_ID_SIZE] = '\0';
changeReplicationId();
server.configfile = NULL; server.configfile = NULL;
server.executable = NULL; server.executable = NULL;
server.hz = CONFIG_DEFAULT_HZ; server.hz = CONFIG_DEFAULT_HZ;
server.runid[CONFIG_RUN_ID_SIZE] = '\0';
server.arch_bits = (sizeof(long) == 8) ? 64 : 32; server.arch_bits = (sizeof(long) == 8) ? 64 : 32;
server.port = CONFIG_DEFAULT_SERVER_PORT; server.port = CONFIG_DEFAULT_SERVER_PORT;
server.tcp_backlog = CONFIG_DEFAULT_TCP_BACKLOG; server.tcp_backlog = CONFIG_DEFAULT_TCP_BACKLOG;
...@@ -1409,7 +1410,7 @@ void initServerConfig(void) { ...@@ -1409,7 +1410,7 @@ void initServerConfig(void) {
server.masterport = 6379; server.masterport = 6379;
server.master = NULL; server.master = NULL;
server.cached_master = NULL; server.cached_master = NULL;
server.repl_master_initial_offset = -1; server.master_initial_offset = -1;
server.repl_state = REPL_STATE_NONE; server.repl_state = REPL_STATE_NONE;
server.repl_syncio_timeout = CONFIG_REPL_SYNCIO_TIMEOUT; server.repl_syncio_timeout = CONFIG_REPL_SYNCIO_TIMEOUT;
server.repl_serve_stale_data = CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA; server.repl_serve_stale_data = CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA;
...@@ -2471,7 +2472,7 @@ int prepareForShutdown(int flags) { ...@@ -2471,7 +2472,7 @@ int prepareForShutdown(int flags) {
if ((server.saveparamslen > 0 && !nosave) || save) { if ((server.saveparamslen > 0 && !nosave) || save) {
serverLog(LL_NOTICE,"Saving the final RDB snapshot before exiting."); serverLog(LL_NOTICE,"Saving the final RDB snapshot before exiting.");
/* Snapshotting. Perform a SYNC SAVE and exit */ /* Snapshotting. Perform a SYNC SAVE and exit */
if (rdbSave(server.rdb_filename) != C_OK) { if (rdbSave(server.rdb_filename,NULL) != C_OK) {
/* Ooops.. error saving! The best we can do is to continue /* Ooops.. error saving! The best we can do is to continue
* operating. Note that if there was a background saving process, * operating. Note that if there was a background saving process,
* in the next cron() Redis will be notified that the background * in the next cron() Redis will be notified that the background
...@@ -3135,12 +3136,18 @@ sds genRedisInfoString(char *section) { ...@@ -3135,12 +3136,18 @@ sds genRedisInfoString(char *section) {
} }
} }
info = sdscatprintf(info, info = sdscatprintf(info,
"master_replid:%s\r\n"
"master_replid2:%s\r\n"
"master_repl_offset:%lld\r\n" "master_repl_offset:%lld\r\n"
"second_repl_offset:%lld\r\n"
"repl_backlog_active:%d\r\n" "repl_backlog_active:%d\r\n"
"repl_backlog_size:%lld\r\n" "repl_backlog_size:%lld\r\n"
"repl_backlog_first_byte_offset:%lld\r\n" "repl_backlog_first_byte_offset:%lld\r\n"
"repl_backlog_histlen:%lld\r\n", "repl_backlog_histlen:%lld\r\n",
server.replid,
server.replid2,
server.master_repl_offset, server.master_repl_offset,
server.second_replid_offset,
server.repl_backlog != NULL, server.repl_backlog != NULL,
server.repl_backlog_size, server.repl_backlog_size,
server.repl_backlog_off, server.repl_backlog_off,
...@@ -3416,7 +3423,7 @@ void loadDataFromDisk(void) { ...@@ -3416,7 +3423,7 @@ void loadDataFromDisk(void) {
if (loadAppendOnlyFile(server.aof_filename) == C_OK) if (loadAppendOnlyFile(server.aof_filename) == C_OK)
serverLog(LL_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000); serverLog(LL_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000);
} else { } else {
if (rdbLoad(server.rdb_filename) == C_OK) { if (rdbLoad(server.rdb_filename,NULL) == C_OK) {
serverLog(LL_NOTICE,"DB loaded from disk: %.3f seconds", serverLog(LL_NOTICE,"DB loaded from disk: %.3f seconds",
(float)(ustime()-start)/1000000); (float)(ustime()-start)/1000000);
} else if (errno != ENOENT) { } else if (errno != ENOENT) {
......
...@@ -293,7 +293,8 @@ typedef long long mstime_t; /* millisecond time type. */ ...@@ -293,7 +293,8 @@ typedef long long mstime_t; /* millisecond time type. */
/* Slave capabilities. */ /* Slave capabilities. */
#define SLAVE_CAPA_NONE 0 #define SLAVE_CAPA_NONE 0
#define SLAVE_CAPA_EOF (1<<0) /* Can parse the RDB EOF streaming format. */ #define SLAVE_CAPA_EOF (1<<0) /* Can parse the RDB EOF streaming format. */
#define SLAVE_CAPA_PSYNC2 (1<<1) /* Supports PSYNC2 protocol. */
/* Synchronous read timeout - slave side */ /* Synchronous read timeout - slave side */
#define CONFIG_REPL_SYNCIO_TIMEOUT 5 #define CONFIG_REPL_SYNCIO_TIMEOUT 5
...@@ -679,8 +680,8 @@ typedef struct client { ...@@ -679,8 +680,8 @@ typedef struct client {
long long psync_initial_offset; /* FULLRESYNC reply offset other slaves long long psync_initial_offset; /* FULLRESYNC reply offset other slaves
copying this slave output buffer copying this slave output buffer
should use. */ should use. */
char replrunid[CONFIG_RUN_ID_SIZE+1]; /* Master run id if is a master. */ char replid[CONFIG_RUN_ID_SIZE+1]; /* Master replication ID (if master). */
int slave_listening_port; /* As configured with: REPLCONF listening-port */ int slave_listening_port; /* As configured with: SLAVECONF listening-port */
char slave_ip[NET_IP_STR_LEN]; /* Optionally given by REPLCONF ip-address */ char slave_ip[NET_IP_STR_LEN]; /* Optionally given by REPLCONF ip-address */
int slave_capa; /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */ int slave_capa; /* Slave capabilities: SLAVE_CAPA_* bitwise OR. */
multiState mstate; /* MULTI/EXEC state */ multiState mstate; /* MULTI/EXEC state */
...@@ -803,6 +804,20 @@ struct redisMemOverhead { ...@@ -803,6 +804,20 @@ struct redisMemOverhead {
} *db; } *db;
}; };
/* This structure can be optionally passed to RDB save/load functions in
* order to implement additional functionalities, by storing and loading
* metadata to the RDB file.
*
* Currently the only use is to select a DB at load time, useful in
* replication in order to make sure that chained slaves (slaves of slaves)
* select the correct DB and are able to accept the stream coming from the
* top-level master. */
typedef struct rdbSaveInfo {
int repl_stream_db; /* DB to select in server.master client. */
} rdbSaveInfo;
#define RDB_SAVE_INFO_INIT {-1}
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
* Global server state * Global server state
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
...@@ -988,15 +1003,19 @@ struct redisServer { ...@@ -988,15 +1003,19 @@ struct redisServer {
char *syslog_ident; /* Syslog ident */ char *syslog_ident; /* Syslog ident */
int syslog_facility; /* Syslog facility */ int syslog_facility; /* Syslog facility */
/* Replication (master) */ /* Replication (master) */
char replid[CONFIG_RUN_ID_SIZE+1]; /* My current replication ID. */
char replid2[CONFIG_RUN_ID_SIZE+1]; /* replid inherited from master*/
long long master_repl_offset; /* My current replication offset */
long long second_replid_offset; /* Accept offsets up to this for replid2. */
int slaveseldb; /* Last SELECTed DB in replication output */ int slaveseldb; /* Last SELECTed DB in replication output */
long long master_repl_offset; /* Global replication offset */
int repl_ping_slave_period; /* Master pings the slave every N seconds */ int repl_ping_slave_period; /* Master pings the slave every N seconds */
char *repl_backlog; /* Replication backlog for partial syncs */ char *repl_backlog; /* Replication backlog for partial syncs */
long long repl_backlog_size; /* Backlog circular buffer size */ long long repl_backlog_size; /* Backlog circular buffer size */
long long repl_backlog_histlen; /* Backlog actual data length */ long long repl_backlog_histlen; /* Backlog actual data length */
long long repl_backlog_idx; /* Backlog circular buffer current offset */ long long repl_backlog_idx; /* Backlog circular buffer current offset,
long long repl_backlog_off; /* Replication offset of first byte in the that is the next byte will'll write to.*/
backlog buffer. */ long long repl_backlog_off; /* Replication "master offset" of first
byte in the replication backlog buffer.*/
time_t repl_backlog_time_limit; /* Time without slaves after the backlog time_t repl_backlog_time_limit; /* Time without slaves after the backlog
gets released. */ gets released. */
time_t repl_no_slaves_since; /* We have no slaves since that time. time_t repl_no_slaves_since; /* We have no slaves since that time.
...@@ -1029,8 +1048,11 @@ struct redisServer { ...@@ -1029,8 +1048,11 @@ struct redisServer {
int slave_priority; /* Reported in INFO and used by Sentinel. */ int slave_priority; /* Reported in INFO and used by Sentinel. */
int slave_announce_port; /* Give the master this listening port. */ int slave_announce_port; /* Give the master this listening port. */
char *slave_announce_ip; /* Give the master this ip address. */ char *slave_announce_ip; /* Give the master this ip address. */
char repl_master_runid[CONFIG_RUN_ID_SIZE+1]; /* Master run id for PSYNC.*/ /* The following two fields is where we store master PSYNC replid/offset
long long repl_master_initial_offset; /* Master PSYNC offset. */ * while the PSYNC is in progress. At the end we'll copy the fields into
* the server->master client structure. */
char master_replid[CONFIG_RUN_ID_SIZE+1]; /* Master PSYNC runid. */
long long master_initial_offset; /* Master PSYNC offset. */
int repl_slave_lazy_flush; /* Lazy FLUSHALL before loading DB? */ int repl_slave_lazy_flush; /* Lazy FLUSHALL before loading DB? */
/* Replication script cache. */ /* Replication script cache. */
dict *repl_scriptcache_dict; /* SHA1 all slaves are aware of. */ dict *repl_scriptcache_dict; /* SHA1 all slaves are aware of. */
...@@ -1259,6 +1281,7 @@ void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask); ...@@ -1259,6 +1281,7 @@ void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask); void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask); void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask); void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask);
void addReplyString(client *c, const char *s, size_t len);
void addReplyBulk(client *c, robj *obj); void addReplyBulk(client *c, robj *obj);
void addReplyBulkCString(client *c, const char *s); void addReplyBulkCString(client *c, const char *s);
void addReplyBulkCBuffer(client *c, const void *p, size_t len); void addReplyBulkCBuffer(client *c, const void *p, size_t len);
...@@ -1393,6 +1416,7 @@ ssize_t syncReadLine(int fd, char *ptr, ssize_t size, long long timeout); ...@@ -1393,6 +1416,7 @@ ssize_t syncReadLine(int fd, char *ptr, ssize_t size, long long timeout);
/* Replication */ /* Replication */
void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc); void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc);
void replicationFeedSlavesFromMasterStream(list *slaves, char *buf, size_t buflen);
void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv, int argc); void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv, int argc);
void updateSlavesWaitingBgsave(int bgsaveerr, int type); void updateSlavesWaitingBgsave(int bgsaveerr, int type);
void replicationCron(void); void replicationCron(void);
...@@ -1414,6 +1438,9 @@ long long replicationGetSlaveOffset(void); ...@@ -1414,6 +1438,9 @@ long long replicationGetSlaveOffset(void);
char *replicationGetSlaveName(client *c); char *replicationGetSlaveName(client *c);
long long getPsyncInitialOffset(void); long long getPsyncInitialOffset(void);
int replicationSetupSlaveForFullResync(client *slave, long long offset); int replicationSetupSlaveForFullResync(client *slave, long long offset);
void changeReplicationId(void);
void clearReplicationId2(void);
void chopReplicationBacklog(void);
/* Generic persistence functions */ /* Generic persistence functions */
void startLoading(FILE *fp); void startLoading(FILE *fp);
...@@ -1422,7 +1449,7 @@ void stopLoading(void); ...@@ -1422,7 +1449,7 @@ void stopLoading(void);
/* RDB persistence */ /* RDB persistence */
#include "rdb.h" #include "rdb.h"
int rdbSaveRio(rio *rdb, int *error, int flags); int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi);
/* AOF persistence */ /* AOF persistence */
void flushAppendOnlyFile(int force); void flushAppendOnlyFile(int force);
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment