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 void syncCommand(redisClient *c) {
 114     /* ignore SYNC if aleady slave or in monitor mode */
 115     if (c->flags & REDIS_SLAVE) return;
 116
 117     /* Refuse SYNC requests if we are a slave but the link with our master
 118      * is not ok... */
 119     if (server.masterhost && server.replstate != REDIS_REPL_CONNECTED) {
 120         addReplyError(c,"Can't SYNC while not connected with my master");
 121         return;
 122     }
 123
 124     /* SYNC can't be issued when the server has pending data to send to
 125      * the client about already issued commands. We need a fresh reply
 126      * buffer registering the differences between the BGSAVE and the current
 127      * dataset, so that we can copy to other slaves if needed. */
 128     if (listLength(c->reply) != 0) {
 129         addReplyError(c,"SYNC is invalid with pending input");
 130         return;
 131     }
 132
 133     redisLog(REDIS_NOTICE,"Slave ask for synchronization");
 134     /* Here we need to check if there is a background saving operation
 135      * in progress, or if it is required to start one */
 136     if (server.bgsavechildpid != -1) {
 137         /* Ok a background save is in progress. Let's check if it is a good
 138          * one for replication, i.e. if there is another slave that is
 139          * registering differences since the server forked to save */
 140         redisClient *slave;
 141         listNode *ln;
 142         listIter li;
 143
 144         listRewind(server.slaves,&li);
 145         while((ln = listNext(&li))) {
 146             slave = ln->value;
 147             if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
 148         }
 149         if (ln) {
 150             /* Perfect, the server is already registering differences for
 151              * another slave. Set the right state, and copy the buffer. */
 152             listRelease(c->reply);
 153             c->reply = listDup(slave->reply);
 154             c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
 155             redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
 156         } else {
 157             /* No way, we need to wait for the next BGSAVE in order to
 158              * register differences */
 159             c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
 160             redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
 161         }
 162     } else {
 163         /* Ok we don't have a BGSAVE in progress, let's start one */
 164         redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC");
 165         if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
 166             redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
 167             addReplyError(c,"Unable to perform background save");
 168             return;
 169         }
 170         c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
 171     }
 172     c->repldbfd = -1;
 173     c->flags |= REDIS_SLAVE;
 174     c->slaveseldb = 0;
 175     listAddNodeTail(server.slaves,c);
 176     return;
 177 }
 178
 179 void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
 180     redisClient *slave = privdata;
 181     REDIS_NOTUSED(el);
 182     REDIS_NOTUSED(mask);
 183     char buf[REDIS_IOBUF_LEN];
 184     ssize_t nwritten, buflen;
 185
 186     if (slave->repldboff == 0) {
 187         /* Write the bulk write count before to transfer the DB. In theory here
 188          * we don't know how much room there is in the output buffer of the
 189          * socket, but in pratice SO_SNDLOWAT (the minimum count for output
 190          * operations) will never be smaller than the few bytes we need. */
 191         sds bulkcount;
 192
 193         bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long)
 194             slave->repldbsize);
 195         if (write(fd,bulkcount,sdslen(bulkcount)) != (signed)sdslen(bulkcount))
 196         {
 197             sdsfree(bulkcount);
 198             freeClient(slave);
 199             return;
 200         }
 201         sdsfree(bulkcount);
 202     }
 203     lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
 204     buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN);
 205     if (buflen <= 0) {
 206         redisLog(REDIS_WARNING,"Read error sending DB to slave: %s",
 207             (buflen == 0) ? "premature EOF" : strerror(errno));
 208         freeClient(slave);
 209         return;
 210     }
 211     if ((nwritten = write(fd,buf,buflen)) == -1) {
 212         redisLog(REDIS_VERBOSE,"Write error sending DB to slave: %s",
 213             strerror(errno));
 214         freeClient(slave);
 215         return;
 216     }
 217     slave->repldboff += nwritten;
 218     if (slave->repldboff == slave->repldbsize) {
 219         close(slave->repldbfd);
 220         slave->repldbfd = -1;
 221         aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
 222         slave->replstate = REDIS_REPL_ONLINE;
 223         if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
 224             sendReplyToClient, slave) == AE_ERR) {
 225             freeClient(slave);
 226             return;
 227         }
 228         addReplySds(slave,sdsempty());
 229         redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
 230     }
 231 }
 232
 233 /* This function is called at the end of every backgrond saving.
 234  * The argument bgsaveerr is REDIS_OK if the background saving succeeded
 235  * otherwise REDIS_ERR is passed to the function.
 236  *
 237  * The goal of this function is to handle slaves waiting for a successful
 238  * background saving in order to perform non-blocking synchronization. */
 239 void updateSlavesWaitingBgsave(int bgsaveerr) {
 240     listNode *ln;
 241     int startbgsave = 0;
 242     listIter li;
 243
 244     listRewind(server.slaves,&li);
 245     while((ln = listNext(&li))) {
 246         redisClient *slave = ln->value;
 247
 248         if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
 249             startbgsave = 1;
 250             slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
 251         } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
 252             struct redis_stat buf;
 253
 254             if (bgsaveerr != REDIS_OK) {
 255                 freeClient(slave);
 256                 redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error");
 257                 continue;
 258             }
 259             if ((slave->repldbfd = open(server.dbfilename,O_RDONLY)) == -1 ||
 260                 redis_fstat(slave->repldbfd,&buf) == -1) {
 261                 freeClient(slave);
 262                 redisLog(REDIS_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno));
 263                 continue;
 264             }
 265             slave->repldboff = 0;
 266             slave->repldbsize = buf.st_size;
 267             slave->replstate = REDIS_REPL_SEND_BULK;
 268             aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
 269             if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) {
 270                 freeClient(slave);
 271                 continue;
 272             }
 273         }
 274     }
 275     if (startbgsave) {
 276         if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
 277             listIter li;
 278
 279             listRewind(server.slaves,&li);
 280             redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
 281             while((ln = listNext(&li))) {
 282                 redisClient *slave = ln->value;
 283
 284                 if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
 285                     freeClient(slave);
 286             }
 287         }
 288     }
 289 }
 290
 291 int syncWithMaster(void) {
 292     char buf[1024], tmpfile[256], authcmd[1024];
 293     long dumpsize;
 294     int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
 295     int dfd, maxtries = 5;
 296
 297     if (fd == -1) {
 298         redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
 299             strerror(errno));
 300         return REDIS_ERR;
 301     }
 302
 303     /* AUTH with the master if required. */
 304     if(server.masterauth) {
 305         snprintf(authcmd, 1024, "AUTH %s\r\n", server.masterauth);
 306         if (syncWrite(fd, authcmd, strlen(server.masterauth)+7, 5) == -1) {
 307             close(fd);
 308             redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s",
 309                 strerror(errno));
 310             return REDIS_ERR;
 311         }
 312         /* Read the AUTH result.  */
 313         if (syncReadLine(fd,buf,1024,3600) == -1) {
 314             close(fd);
 315             redisLog(REDIS_WARNING,"I/O error reading auth result from MASTER: %s",
 316                 strerror(errno));
 317             return REDIS_ERR;
 318         }
 319         if (buf[0] != '+') {
 320             close(fd);
 321             redisLog(REDIS_WARNING,"Cannot AUTH to MASTER, is the masterauth password correct?");
 322             return REDIS_ERR;
 323         }
 324     }
 325
 326     /* Issue the SYNC command */
 327     if (syncWrite(fd,"SYNC \r\n",7,5) == -1) {
 328         close(fd);
 329         redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
 330             strerror(errno));
 331         return REDIS_ERR;
 332     }
 333     /* Read the bulk write count */
 334     if (syncReadLine(fd,buf,1024,3600) == -1) {
 335         close(fd);
 336         redisLog(REDIS_WARNING,"I/O error reading bulk count from MASTER: %s",
 337             strerror(errno));
 338         return REDIS_ERR;
 339     }
 340     if (buf[0] == '-') {
 341         close(fd);
 342         redisLog(REDIS_WARNING,"MASTER aborted replication with an error: %s",
 343             buf+1);
 344         return REDIS_ERR;
 345     } else if (buf[0] != '$') {
 346         close(fd);
 347         redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
 348         return REDIS_ERR;
 349     }
 350     dumpsize = strtol(buf+1,NULL,10);
 351     redisLog(REDIS_NOTICE,"Receiving %ld bytes data dump from MASTER",dumpsize);
 352     /* Read the bulk write data on a temp file */
 353     while(maxtries--) {
 354         snprintf(tmpfile,256,
 355             "temp-%d.%ld.rdb",(int)time(NULL),(long int)getpid());
 356         dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
 357         if (dfd != -1) break;
 358         sleep(1);
 359     }
 360     if (dfd == -1) {
 361         close(fd);
 362         redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
 363         return REDIS_ERR;
 364     }
 365     while(dumpsize) {
 366         int nread, nwritten;
 367
 368         nread = read(fd,buf,(dumpsize < 1024)?dumpsize:1024);
 369         if (nread <= 0) {
 370             redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
 371                 (nread == -1) ? strerror(errno) : "connection lost");
 372             close(fd);
 373             close(dfd);
 374             return REDIS_ERR;
 375         }
 376         nwritten = write(dfd,buf,nread);
 377         if (nwritten == -1) {
 378             redisLog(REDIS_WARNING,"Write error writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
 379             close(fd);
 380             close(dfd);
 381             return REDIS_ERR;
 382         }
 383         dumpsize -= nread;
 384     }
 385     close(dfd);
 386     if (rename(tmpfile,server.dbfilename) == -1) {
 387         redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
 388         unlink(tmpfile);
 389         close(fd);
 390         return REDIS_ERR;
 391     }
 392     emptyDb();
 393     if (rdbLoad(server.dbfilename) != REDIS_OK) {
 394         redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
 395         close(fd);
 396         return REDIS_ERR;
 397     }
 398     server.master = createClient(fd);
 399     server.master->flags |= REDIS_MASTER;
 400     server.master->authenticated = 1;
 401     server.replstate = REDIS_REPL_CONNECTED;
 402     return REDIS_OK;
 403 }
 404
 405 void slaveofCommand(redisClient *c) {
 406     if (!strcasecmp(c->argv[1]->ptr,"no") &&
 407         !strcasecmp(c->argv[2]->ptr,"one")) {
 408         if (server.masterhost) {
 409             sdsfree(server.masterhost);
 410             server.masterhost = NULL;
 411             if (server.master) freeClient(server.master);
 412             server.replstate = REDIS_REPL_NONE;
 413             redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
 414         }
 415     } else {
 416         sdsfree(server.masterhost);
 417         server.masterhost = sdsdup(c->argv[1]->ptr);
 418         server.masterport = atoi(c->argv[2]->ptr);
 419         if (server.master) freeClient(server.master);
 420         server.replstate = REDIS_REPL_CONNECT;
 421         redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
 422             server.masterhost, server.masterport);
 423     }
 424     addReply(c,shared.ok);
 425 }