src/replication.c

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