1 #include "redis.h" 2 3 #include <sys/time.h> 4 #include <unistd.h> 5 #include <fcntl.h> 6 #include <sys/stat.h> 7 8 /* ---------------------------------- MASTER -------------------------------- */ 9 10 void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) { 11 listNode *ln; 12 listIter li; 13 int j; 14 15 listRewind(slaves,&li); 16 while((ln = listNext(&li))) { 17 redisClient *slave = ln->value; 18 19 /* Don't feed slaves that are still waiting for BGSAVE to start */ 20 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue; 21 22 /* Feed slaves that are waiting for the initial SYNC (so these commands 23 * are queued in the output buffer until the intial SYNC completes), 24 * or are already in sync with the master. */ 25 if (slave->slaveseldb != dictid) { 26 robj *selectcmd; 27 28 switch(dictid) { 29 case 0: selectcmd = shared.select0; break; 30 case 1: selectcmd = shared.select1; break; 31 case 2: selectcmd = shared.select2; break; 32 case 3: selectcmd = shared.select3; break; 33 case 4: selectcmd = shared.select4; break; 34 case 5: selectcmd = shared.select5; break; 35 case 6: selectcmd = shared.select6; break; 36 case 7: selectcmd = shared.select7; break; 37 case 8: selectcmd = shared.select8; break; 38 case 9: selectcmd = shared.select9; break; 39 default: 40 selectcmd = createObject(REDIS_STRING, 41 sdscatprintf(sdsempty(),"select %d\r\n",dictid)); 42 selectcmd->refcount = 0; 43 break; 44 } 45 addReply(slave,selectcmd); 46 slave->slaveseldb = dictid; 47 } 48 addReplyMultiBulkLen(slave,argc); 49 for (j = 0; j < argc; j++) addReplyBulk(slave,argv[j]); 50 } 51 } 52 53 void replicationFeedMonitors(redisClient *c, list *monitors, int dictid, robj **argv, int argc) { 54 listNode *ln; 55 listIter li; 56 int j, port; 57 sds cmdrepr = sdsnew("+"); 58 robj *cmdobj; 59 char ip[32]; 60 struct timeval tv; 61 62 gettimeofday(&tv,NULL); 63 cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec); 64 if (c->flags & REDIS_LUA_CLIENT) { 65 cmdrepr = sdscatprintf(cmdrepr,"[%d lua] ", dictid); 66 } else { 67 anetPeerToString(c->fd,ip,&port); 68 cmdrepr = sdscatprintf(cmdrepr,"[%d %s:%d] ", dictid,ip,port); 69 } 70 71 for (j = 0; j < argc; j++) { 72 if (argv[j]->encoding == REDIS_ENCODING_INT) { 73 cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)argv[j]->ptr); 74 } else { 75 cmdrepr = sdscatrepr(cmdrepr,(char*)argv[j]->ptr, 76 sdslen(argv[j]->ptr)); 77 } 78 if (j != argc-1) 79 cmdrepr = sdscatlen(cmdrepr," ",1); 80 } 81 cmdrepr = sdscatlen(cmdrepr,"\r\n",2); 82 cmdobj = createObject(REDIS_STRING,cmdrepr); 83 84 listRewind(monitors,&li); 85 while((ln = listNext(&li))) { 86 redisClient *monitor = ln->value; 87 addReply(monitor,cmdobj); 88 } 89 decrRefCount(cmdobj); 90 } 91 92 void syncCommand(redisClient *c) { 93 /* ignore SYNC if aleady slave or in monitor mode */ 94 if (c->flags & REDIS_SLAVE) return; 95 96 /* Refuse SYNC requests if we are a slave but the link with our master 97 * is not ok... */ 98 if (server.masterhost && server.repl_state != REDIS_REPL_CONNECTED) { 99 addReplyError(c,"Can't SYNC while not connected with my master"); 100 return; 101 } 102 103 /* SYNC can't be issued when the server has pending data to send to 104 * the client about already issued commands. We need a fresh reply 105 * buffer registering the differences between the BGSAVE and the current 106 * dataset, so that we can copy to other slaves if needed. */ 107 if (listLength(c->reply) != 0) { 108 addReplyError(c,"SYNC is invalid with pending input"); 109 return; 110 } 111 112 redisLog(REDIS_NOTICE,"Slave ask for synchronization"); 113 /* Here we need to check if there is a background saving operation 114 * in progress, or if it is required to start one */ 115 if (server.rdb_child_pid != -1) { 116 /* Ok a background save is in progress. Let's check if it is a good 117 * one for replication, i.e. if there is another slave that is 118 * registering differences since the server forked to save */ 119 redisClient *slave; 120 listNode *ln; 121 listIter li; 122 123 listRewind(server.slaves,&li); 124 while((ln = listNext(&li))) { 125 slave = ln->value; 126 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break; 127 } 128 if (ln) { 129 /* Perfect, the server is already registering differences for 130 * another slave. Set the right state, and copy the buffer. */ 131 copyClientOutputBuffer(c,slave); 132 c->replstate = REDIS_REPL_WAIT_BGSAVE_END; 133 redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC"); 134 } else { 135 /* No way, we need to wait for the next BGSAVE in order to 136 * register differences */ 137 c->replstate = REDIS_REPL_WAIT_BGSAVE_START; 138 redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC"); 139 } 140 } else { 141 /* Ok we don't have a BGSAVE in progress, let's start one */ 142 redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC"); 143 if (rdbSaveBackground(server.rdb_filename) != REDIS_OK) { 144 redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE"); 145 addReplyError(c,"Unable to perform background save"); 146 return; 147 } 148 c->replstate = REDIS_REPL_WAIT_BGSAVE_END; 149 } 150 c->repldbfd = -1; 151 c->flags |= REDIS_SLAVE; 152 c->slaveseldb = 0; 153 listAddNodeTail(server.slaves,c); 154 return; 155 } 156 157 void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) { 158 redisClient *slave = privdata; 159 REDIS_NOTUSED(el); 160 REDIS_NOTUSED(mask); 161 char buf[REDIS_IOBUF_LEN]; 162 ssize_t nwritten, buflen; 163 164 if (slave->repldboff == 0) { 165 /* Write the bulk write count before to transfer the DB. In theory here 166 * we don't know how much room there is in the output buffer of the 167 * socket, but in pratice SO_SNDLOWAT (the minimum count for output 168 * operations) will never be smaller than the few bytes we need. */ 169 sds bulkcount; 170 171 bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long) 172 slave->repldbsize); 173 if (write(fd,bulkcount,sdslen(bulkcount)) != (signed)sdslen(bulkcount)) 174 { 175 sdsfree(bulkcount); 176 freeClient(slave); 177 return; 178 } 179 sdsfree(bulkcount); 180 } 181 lseek(slave->repldbfd,slave->repldboff,SEEK_SET); 182 buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN); 183 if (buflen <= 0) { 184 redisLog(REDIS_WARNING,"Read error sending DB to slave: %s", 185 (buflen == 0) ? "premature EOF" : strerror(errno)); 186 freeClient(slave); 187 return; 188 } 189 if ((nwritten = write(fd,buf,buflen)) == -1) { 190 redisLog(REDIS_VERBOSE,"Write error sending DB to slave: %s", 191 strerror(errno)); 192 freeClient(slave); 193 return; 194 } 195 slave->repldboff += nwritten; 196 if (slave->repldboff == slave->repldbsize) { 197 close(slave->repldbfd); 198 slave->repldbfd = -1; 199 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE); 200 slave->replstate = REDIS_REPL_ONLINE; 201 if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, 202 sendReplyToClient, slave) == AE_ERR) { 203 freeClient(slave); 204 return; 205 } 206 addReplySds(slave,sdsempty()); 207 redisLog(REDIS_NOTICE,"Synchronization with slave succeeded"); 208 } 209 } 210 211 /* This function is called at the end of every backgrond saving. 212 * The argument bgsaveerr is REDIS_OK if the background saving succeeded 213 * otherwise REDIS_ERR is passed to the function. 214 * 215 * The goal of this function is to handle slaves waiting for a successful 216 * background saving in order to perform non-blocking synchronization. */ 217 void updateSlavesWaitingBgsave(int bgsaveerr) { 218 listNode *ln; 219 int startbgsave = 0; 220 listIter li; 221 222 listRewind(server.slaves,&li); 223 while((ln = listNext(&li))) { 224 redisClient *slave = ln->value; 225 226 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) { 227 startbgsave = 1; 228 slave->replstate = REDIS_REPL_WAIT_BGSAVE_END; 229 } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) { 230 struct redis_stat buf; 231 232 if (bgsaveerr != REDIS_OK) { 233 freeClient(slave); 234 redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error"); 235 continue; 236 } 237 if ((slave->repldbfd = open(server.rdb_filename,O_RDONLY)) == -1 || 238 redis_fstat(slave->repldbfd,&buf) == -1) { 239 freeClient(slave); 240 redisLog(REDIS_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno)); 241 continue; 242 } 243 slave->repldboff = 0; 244 slave->repldbsize = buf.st_size; 245 slave->replstate = REDIS_REPL_SEND_BULK; 246 aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE); 247 if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) { 248 freeClient(slave); 249 continue; 250 } 251 } 252 } 253 if (startbgsave) { 254 if (rdbSaveBackground(server.rdb_filename) != REDIS_OK) { 255 listIter li; 256 257 listRewind(server.slaves,&li); 258 redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed"); 259 while((ln = listNext(&li))) { 260 redisClient *slave = ln->value; 261 262 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) 263 freeClient(slave); 264 } 265 } 266 } 267 } 268 269 /* ----------------------------------- SLAVE -------------------------------- */ 270 271 /* Abort the async download of the bulk dataset while SYNC-ing with master */ 272 void replicationAbortSyncTransfer(void) { 273 redisAssert(server.repl_state == REDIS_REPL_TRANSFER); 274 275 aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE); 276 close(server.repl_transfer_s); 277 close(server.repl_transfer_fd); 278 unlink(server.repl_transfer_tmpfile); 279 zfree(server.repl_transfer_tmpfile); 280 server.repl_state = REDIS_REPL_CONNECT; 281 } 282 283 /* Asynchronously read the SYNC payload we receive from a master */ 284 void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) { 285 char buf[4096]; 286 ssize_t nread, readlen; 287 REDIS_NOTUSED(el); 288 REDIS_NOTUSED(privdata); 289 REDIS_NOTUSED(mask); 290 291 /* If repl_transfer_left == -1 we still have to read the bulk length 292 * from the master reply. */ 293 if (server.repl_transfer_left == -1) { 294 if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout) == -1) { 295 redisLog(REDIS_WARNING, 296 "I/O error reading bulk count from MASTER: %s", 297 strerror(errno)); 298 goto error; 299 } 300 301 if (buf[0] == '-') { 302 redisLog(REDIS_WARNING, 303 "MASTER aborted replication with an error: %s", 304 buf+1); 305 goto error; 306 } else if (buf[0] == '\0') { 307 /* At this stage just a newline works as a PING in order to take 308 * the connection live. So we refresh our last interaction 309 * timestamp. */ 310 server.repl_transfer_lastio = time(NULL); 311 return; 312 } else if (buf[0] != '$') { 313 redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?"); 314 goto error; 315 } 316 server.repl_transfer_left = strtol(buf+1,NULL,10); 317 redisLog(REDIS_NOTICE, 318 "MASTER <-> SLAVE sync: receiving %ld bytes from master", 319 server.repl_transfer_left); 320 return; 321 } 322 323 /* Read bulk data */ 324 readlen = (server.repl_transfer_left < (signed)sizeof(buf)) ? 325 server.repl_transfer_left : (signed)sizeof(buf); 326 nread = read(fd,buf,readlen); 327 if (nread <= 0) { 328 redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s", 329 (nread == -1) ? strerror(errno) : "connection lost"); 330 replicationAbortSyncTransfer(); 331 return; 332 } 333 server.repl_transfer_lastio = time(NULL); 334 if (write(server.repl_transfer_fd,buf,nread) != nread) { 335 redisLog(REDIS_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno)); 336 goto error; 337 } 338 server.repl_transfer_left -= nread; 339 /* Check if the transfer is now complete */ 340 if (server.repl_transfer_left == 0) { 341 if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) { 342 redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno)); 343 replicationAbortSyncTransfer(); 344 return; 345 } 346 redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory"); 347 emptyDb(); 348 /* Before loading the DB into memory we need to delete the readable 349 * handler, otherwise it will get called recursively since 350 * rdbLoad() will call the event loop to process events from time to 351 * time for non blocking loading. */ 352 aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE); 353 if (rdbLoad(server.rdb_filename) != REDIS_OK) { 354 redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk"); 355 replicationAbortSyncTransfer(); 356 return; 357 } 358 /* Final setup of the connected slave <- master link */ 359 zfree(server.repl_transfer_tmpfile); 360 close(server.repl_transfer_fd); 361 server.master = createClient(server.repl_transfer_s); 362 server.master->flags |= REDIS_MASTER; 363 server.master->authenticated = 1; 364 server.repl_state = REDIS_REPL_CONNECTED; 365 redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success"); 366 /* Restart the AOF subsystem now that we finished the sync. This 367 * will trigger an AOF rewrite, and when done will start appending 368 * to the new file. */ 369 if (server.aof_state != REDIS_AOF_OFF) { 370 int retry = 10; 371 372 stopAppendOnly(); 373 while (retry-- && startAppendOnly() == REDIS_ERR) { 374 redisLog(REDIS_WARNING,"Failed enabling the AOF after successful master synchrnization! Trying it again in one second."); 375 sleep(1); 376 } 377 if (!retry) { 378 redisLog(REDIS_WARNING,"FATAL: this slave instance finished the synchronization with its master, but the AOF can't be turned on. Exiting now."); 379 exit(1); 380 } 381 } 382 } 383 384 return; 385 386 error: 387 replicationAbortSyncTransfer(); 388 return; 389 } 390 391 void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) { 392 char buf[1024], tmpfile[256]; 393 int dfd, maxtries = 5; 394 REDIS_NOTUSED(el); 395 REDIS_NOTUSED(privdata); 396 REDIS_NOTUSED(mask); 397 398 /* If this event fired after the user turned the instance into a master 399 * with SLAVEOF NO ONE we must just return ASAP. */ 400 if (server.repl_state == REDIS_REPL_NONE) { 401 close(fd); 402 return; 403 } 404 405 redisLog(REDIS_NOTICE,"Non blocking connect for SYNC fired the event."); 406 /* This event should only be triggered once since it is used to have a 407 * non-blocking connect(2) to the master. It has been triggered when this 408 * function is called, so we can delete it. */ 409 aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE); 410 411 /* AUTH with the master if required. */ 412 if(server.masterauth) { 413 char authcmd[1024]; 414 size_t authlen; 415 416 authlen = snprintf(authcmd,sizeof(authcmd),"AUTH %s\r\n",server.masterauth); 417 if (syncWrite(fd,authcmd,authlen,server.repl_syncio_timeout) == -1) { 418 redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s", 419 strerror(errno)); 420 goto error; 421 } 422 /* Read the AUTH result. */ 423 if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout) == -1) { 424 redisLog(REDIS_WARNING,"I/O error reading auth result from MASTER: %s", 425 strerror(errno)); 426 goto error; 427 } 428 if (buf[0] != '+') { 429 redisLog(REDIS_WARNING,"Cannot AUTH to MASTER, is the masterauth password correct?"); 430 goto error; 431 } 432 } 433 434 /* Issue the SYNC command */ 435 if (syncWrite(fd,"SYNC \r\n",7,server.repl_syncio_timeout) == -1) { 436 redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s", 437 strerror(errno)); 438 goto error; 439 } 440 441 /* Prepare a suitable temp file for bulk transfer */ 442 while(maxtries--) { 443 snprintf(tmpfile,256, 444 "temp-%d.%ld.rdb",(int)time(NULL),(long int)getpid()); 445 dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644); 446 if (dfd != -1) break; 447 sleep(1); 448 } 449 if (dfd == -1) { 450 redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno)); 451 goto error; 452 } 453 454 /* Setup the non blocking download of the bulk file. */ 455 if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL) 456 == AE_ERR) 457 { 458 redisLog(REDIS_WARNING,"Can't create readable event for SYNC"); 459 goto error; 460 } 461 462 server.repl_state = REDIS_REPL_TRANSFER; 463 server.repl_transfer_left = -1; 464 server.repl_transfer_fd = dfd; 465 server.repl_transfer_lastio = time(NULL); 466 server.repl_transfer_tmpfile = zstrdup(tmpfile); 467 return; 468 469 error: 470 server.repl_state = REDIS_REPL_CONNECT; 471 close(fd); 472 return; 473 } 474 475 int connectWithMaster(void) { 476 int fd; 477 478 fd = anetTcpNonBlockConnect(NULL,server.masterhost,server.masterport); 479 if (fd == -1) { 480 redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s", 481 strerror(errno)); 482 return REDIS_ERR; 483 } 484 485 if (aeCreateFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE,syncWithMaster,NULL) == 486 AE_ERR) 487 { 488 close(fd); 489 redisLog(REDIS_WARNING,"Can't create readable event for SYNC"); 490 return REDIS_ERR; 491 } 492 493 server.repl_transfer_lastio = time(NULL); 494 server.repl_transfer_s = fd; 495 server.repl_state = REDIS_REPL_CONNECTING; 496 return REDIS_OK; 497 } 498 499 /* This function can be called when a non blocking connection is currently 500 * in progress to undo it. */ 501 void undoConnectWithMaster(void) { 502 int fd = server.repl_transfer_s; 503 504 redisAssert(server.repl_state == REDIS_REPL_CONNECTING); 505 aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE); 506 close(fd); 507 server.repl_transfer_s = -1; 508 server.repl_state = REDIS_REPL_CONNECT; 509 } 510 511 void slaveofCommand(redisClient *c) { 512 if (!strcasecmp(c->argv[1]->ptr,"no") && 513 !strcasecmp(c->argv[2]->ptr,"one")) { 514 if (server.masterhost) { 515 sdsfree(server.masterhost); 516 server.masterhost = NULL; 517 if (server.master) freeClient(server.master); 518 if (server.repl_state == REDIS_REPL_TRANSFER) 519 replicationAbortSyncTransfer(); 520 else if (server.repl_state == REDIS_REPL_CONNECTING) 521 undoConnectWithMaster(); 522 server.repl_state = REDIS_REPL_NONE; 523 redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)"); 524 } 525 } else { 526 long port; 527 528 if ((getLongFromObjectOrReply(c, c->argv[2], &port, NULL) != REDIS_OK)) 529 return; 530 531 /* Check if we are already attached to the specified slave */ 532 if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr) 533 && server.masterport == port) { 534 redisLog(REDIS_NOTICE,"SLAVE OF would result into synchronization with the master we are already connected with. No operation performed."); 535 addReplySds(c,sdsnew("+OK Already connected to specified master\r\n")); 536 return; 537 } 538 /* There was no previous master or the user specified a different one, 539 * we can continue. */ 540 sdsfree(server.masterhost); 541 server.masterhost = sdsdup(c->argv[1]->ptr); 542 server.masterport = port; 543 if (server.master) freeClient(server.master); 544 if (server.repl_state == REDIS_REPL_TRANSFER) 545 replicationAbortSyncTransfer(); 546 server.repl_state = REDIS_REPL_CONNECT; 547 redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)", 548 server.masterhost, server.masterport); 549 } 550 addReply(c,shared.ok); 551 } 552 553 /* --------------------------- REPLICATION CRON ---------------------------- */ 554 555 void replicationCron(void) { 556 /* Non blocking connection timeout? */ 557 if (server.masterhost && server.repl_state == REDIS_REPL_CONNECTING && 558 (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout) 559 { 560 redisLog(REDIS_WARNING,"Timeout connecting to the MASTER..."); 561 undoConnectWithMaster(); 562 } 563 564 /* Bulk transfer I/O timeout? */ 565 if (server.masterhost && server.repl_state == REDIS_REPL_TRANSFER && 566 (time(NULL)-server.repl_transfer_lastio) > server.repl_timeout) 567 { 568 redisLog(REDIS_WARNING,"Timeout receiving bulk data from MASTER..."); 569 replicationAbortSyncTransfer(); 570 } 571 572 /* Timed out master when we are an already connected slave? */ 573 if (server.masterhost && server.repl_state == REDIS_REPL_CONNECTED && 574 (time(NULL)-server.master->lastinteraction) > server.repl_timeout) 575 { 576 redisLog(REDIS_WARNING,"MASTER time out: no data nor PING received..."); 577 freeClient(server.master); 578 } 579 580 /* Check if we should connect to a MASTER */ 581 if (server.repl_state == REDIS_REPL_CONNECT) { 582 redisLog(REDIS_NOTICE,"Connecting to MASTER..."); 583 if (connectWithMaster() == REDIS_OK) { 584 redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync started"); 585 } 586 } 587 588 /* If we have attached slaves, PING them from time to time. 589 * So slaves can implement an explicit timeout to masters, and will 590 * be able to detect a link disconnection even if the TCP connection 591 * will not actually go down. */ 592 if (!(server.cronloops % (server.repl_ping_slave_period*10))) { 593 listIter li; 594 listNode *ln; 595 596 listRewind(server.slaves,&li); 597 while((ln = listNext(&li))) { 598 redisClient *slave = ln->value; 599 600 /* Don't ping slaves that are in the middle of a bulk transfer 601 * with the master for first synchronization. */ 602 if (slave->replstate == REDIS_REPL_SEND_BULK) continue; 603 if (slave->replstate == REDIS_REPL_ONLINE) { 604 /* If the slave is online send a normal ping */ 605 addReplySds(slave,sdsnew("*1\r\n$4\r\nPING\r\n")); 606 } else { 607 /* Otherwise we are in the pre-synchronization stage. 608 * Just a newline will do the work of refreshing the 609 * connection last interaction time, and at the same time 610 * we'll be sure that being a single char there are no 611 * short-write problems. */ 612 if (write(slave->fd, "\n", 1) == -1) { 613 /* Don't worry, it's just a ping. */ 614 } 615 } 616 } 617 } 618 } 619