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