[redis.git] / src / replication.c

#include "redis.h"

#include <sys/time.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/stat.h>

/* ---------------------------------- MASTER -------------------------------- */

void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
    listNode *ln;
    listIter li;
    int j;

    listRewind(slaves,&li);
    while((ln = listNext(&li))) {
        redisClient *slave = ln->value;

        /* Don't feed slaves that are still waiting for BGSAVE to start */
        if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) continue;

        /* Feed slaves that are waiting for the initial SYNC (so these commands
         * are queued in the output buffer until the intial SYNC completes),
         * or are already in sync with the master. */
        if (slave->slaveseldb != dictid) {
            robj *selectcmd;

            switch(dictid) {
            case 0: selectcmd = shared.select0; break;
            case 1: selectcmd = shared.select1; break;
            case 2: selectcmd = shared.select2; break;
            case 3: selectcmd = shared.select3; break;
            case 4: selectcmd = shared.select4; break;
            case 5: selectcmd = shared.select5; break;
            case 6: selectcmd = shared.select6; break;
            case 7: selectcmd = shared.select7; break;
            case 8: selectcmd = shared.select8; break;
            case 9: selectcmd = shared.select9; break;
            default:
                selectcmd = createObject(REDIS_STRING,
                    sdscatprintf(sdsempty(),"select %d\r\n",dictid));
                selectcmd->refcount = 0;
                break;
            }
            addReply(slave,selectcmd);
            slave->slaveseldb = dictid;
        }
        addReplyMultiBulkLen(slave,argc);
        for (j = 0; j < argc; j++) addReplyBulk(slave,argv[j]);
    }
}

void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc) {
    listNode *ln;
    listIter li;
    int j;
    sds cmdrepr = sdsnew("+");
    robj *cmdobj;
    struct timeval tv;

    gettimeofday(&tv,NULL);
    cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec);
    if (dictid != 0) cmdrepr = sdscatprintf(cmdrepr,"(db %d) ", dictid);

    for (j = 0; j < argc; j++) {
        if (argv[j]->encoding == REDIS_ENCODING_INT) {
            cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)argv[j]->ptr);
        } else {
            cmdrepr = sdscatrepr(cmdrepr,(char*)argv[j]->ptr,
                        sdslen(argv[j]->ptr));
        }
        if (j != argc-1)
            cmdrepr = sdscatlen(cmdrepr," ",1);
    }
    cmdrepr = sdscatlen(cmdrepr,"\r\n",2);
    cmdobj = createObject(REDIS_STRING,cmdrepr);

    listRewind(monitors,&li);
    while((ln = listNext(&li))) {
        redisClient *monitor = ln->value;
        addReply(monitor,cmdobj);
    }
    decrRefCount(cmdobj);
}

void syncCommand(redisClient *c) {
    /* ignore SYNC if aleady slave or in monitor mode */
    if (c->flags & REDIS_SLAVE) return;

    /* Refuse SYNC requests if we are a slave but the link with our master
     * is not ok... */
    if (server.masterhost && server.replstate != REDIS_REPL_CONNECTED) {
        addReplyError(c,"Can't SYNC while not connected with my master");
        return;
    }

    /* SYNC can't be issued when the server has pending data to send to
     * the client about already issued commands. We need a fresh reply
     * buffer registering the differences between the BGSAVE and the current
     * dataset, so that we can copy to other slaves if needed. */
    if (listLength(c->reply) != 0) {
        addReplyError(c,"SYNC is invalid with pending input");
        return;
    }

    redisLog(REDIS_NOTICE,"Slave ask for synchronization");
    /* Here we need to check if there is a background saving operation
     * in progress, or if it is required to start one */
    if (server.bgsavechildpid != -1) {
        /* Ok a background save is in progress. Let's check if it is a good
         * one for replication, i.e. if there is another slave that is
         * registering differences since the server forked to save */
        redisClient *slave;
        listNode *ln;
        listIter li;

        listRewind(server.slaves,&li);
        while((ln = listNext(&li))) {
            slave = ln->value;
            if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) break;
        }
        if (ln) {
            /* Perfect, the server is already registering differences for
             * another slave. Set the right state, and copy the buffer. */
            listRelease(c->reply);
            c->reply = listDup(slave->reply);
            c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
            redisLog(REDIS_NOTICE,"Waiting for end of BGSAVE for SYNC");
        } else {
            /* No way, we need to wait for the next BGSAVE in order to
             * register differences */
            c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
            redisLog(REDIS_NOTICE,"Waiting for next BGSAVE for SYNC");
        }
    } else {
        /* Ok we don't have a BGSAVE in progress, let's start one */
        redisLog(REDIS_NOTICE,"Starting BGSAVE for SYNC");
        if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
            redisLog(REDIS_NOTICE,"Replication failed, can't BGSAVE");
            addReplyError(c,"Unable to perform background save");
            return;
        }
        c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
    }
    c->repldbfd = -1;
    c->flags |= REDIS_SLAVE;
    c->slaveseldb = 0;
    listAddNodeTail(server.slaves,c);
    return;
}

void sendBulkToSlave(aeEventLoop *el, int fd, void *privdata, int mask) {
    redisClient *slave = privdata;
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(mask);
    char buf[REDIS_IOBUF_LEN];
    ssize_t nwritten, buflen;

    if (slave->repldboff == 0) {
        /* Write the bulk write count before to transfer the DB. In theory here
         * we don't know how much room there is in the output buffer of the
         * socket, but in pratice SO_SNDLOWAT (the minimum count for output
         * operations) will never be smaller than the few bytes we need. */
        sds bulkcount;

        bulkcount = sdscatprintf(sdsempty(),"$%lld\r\n",(unsigned long long)
            slave->repldbsize);
        if (write(fd,bulkcount,sdslen(bulkcount)) != (signed)sdslen(bulkcount))
        {
            sdsfree(bulkcount);
            freeClient(slave);
            return;
        }
        sdsfree(bulkcount);
    }
    lseek(slave->repldbfd,slave->repldboff,SEEK_SET);
    buflen = read(slave->repldbfd,buf,REDIS_IOBUF_LEN);
    if (buflen <= 0) {
        redisLog(REDIS_WARNING,"Read error sending DB to slave: %s",
            (buflen == 0) ? "premature EOF" : strerror(errno));
        freeClient(slave);
        return;
    }
    if ((nwritten = write(fd,buf,buflen)) == -1) {
        redisLog(REDIS_VERBOSE,"Write error sending DB to slave: %s",
            strerror(errno));
        freeClient(slave);
        return;
    }
    slave->repldboff += nwritten;
    if (slave->repldboff == slave->repldbsize) {
        close(slave->repldbfd);
        slave->repldbfd = -1;
        aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
        slave->replstate = REDIS_REPL_ONLINE;
        if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE,
            sendReplyToClient, slave) == AE_ERR) {
            freeClient(slave);
            return;
        }
        addReplySds(slave,sdsempty());
        redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
    }
}

/* This function is called at the end of every backgrond saving.
 * The argument bgsaveerr is REDIS_OK if the background saving succeeded
 * otherwise REDIS_ERR is passed to the function.
 *
 * The goal of this function is to handle slaves waiting for a successful
 * background saving in order to perform non-blocking synchronization. */
void updateSlavesWaitingBgsave(int bgsaveerr) {
    listNode *ln;
    int startbgsave = 0;
    listIter li;

    listRewind(server.slaves,&li);
    while((ln = listNext(&li))) {
        redisClient *slave = ln->value;

        if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START) {
            startbgsave = 1;
            slave->replstate = REDIS_REPL_WAIT_BGSAVE_END;
        } else if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_END) {
            struct redis_stat buf;

            if (bgsaveerr != REDIS_OK) {
                freeClient(slave);
                redisLog(REDIS_WARNING,"SYNC failed. BGSAVE child returned an error");
                continue;
            }
            if ((slave->repldbfd = open(server.dbfilename,O_RDONLY)) == -1 ||
                redis_fstat(slave->repldbfd,&buf) == -1) {
                freeClient(slave);
                redisLog(REDIS_WARNING,"SYNC failed. Can't open/stat DB after BGSAVE: %s", strerror(errno));
                continue;
            }
            slave->repldboff = 0;
            slave->repldbsize = buf.st_size;
            slave->replstate = REDIS_REPL_SEND_BULK;
            aeDeleteFileEvent(server.el,slave->fd,AE_WRITABLE);
            if (aeCreateFileEvent(server.el, slave->fd, AE_WRITABLE, sendBulkToSlave, slave) == AE_ERR) {
                freeClient(slave);
                continue;
            }
        }
    }
    if (startbgsave) {
        if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
            listIter li;

            listRewind(server.slaves,&li);
            redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
            while((ln = listNext(&li))) {
                redisClient *slave = ln->value;

                if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
                    freeClient(slave);
            }
        }
    }
}

/* ----------------------------------- SLAVE -------------------------------- */

/* Abort the async download of the bulk dataset while SYNC-ing with master */
void replicationAbortSyncTransfer(void) {
    redisAssert(server.replstate == REDIS_REPL_TRANSFER);

    aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
    close(server.repl_transfer_s);
    close(server.repl_transfer_fd);
    unlink(server.repl_transfer_tmpfile);
    zfree(server.repl_transfer_tmpfile);
    server.replstate = REDIS_REPL_CONNECT;
}

/* Asynchronously read the SYNC payload we receive from a master */
void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
    char buf[4096];
    ssize_t nread, readlen;
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(privdata);
    REDIS_NOTUSED(mask);

    /* If repl_transfer_left == -1 we still have to read the bulk length
     * from the master reply. */
    if (server.repl_transfer_left == -1) {
        if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout) == -1) {
            redisLog(REDIS_WARNING,
                "I/O error reading bulk count from MASTER: %s",
                strerror(errno));
            goto error;
        }

        if (buf[0] == '-') {
            redisLog(REDIS_WARNING,
                "MASTER aborted replication with an error: %s",
                buf+1);
            goto error;
        } else if (buf[0] == '\0') {
            /* At this stage just a newline works as a PING in order to take
             * the connection live. So we refresh our last interaction
             * timestamp. */
            server.repl_transfer_lastio = time(NULL);
            return;
        } else if (buf[0] != '$') {
            redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
            goto error;
        }
        server.repl_transfer_left = strtol(buf+1,NULL,10);
        redisLog(REDIS_NOTICE,
            "MASTER <-> SLAVE sync: receiving %ld bytes from master",
            server.repl_transfer_left);
        return;
    }

    /* Read bulk data */
    readlen = (server.repl_transfer_left < (signed)sizeof(buf)) ?
        server.repl_transfer_left : (signed)sizeof(buf);
    nread = read(fd,buf,readlen);
    if (nread <= 0) {
        redisLog(REDIS_WARNING,"I/O error trying to sync with MASTER: %s",
            (nread == -1) ? strerror(errno) : "connection lost");
        replicationAbortSyncTransfer();
        return;
    }
    server.repl_transfer_lastio = time(NULL);
    if (write(server.repl_transfer_fd,buf,nread) != nread) {
        redisLog(REDIS_WARNING,"Write error or short write writing to the DB dump file needed for MASTER <-> SLAVE synchrnonization: %s", strerror(errno));
        goto error;
    }
    server.repl_transfer_left -= nread;
    /* Check if the transfer is now complete */
    if (server.repl_transfer_left == 0) {
        if (rename(server.repl_transfer_tmpfile,server.dbfilename) == -1) {
            redisLog(REDIS_WARNING,"Failed trying to rename the temp DB into dump.rdb in MASTER <-> SLAVE synchronization: %s", strerror(errno));
            replicationAbortSyncTransfer();
            return;
        }
        redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory");
        emptyDb();
        /* Before loading the DB into memory we need to delete the readable
         * handler, otherwise it will get called recursively since
         * rdbLoad() will call the event loop to process events from time to
         * time for non blocking loading. */
        aeDeleteFileEvent(server.el,server.repl_transfer_s,AE_READABLE);
        if (rdbLoad(server.dbfilename) != REDIS_OK) {
            redisLog(REDIS_WARNING,"Failed trying to load the MASTER synchronization DB from disk");
            replicationAbortSyncTransfer();
            return;
        }
        /* Final setup of the connected slave <- master link */
        zfree(server.repl_transfer_tmpfile);
        close(server.repl_transfer_fd);
        server.master = createClient(server.repl_transfer_s);
        server.master->flags |= REDIS_MASTER;
        server.master->authenticated = 1;
        server.replstate = REDIS_REPL_CONNECTED;
        redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success");
        /* Rewrite the AOF file now that the dataset changed. */
        if (server.appendonly) rewriteAppendOnlyFileBackground();
    }

    return;

error:
    replicationAbortSyncTransfer();
    return;
}

void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
    char buf[1024], tmpfile[256];
    int dfd, maxtries = 5;
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(privdata);
    REDIS_NOTUSED(mask);

    redisLog(REDIS_NOTICE,"Non blocking connect for SYNC fired the event.");
    /* This event should only be triggered once since it is used to have a
     * non-blocking connect(2) to the master. It has been triggered when this
     * function is called, so we can delete it. */
    aeDeleteFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE);

    /* AUTH with the master if required. */
    if(server.masterauth) {
        char authcmd[1024];
        size_t authlen;

        authlen = snprintf(authcmd,sizeof(authcmd),"AUTH %s\r\n",server.masterauth);
        if (syncWrite(fd,authcmd,authlen,server.repl_syncio_timeout) == -1) {
            redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s",
                strerror(errno));
            goto error;
        }
        /* Read the AUTH result.  */
        if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout) == -1) {
            redisLog(REDIS_WARNING,"I/O error reading auth result from MASTER: %s",
                strerror(errno));
            goto error;
        }
        if (buf[0] != '+') {
            redisLog(REDIS_WARNING,"Cannot AUTH to MASTER, is the masterauth password correct?");
            goto error;
        }
    }

    /* Issue the SYNC command */
    if (syncWrite(fd,"SYNC \r\n",7,server.repl_syncio_timeout) == -1) {
        redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
            strerror(errno));
        goto error;
    }

    /* Prepare a suitable temp file for bulk transfer */
    while(maxtries--) {
        snprintf(tmpfile,256,
            "temp-%d.%ld.rdb",(int)time(NULL),(long int)getpid());
        dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
        if (dfd != -1) break;
        sleep(1);
    }
    if (dfd == -1) {
        redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
        goto error;
    }

    /* Setup the non blocking download of the bulk file. */
    if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL)
            == AE_ERR)
    {
        redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
        goto error;
    }

    server.replstate = REDIS_REPL_TRANSFER;
    server.repl_transfer_left = -1;
    server.repl_transfer_fd = dfd;
    server.repl_transfer_lastio = time(NULL);
    server.repl_transfer_tmpfile = zstrdup(tmpfile);
    return;

error:
    server.replstate = REDIS_REPL_CONNECT;
    close(fd);
    return;
}

int connectWithMaster(void) {
    int fd;

    fd = anetTcpNonBlockConnect(NULL,server.masterhost,server.masterport);
    if (fd == -1) {
        redisLog(REDIS_WARNING,"Unable to connect to MASTER: %s",
            strerror(errno));
        return REDIS_ERR;
    }

    if (aeCreateFileEvent(server.el,fd,AE_READABLE|AE_WRITABLE,syncWithMaster,NULL) ==
            AE_ERR)
    {
        close(fd);
        redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
        return REDIS_ERR;
    }

    server.repl_transfer_s = fd;
    server.replstate = REDIS_REPL_CONNECTING;
    return REDIS_OK;
}

void slaveofCommand(redisClient *c) {
    if (!strcasecmp(c->argv[1]->ptr,"no") &&
        !strcasecmp(c->argv[2]->ptr,"one")) {
        if (server.masterhost) {
            sdsfree(server.masterhost);
            server.masterhost = NULL;
            if (server.master) freeClient(server.master);
            if (server.replstate == REDIS_REPL_TRANSFER)
                replicationAbortSyncTransfer();
            server.replstate = REDIS_REPL_NONE;
            redisLog(REDIS_NOTICE,"MASTER MODE enabled (user request)");
        }
    } else {
        sdsfree(server.masterhost);
        server.masterhost = sdsdup(c->argv[1]->ptr);
        server.masterport = atoi(c->argv[2]->ptr);
        if (server.master) freeClient(server.master);
        if (server.replstate == REDIS_REPL_TRANSFER)
            replicationAbortSyncTransfer();
        server.replstate = REDIS_REPL_CONNECT;
        redisLog(REDIS_NOTICE,"SLAVE OF %s:%d enabled (user request)",
            server.masterhost, server.masterport);
    }
    addReply(c,shared.ok);
}

/* --------------------------- REPLICATION CRON  ---------------------------- */

#define REDIS_REPL_TIMEOUT 60
#define REDIS_REPL_PING_SLAVE_PERIOD 10

void replicationCron(void) {
    /* Bulk transfer I/O timeout? */
    if (server.masterhost && server.replstate == REDIS_REPL_TRANSFER &&
        (time(NULL)-server.repl_transfer_lastio) > REDIS_REPL_TIMEOUT)
    {
        redisLog(REDIS_WARNING,"Timeout receiving bulk data from MASTER...");
        replicationAbortSyncTransfer();
    }

    /* Timed out master when we are an already connected slave? */
    if (server.masterhost && server.replstate == REDIS_REPL_CONNECTED &&
        (time(NULL)-server.master->lastinteraction) > REDIS_REPL_TIMEOUT)
    {
        redisLog(REDIS_WARNING,"MASTER time out: no data nor PING received...");
        freeClient(server.master);
    }

    /* Check if we should connect to a MASTER */
    if (server.replstate == REDIS_REPL_CONNECT) {
        redisLog(REDIS_NOTICE,"Connecting to MASTER...");
        if (connectWithMaster() == REDIS_OK) {
            redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync started");
        }
    }
    
    /* If we have attached slaves, PING them from time to time.
     * So slaves can implement an explicit timeout to masters, and will
     * be able to detect a link disconnection even if the TCP connection
     * will not actually go down. */
    if (!(server.cronloops % (REDIS_REPL_PING_SLAVE_PERIOD*10))) {
        listIter li;
        listNode *ln;

        listRewind(server.slaves,&li);
        while((ln = listNext(&li))) {
            redisClient *slave = ln->value;

            /* Don't ping slaves that are in the middle of a bulk transfer
             * with the master for first synchronization. */
            if (slave->replstate == REDIS_REPL_SEND_BULK) continue;
            if (slave->replstate == REDIS_REPL_ONLINE) {
                /* If the slave is online send a normal ping */
                addReplySds(slave,sdsnew("PING\r\n"));
            } else {
                /* Otherwise we are in the pre-synchronization stage.
                 * Just a newline will do the work of refreshing the
                 * connection last interaction time, and at the same time
                 * we'll be sure that being a single char there are no
                 * short-write problems. */
                if (write(slave->fd, "\n", 1) == -1) {
                    /* Don't worry, it's just a ping. */
                }
            }
        }
    }
}
Commit	Line	Data
e2641e09	1	#include "redis.h"
	2
	3	#include <sys/time.h>
	4	#include <unistd.h>
	5	#include <fcntl.h>
	6	#include <sys/stat.h>
	7
f4aa600b	8	/* ---------------------------------- MASTER -------------------------------- */
f4aa600b	9
e2641e09	10	void replicationFeedSlaves(list slaves, int dictid, robj *argv, int argc) {
	11	listNode *ln;
	12	listIter li;
632e4c09	13	int j;
e2641e09	14
e2641e09	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
632e4c09 PN	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. */
e2641e09	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	}
632e4c09 PN	48	addReplyMultiBulkLen(slave,argc);
632e4c09 PN	49	for (j = 0; j < argc; j++) addReplyBulk(slave,argv[j]);
e2641e09	50	}
e2641e09	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);
2b2eca1f	62	cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec);
e2641e09	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) {
d3b958c3	67	cmdrepr = sdscatprintf(cmdrepr, "\"%ld\"", (long)argv[j]->ptr);
e2641e09	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
e2641e09	86	void syncCommand(redisClient *c) {
	87	/* ignore SYNC if aleady slave or in monitor mode */
	88	if (c->flags & REDIS_SLAVE) return;
	89
778b2210	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) {
3ab20376	93	addReplyError(c,"Can't SYNC while not connected with my master");
778b2210	94	return;
	95	}
	96
e2641e09	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) {
3ab20376	102	addReplyError(c,"SYNC is invalid with pending input");
e2641e09	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");
3ab20376	140	addReplyError(c,"Unable to perform background save");
e2641e09	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
f4aa600b	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) {
26b33669	280	char buf[4096];
62ec599c	281	ssize_t nread, readlen;
	282	REDIS_NOTUSED(el);
	283	REDIS_NOTUSED(privdata);
	284	REDIS_NOTUSED(mask);
f4aa600b	285
26b33669	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) {
890a2ed9	289	if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout) == -1) {
26b33669	290	redisLog(REDIS_WARNING,
	291	"I/O error reading bulk count from MASTER: %s",
	292	strerror(errno));
b075621f	293	goto error;
26b33669	294	}
b075621f	295
26b33669	296	if (buf[0] == '-') {
	297	redisLog(REDIS_WARNING,
	298	"MASTER aborted replication with an error: %s",
	299	buf+1);
b075621f	300	goto error;
89a1433e	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;
26b33669	307	} else if (buf[0] != '$') {
26b33669	308	redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
b075621f	309	goto error;
26b33669	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 */
62ec599c	319	readlen = (server.repl_transfer_left < (signed)sizeof(buf)) ?
62ec599c	320	server.repl_transfer_left : (signed)sizeof(buf);
f4aa600b	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));
b075621f	331	goto error;
f4aa600b	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	}
f6433915	341	redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Loading DB in memory");
f4aa600b	342	emptyDb();
9fd01051	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);
f4aa600b	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 */
f4aa600b	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;
26b33669	360	redisLog(REDIS_NOTICE, "MASTER <-> SLAVE sync: Finished with success");
45029d37	361	/* Rewrite the AOF file now that the dataset changed. */
45029d37	362	if (server.appendonly) rewriteAppendOnlyFileBackground();
f4aa600b	363	}
b075621f PN	364
	365	return;
	366
	367	error:
	368	replicationAbortSyncTransfer();
	369	return;
f4aa600b	370	}
f4aa600b	371
a3309139 PN	372	void syncWithMaster(aeEventLoop el, int fd, void privdata, int mask) {
a3309139 PN	373	char buf[1024], tmpfile[256];
e2641e09	374	int dfd, maxtries = 5;
a3309139 PN	375	REDIS_NOTUSED(el);
	376	REDIS_NOTUSED(privdata);
	377	REDIS_NOTUSED(mask);
e2641e09	378
45029d37	379	redisLog(REDIS_NOTICE,"Non blocking connect for SYNC fired the event.");
b075621f PN	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. */
45029d37	383	aeDeleteFileEvent(server.el,fd,AE_READABLE\|AE_WRITABLE);
e2641e09	384
	385	/* AUTH with the master if required. */
	386	if(server.masterauth) {
a3309139 PN	387	char authcmd[1024];
	388	size_t authlen;
	389
	390	authlen = snprintf(authcmd,sizeof(authcmd),"AUTH %s\r\n",server.masterauth);
890a2ed9	391	if (syncWrite(fd,authcmd,authlen,server.repl_syncio_timeout) == -1) {
e2641e09	392	redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s",
e2641e09	393	strerror(errno));
a3309139 PN	394	goto error;
a3309139 PN	395	}
e2641e09	396	/* Read the AUTH result. */
890a2ed9	397	if (syncReadLine(fd,buf,1024,server.repl_syncio_timeout) == -1) {
e2641e09	398	redisLog(REDIS_WARNING,"I/O error reading auth result from MASTER: %s",
e2641e09	399	strerror(errno));
a3309139	400	goto error;
e2641e09	401	}
e2641e09	402	if (buf[0] != '+') {
e2641e09	403	redisLog(REDIS_WARNING,"Cannot AUTH to MASTER, is the masterauth password correct?");
a3309139	404	goto error;
e2641e09	405	}
	406	}
	407
	408	/* Issue the SYNC command */
890a2ed9	409	if (syncWrite(fd,"SYNC \r\n",7,server.repl_syncio_timeout) == -1) {
e2641e09	410	redisLog(REDIS_WARNING,"I/O error writing to MASTER: %s",
e2641e09	411	strerror(errno));
a3309139	412	goto error;
e2641e09	413	}
26b33669	414
26b33669	415	/* Prepare a suitable temp file for bulk transfer */
e2641e09	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) {
e2641e09	424	redisLog(REDIS_WARNING,"Opening the temp file needed for MASTER <-> SLAVE synchronization: %s",strerror(errno));
a3309139	425	goto error;
e2641e09	426	}
e2641e09	427
f4aa600b	428	/* Setup the non blocking download of the bulk file. */
a3309139	429	if (aeCreateFileEvent(server.el,fd, AE_READABLE,readSyncBulkPayload,NULL)
62ec599c	430	== AE_ERR)
f4aa600b	431	{
f4aa600b	432	redisLog(REDIS_WARNING,"Can't create readable event for SYNC");
a3309139	433	goto error;
e2641e09	434	}
a3309139	435
f4aa600b	436	server.replstate = REDIS_REPL_TRANSFER;
26b33669	437	server.repl_transfer_left = -1;
f4aa600b	438	server.repl_transfer_fd = dfd;
	439	server.repl_transfer_lastio = time(NULL);
	440	server.repl_transfer_tmpfile = zstrdup(tmpfile);
a3309139 PN	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
45029d37	459	if (aeCreateFileEvent(server.el,fd,AE_READABLE\|AE_WRITABLE,syncWithMaster,NULL) ==
b075621f	460	AE_ERR)
a3309139 PN	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;
e2641e09	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);
f4aa600b	479	if (server.replstate == REDIS_REPL_TRANSFER)
f4aa600b	480	replicationAbortSyncTransfer();
e2641e09	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);
f4aa600b	489	if (server.replstate == REDIS_REPL_TRANSFER)
f4aa600b	490	replicationAbortSyncTransfer();
e2641e09	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	}
f4aa600b	497
	498	/* --------------------------- REPLICATION CRON ---------------------------- */
	499
89a1433e	500	#define REDIS_REPL_TIMEOUT 60
89a1433e	501	#define REDIS_REPL_PING_SLAVE_PERIOD 10
f4aa600b	502
	503	void replicationCron(void) {
	504	/* Bulk transfer I/O timeout? */
	505	if (server.masterhost && server.replstate == REDIS_REPL_TRANSFER &&
89a1433e	506	(time(NULL)-server.repl_transfer_lastio) > REDIS_REPL_TIMEOUT)
f4aa600b	507	{
	508	redisLog(REDIS_WARNING,"Timeout receiving bulk data from MASTER...");
	509	replicationAbortSyncTransfer();
	510	}
	511
89a1433e	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
f4aa600b	520	/* Check if we should connect to a MASTER */
62ec599c	521	if (server.replstate == REDIS_REPL_CONNECT) {
f4aa600b	522	redisLog(REDIS_NOTICE,"Connecting to MASTER...");
a3309139 PN	523	if (connectWithMaster() == REDIS_OK) {
a3309139 PN	524	redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync started");
f4aa600b	525	}
f4aa600b	526	}
89a1433e	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. */
f96a9f82	552	if (write(slave->fd, "\n", 1) == -1) {
	553	/* Don't worry, it's just a ping. */
	554	}
89a1433e	555	}
	556	}
	557	}
f4aa600b	558	}