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.%ld ",(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     unsigned char buf[4096]
 310     size_t nread, readlen;
 311
 312     readlen = (server.repl_transfer_left < sizeof(buf)) ?
 313         server.repl_transfer_left : sizeof(buf);
 314     nread = read(fd,buf,readlen);
 315     if (nread <= 0) {
 316         redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
 317             (nread == -1) ? strerror(errno) : "connection lost");
 318         replicationAbortSyncTransfer();
 319         return;
 320     }
 321     server.repl_transfer_lastio = time(NULL);
 322     if (write(server.repl_transfer_fd,buf,nread) != nread) {
 323         redisLog(REDIS_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
 324         replicationAbortSyncTransfer();
 325         return;
 326     }
 327     server.repl_transfer_left -= nread;
 328     /* Check if the transfer is now complete */
 329     if (server.repl_transfer_left == 0) {
 330         if (rename(server.repl_transfer_tmpfile,server.dbfilename) == -1) {
 331             redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
 332             replicationAbortSyncTransfer();
 333             return;
 334         }
 335         emptyDb();
 336         if (rdbLoad(server.dbfilename) != REDIS_OK) {
 337             redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
 338             replicationAbortSyncTransfer();
 339             return;
 340         }
 341         /* Final setup of the connected slave <- master link */
 342         aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
 343         zfree(server.repl_transfer_tmpfile);
 344         close(server.repl_transfer_fd);
 345         server.master = createClient(server.repl_transfer_s);
 346         server.master->flags |= REDIS_MASTER;
 347         server.master->authenticated = 1;
 348         server.replstate = REDIS_REPL_CONNECTED;
 349     }
 350 }
 351
 352 int syncWithMaster(void) {
 353     char buf[1024], tmpfile[256], authcmd[1024];
 354     off_t dumpsize;
 355     int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
 356     int dfd, maxtries = 5;
 357
 358     if (fd == -1) {
 359         redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
 360             strerror(errno));
 361         return REDIS_ERR;
 362     }
 363
 364     /* AUTH with the master if required. */
 365     if(server.masterauth) {
 366         snprintf(authcmd, 1024, "AUTH %s\r\n", server.masterauth);
 367         if (syncWrite(fd, authcmd, strlen(server.masterauth)+7, 5) == -1) {
 368             close(fd);
 369             redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s",
 370                 strerror(errno));
 371             return REDIS_ERR;
 372         }
 373         /* Read the AUTH result.  */
 374         if (syncReadLine(fd,buf,1024,3600) == -1) {
 375             close(fd);
 376             redisLog(REDIS_WARNING,"I/O error reading auth result from MASTER: %s",
 377                 strerror(errno));
 378             return REDIS_ERR;
 379         }
 380         if (buf[0] != '+') {
 381             close(fd);
 382             redisLog(REDIS_WARNING,"Cannot AUTH to MASTER, is the masterauth password correct?");
 383             return REDIS_ERR;
 384         }
 385     }
 386
 387     /* Issue the SYNC command */
 388     if (syncWrite(fd,"SYNC \r\n",7,5) == -1) {
 389         close(fd);
 390         redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
 391             strerror(errno));
 392         return REDIS_ERR;
 393     }
 394     /* Read the bulk write count */
 395     if (syncReadLine(fd,buf,1024,3600) == -1) {
 396         close(fd);
 397         redisLog(REDIS_WARNING,"I/O error reading bulk count from MASTER: %s",
 398             strerror(errno));
 399         return REDIS_ERR;
 400     }
 401     if (buf[0] == '-') {
 402         close(fd);
 403         redisLog(REDIS_WARNING,"MASTER aborted replication with an error: %s",
 404             buf+1);
 405         return REDIS_ERR;
 406     } else if (buf[0] != '$') {
 407         close(fd);
 408         redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
 409         return REDIS_ERR;
 410     }
 411     dumpsize = strtol(buf+1,NULL,10);
 412     redisLog(REDIS_NOTICE,"Receiving %ld bytes data dump from MASTER",dumpsize);
 413     /* Read the bulk write data on a temp file */
 414     while(maxtries--) {
 415         snprintf(tmpfile,256,
 416             "temp-%d.%ld.rdb",(int)time(NULL),(long int)getpid());
 417         dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
 418         if (dfd != -1) break;
 419         sleep(1);
 420     }
 421     if (dfd == -1) {
 422         close(fd);
 423         redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
 424         return REDIS_ERR;
 425     }
 426
 427     /* Setup the non blocking download of the bulk file. */
 428     if (aeCreateFileEvent(server.el, fd, AE_READABLE,readSyncBulkPayload) ==
 429         AE_ERR)
 430     {
 431         close(fd);
 432         redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
 433         return REDIS_ERR;
 434     }
 435     server.replstate = REDIS_REPL_TRANSFER;
 436     server.repl_transfer_left = dumpsize;
 437     server.repl_transfer_s = fd;
 438     server.repl_transfer_fd = dfd;
 439     server.repl_transfer_lastio = time(NULL);
 440     server.repl_transfer_tmpfile = zstrdup(tmpfile);
 441     return REDIS_OK;
 442 }
 443
 444 void slaveofCommand(redisClient *c) {
 445     if (!strcasecmp(c->argv[1]->ptr,"no") &&
 446         !strcasecmp(c->argv[2]->ptr,"one")) {
 447         if (server.masterhost) {
 448             sdsfree(server.masterhost);
 449             server.masterhost = NULL;
 450             if (server.master) freeClient(server.master);
 451             if (server.replstate == REDIS_REPL_TRANSFER)
 452                 replicationAbortSyncTransfer();
 453             server.replstate = REDIS_REPL_NONE;
 454             redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
 455         }
 456     } else {
 457         sdsfree(server.masterhost);
 458         server.masterhost = sdsdup(c->argv[1]->ptr);
 459         server.masterport = atoi(c->argv[2]->ptr);
 460         if (server.master) freeClient(server.master);
 461         if (server.replstate == REDIS_REPL_TRANSFER)
 462             replicationAbortSyncTransfer();
 463         server.replstate = REDIS_REPL_CONNECT;
 464         redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
 465             server.masterhost, server.masterport);
 466     }
 467     addReply(c,shared.ok);
 468 }
 469
 470 /* --------------------------- REPLICATION CRON  ---------------------------- */
 471
 472 #define REDIS_REPL_TRANSFER_TIMEOUT 60
 473
 474 void replicationCron(void) {
 475     /* Bulk transfer I/O timeout? */
 476     if (server.masterhost && server.replstate == REDIS_REPL_TRANSFER &&
 477         (time(NULL)-server.repl_transfer_lastio) > REDIS_REPL_TRANSFER_TIMEOUT)
 478     {
 479         redisLog(REDIS_WARNING,"Timeout receiving bulk data from MASTER...");
 480         replicationAbortSyncTransfer();
 481     }
 482
 483     /* Check if we should connect to a MASTER */
 484     if (server.replstate == REDIS_REPL_CONNECT && !(loops % 10)) {
 485         redisLog(REDIS_NOTICE,"Connecting to MASTER...");
 486         if (syncWithMaster() == REDIS_OK) {
 487             redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync succeeded");
 488             if (server.appendonly) rewriteAppendOnlyFileBackground();
 489         }
 490     }
 491 }