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