Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Browse files

Fixed issue #435 and at the same time introduced explicit ping in the…

… master-slave channel that will detect a blocked master or a broken even if apparently connected TCP link.
  • Loading branch information...
commit db0e263b428a33d458f4708eef9d9e9b974ce85a 1 parent d526d09
@antirez authored
Showing with 48 additions and 4 deletions.
  1. +0 −1  src/networking.c
  2. +2 −1  src/redis.c
  3. +46 −2 src/replication.c
View
1  src/networking.c
@@ -498,7 +498,6 @@ void freeClient(redisClient *c) {
/* Case 2: we lost the connection with the master. */
if (c->flags & REDIS_MASTER) {
server.master = NULL;
- /* FIXME */
server.replstate = REDIS_REPL_CONNECT;
/* Since we lost the connection with the master, we should also
* close the connection with all our slaves if we have any, so
View
3  src/redis.c
@@ -516,7 +516,7 @@ void updateLRUClock(void) {
}
int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
- int j, loops = server.cronloops++;
+ int j, loops = server.cronloops;
REDIS_NOTUSED(eventLoop);
REDIS_NOTUSED(id);
REDIS_NOTUSED(clientData);
@@ -645,6 +645,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
* to detect transfer failures. */
if (!(loops % 10)) replicationCron();
+ server.cronloops++;
return 100;
}
View
48 src/replication.c
@@ -328,6 +328,12 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
buf+1);
replicationAbortSyncTransfer();
return;
+ } else if (buf[0] == '\0') {
+ /* At this stage just a newline works as a PING in order to take
+ * the connection live. So we refresh our last interaction
+ * timestamp. */
+ server.repl_transfer_lastio = time(NULL);
+ return;
} 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?");
replicationAbortSyncTransfer();
@@ -488,17 +494,26 @@ void slaveofCommand(redisClient *c) {
/* --------------------------- REPLICATION CRON ---------------------------- */
-#define REDIS_REPL_TRANSFER_TIMEOUT 60
+#define REDIS_REPL_TIMEOUT 60
+#define REDIS_REPL_PING_SLAVE_PERIOD 10
void replicationCron(void) {
/* Bulk transfer I/O timeout? */
if (server.masterhost && server.replstate == REDIS_REPL_TRANSFER &&
- (time(NULL)-server.repl_transfer_lastio) > REDIS_REPL_TRANSFER_TIMEOUT)
+ (time(NULL)-server.repl_transfer_lastio) > REDIS_REPL_TIMEOUT)
{
redisLog(REDIS_WARNING,"Timeout receiving bulk data from MASTER...");
replicationAbortSyncTransfer();
}
+ /* Timed out master when we are an already connected slave? */
+ if (server.masterhost && server.replstate == REDIS_REPL_CONNECTED &&
+ (time(NULL)-server.master->lastinteraction) > REDIS_REPL_TIMEOUT)
+ {
+ redisLog(REDIS_WARNING,"MASTER time out: no data nor PING received...");
+ freeClient(server.master);
+ }
+
/* Check if we should connect to a MASTER */
if (server.replstate == REDIS_REPL_CONNECT) {
redisLog(REDIS_NOTICE,"Connecting to MASTER...");
@@ -507,4 +522,33 @@ void replicationCron(void) {
if (server.appendonly) rewriteAppendOnlyFileBackground();
}
}
+
+ /* If we have attached slaves, PING them from time to time.
+ * So slaves can implement an explicit timeout to masters, and will
+ * be able to detect a link disconnection even if the TCP connection
+ * will not actually go down. */
+ if (!(server.cronloops % (REDIS_REPL_PING_SLAVE_PERIOD*10))) {
+ listIter li;
+ listNode *ln;
+
+ listRewind(server.slaves,&li);
+ while((ln = listNext(&li))) {
+ redisClient *slave = ln->value;
+
+ /* Don't ping slaves that are in the middle of a bulk transfer
+ * with the master for first synchronization. */
+ if (slave->replstate == REDIS_REPL_SEND_BULK) continue;
+ if (slave->replstate == REDIS_REPL_ONLINE) {
+ /* If the slave is online send a normal ping */
+ addReplySds(slave,sdsnew("PING\r\n"));
+ } else {
+ /* Otherwise we are in the pre-synchronization stage.
+ * Just a newline will do the work of refreshing the
+ * connection last interaction time, and at the same time
+ * we'll be sure that being a single char there are no
+ * short-write problems. */
+ write(slave->fd, "\n", 1);
+ }
+ }
+ }
}
Please sign in to comment.
Something went wrong with that request. Please try again.