]> git.saurik.com Git - redis.git/blobdiff - redis.c
Non blocking replication (finally!). C-side linked lists API improved.
[redis.git] / redis.c
diff --git a/redis.c b/redis.c
index 7308e2f21c91c13b8afe0d89bd1ed2ce96b475c2..7c382af74993d9b33059a6e0f82e5958fdd7a334 100644 (file)
--- a/redis.c
+++ b/redis.c
@@ -65,7 +65,7 @@
 /* Static server configuration */
 #define REDIS_SERVERPORT        6379    /* TCP port */
 #define REDIS_MAXIDLETIME       (60*5)  /* default client timeout */
-#define REDIS_QUERYBUF_LEN      1024
+#define REDIS_IOBUF_LEN         1024
 #define REDIS_LOADBUF_LEN       1024
 #define REDIS_MAX_ARGS          16
 #define REDIS_DEFAULT_DBNUM     16
 #define REDIS_MASTER 4      /* This client is a master server */
 #define REDIS_MONITOR 8      /* This client is a slave monitor, see MONITOR */
 
-/* Server replication state */
+/* Slave replication state - slave side */
 #define REDIS_REPL_NONE 0   /* No active replication */
 #define REDIS_REPL_CONNECT 1    /* Must connect to master */
 #define REDIS_REPL_CONNECTED 2  /* Connected to master */
 
+/* Slave replication state - from the point of view of master
+ * Note that in SEND_BULK and ONLINE state the slave receives new updates
+ * in its output queue. In the WAIT_BGSAVE state instead the server is waiting
+ * to start the next background saving in order to send updates to it. */
+#define REDIS_REPL_WAIT_BGSAVE_START 3 /* master waits bgsave to start feeding it */
+#define REDIS_REPL_WAIT_BGSAVE_END 4 /* master waits bgsave to start bulk DB transmission */
+#define REDIS_REPL_SEND_BULK 5 /* master is sending the bulk DB */
+#define REDIS_REPL_ONLINE 6 /* bulk DB already transmitted, receive updates */
+
 /* List related stuff */
 #define REDIS_HEAD 0
 #define REDIS_TAIL 1
@@ -176,13 +185,17 @@ typedef struct redisClient {
     sds querybuf;
     robj *argv[REDIS_MAX_ARGS];
     int argc;
-    int bulklen;    /* bulk read len. -1 if not in bulk read mode */
+    int bulklen;            /* bulk read len. -1 if not in bulk read mode */
     list *reply;
     int sentlen;
     time_t lastinteraction; /* time of the last interaction, used for timeout */
-    int flags; /* REDIS_CLOSE | REDIS_SLAVE | REDIS_MONITOR */
-    int slaveseldb; /* slave selected db, if this client is a slave */
-    int authenticated;    /* when requirepass is non-NULL */
+    int flags;              /* REDIS_CLOSE | REDIS_SLAVE | REDIS_MONITOR */
+    int slaveseldb;         /* slave selected db, if this client is a slave */
+    int authenticated;      /* when requirepass is non-NULL */
+    int replstate;          /* replication state if this is a slave */
+    int repldbfd;           /* replication DB file descriptor */
+    long repldboff;          /* replication DB file offset */
+    off_t repldbsize;       /* replication DB file size */
 } redisClient;
 
 struct saveparam {
@@ -229,7 +242,7 @@ struct redisServer {
     int isslave;
     char *masterhost;
     int masterport;
-    redisClient *master;
+    redisClient *master;    /* client that is master for this slave */
     int replstate;
     /* Sort parameters - qsort_r() is only available under BSD so we
      * have to take this state global, in order to pass it to sortCompare() */
@@ -291,6 +304,7 @@ static int deleteIfVolatile(redisDb *db, robj *key);
 static int deleteKey(redisDb *db, robj *key);
 static time_t getExpire(redisDb *db, robj *key);
 static int setExpire(redisDb *db, robj *key, time_t when);
+static void updateSalvesWaitingBgsave(int bgsaveerr);
 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);
@@ -331,6 +345,8 @@ static void sismemberCommand(redisClient *c);
 static void scardCommand(redisClient *c);
 static void sinterCommand(redisClient *c);
 static void sinterstoreCommand(redisClient *c);
+static void sunionCommand(redisClient *c);
+static void sunionstoreCommand(redisClient *c);
 static void syncCommand(redisClient *c);
 static void flushdbCommand(redisClient *c);
 static void flushallCommand(redisClient *c);
@@ -370,6 +386,8 @@ static struct redisCommand cmdTable[] = {
     {"scard",scardCommand,2,REDIS_CMD_INLINE},
     {"sinter",sinterCommand,-2,REDIS_CMD_INLINE},
     {"sinterstore",sinterstoreCommand,-3,REDIS_CMD_INLINE},
+    {"sunion",sunionCommand,-2,REDIS_CMD_INLINE},
+    {"sunionstore",sunionstoreCommand,-3,REDIS_CMD_INLINE},
     {"smembers",sinterCommand,2,REDIS_CMD_INLINE},
     {"incrby",incrbyCommand,3,REDIS_CMD_INLINE},
     {"decrby",decrbyCommand,3,REDIS_CMD_INLINE},
@@ -616,13 +634,11 @@ static void oom(const char *msg) {
 /* ====================== Redis server networking stuff ===================== */
 void closeTimedoutClients(void) {
     redisClient *c;
-    listIter *li;
     listNode *ln;
     time_t now = time(NULL);
 
-    li = listGetIterator(server.clients,AL_START_HEAD);
-    if (!li) return;
-    while ((ln = listNextElement(li)) != NULL) {
+    listRewind(server.clients);
+    while ((ln = listYield(server.clients)) != NULL) {
         c = listNodeValue(ln);
         if (!(c->flags & REDIS_SLAVE) &&    /* no timeout for slaves */
              (now - c->lastinteraction > server.maxidletime)) {
@@ -630,7 +646,6 @@ void closeTimedoutClients(void) {
             freeClient(c);
         }
     }
-    listReleaseIterator(li);
 }
 
 int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
@@ -678,6 +693,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     /* Check if a background saving in progress terminated */
     if (server.bgsaveinprogress) {
         int statloc;
+        /* XXX: TODO handle the case of the saving child killed */
         if (wait4(-1,&statloc,WNOHANG,NULL)) {
             int exitcode = WEXITSTATUS(statloc);
             if (exitcode == 0) {
@@ -690,6 +706,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
                     "Background saving error");
             }
             server.bgsaveinprogress = 0;
+            updateSalvesWaitingBgsave(exitcode == 0 ? REDIS_OK : REDIS_ERR);
         }
     } else {
         /* If there is not a background saving in progress check if
@@ -1024,6 +1041,8 @@ static void freeClient(redisClient *c) {
     assert(ln != NULL);
     listDelNode(server.clients,ln);
     if (c->flags & REDIS_SLAVE) {
+        if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1)
+            close(c->repldbfd);
         list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves;
         ln = listSearchKey(l,c);
         assert(ln != NULL);
@@ -1038,13 +1057,13 @@ static void freeClient(redisClient *c) {
 
 static void glueReplyBuffersIfNeeded(redisClient *c) {
     int totlen = 0;
-    listNode *ln = c->reply->head, *next;
+    listNode *ln;
     robj *o;
 
-    while(ln) {
+    listRewind(c->reply);
+    while((ln = listYield(c->reply))) {
         o = ln->value;
         totlen += sdslen(o->ptr);
-        ln = ln->next;
         /* This optimization makes more sense if we don't have to copy
          * too much data */
         if (totlen > 1024) return;
@@ -1053,14 +1072,12 @@ static void glueReplyBuffersIfNeeded(redisClient *c) {
         char buf[1024];
         int copylen = 0;
 
-        ln = c->reply->head;
-        while(ln) {
-            next = ln->next;
+        listRewind(c->reply);
+        while((ln = listYield(c->reply))) {
             o = ln->value;
             memcpy(buf+copylen,o->ptr,sdslen(o->ptr));
             copylen += sdslen(o->ptr);
             listDelNode(c->reply,ln);
-            ln = next;
         }
         /* Now the output buffer is empty, add the new single element */
         addReplySds(c,sdsnewlen(buf,totlen));
@@ -1214,7 +1231,7 @@ static int processCommand(redisClient *c) {
 }
 
 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc) {
-    listNode *ln = slaves->head;
+    listNode *ln;
     robj *outv[REDIS_MAX_ARGS*4]; /* enough room for args, spaces, newlines */
     int outc = 0, j;
     
@@ -1232,8 +1249,18 @@ static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int di
     }
     outv[outc++] = shared.crlf;
 
-    while(ln) {
+    /* Increment all the refcounts at start and decrement at end in order to
+     * be sure to free objects if there is no slave in a replication state
+     * able to be feed with commands */
+    for (j = 0; j < outc; j++) incrRefCount(outv[j]);
+    listRewind(slaves);
+    while((ln = listYield(slaves))) {
         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 all the other slaves, MONITORs and so on */
         if (slave->slaveseldb != dictid) {
             robj *selectcmd;
 
@@ -1258,18 +1285,18 @@ static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int di
             slave->slaveseldb = dictid;
         }
         for (j = 0; j < outc; j++) addReply(slave,outv[j]);
-        ln = ln->next;
     }
+    for (j = 0; j < outc; j++) decrRefCount(outv[j]);
 }
 
 static void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     redisClient *c = (redisClient*) privdata;
-    char buf[REDIS_QUERYBUF_LEN];
+    char buf[REDIS_IOBUF_LEN];
     int nread;
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
 
-    nread = read(fd, buf, REDIS_QUERYBUF_LEN);
+    nread = read(fd, buf, REDIS_IOBUF_LEN);
     if (nread == -1) {
         if (errno == EAGAIN) {
             nread = 0;
@@ -1363,6 +1390,11 @@ static int selectDb(redisClient *c, int id) {
     return REDIS_OK;
 }
 
+static void *dupClientReplyValue(void *o) {
+    incrRefCount((robj*)o);
+    return 0;
+}
+
 static redisClient *createClient(int fd) {
     redisClient *c = zmalloc(sizeof(*c));
 
@@ -1378,8 +1410,10 @@ static redisClient *createClient(int fd) {
     c->flags = 0;
     c->lastinteraction = time(NULL);
     c->authenticated = 0;
+    c->replstate = REDIS_REPL_NONE;
     if ((c->reply = listCreate()) == NULL) oom("listCreate");
     listSetFreeMethod(c->reply,decrRefCount);
+    listSetDupMethod(c->reply,dupClientReplyValue);
     if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
         readQueryFromClient, c, NULL) == AE_ERR) {
         freeClient(c);
@@ -1391,6 +1425,8 @@ static redisClient *createClient(int fd) {
 
 static void addReply(redisClient *c, robj *obj) {
     if (listLength(c->reply) == 0 &&
+        (c->replstate == REDIS_REPL_NONE ||
+         c->replstate == REDIS_REPL_ONLINE) &&
         aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,
         sendReplyToClient, c, NULL) == AE_ERR) return;
     if (!listAddNodeTail(c->reply,obj)) oom("listAddNodeTail");
@@ -1767,14 +1803,14 @@ static int rdbSave(char *filename) {
             } else if (o->type == REDIS_LIST) {
                 /* Save a list value */
                 list *list = o->ptr;
-                listNode *ln = list->head;
+                listNode *ln;
 
+                listRewind(list);
                 if (rdbSaveLen(fp,listLength(list)) == -1) goto werr;
-                while(ln) {
+                while((ln = listYield(list))) {
                     robj *eleobj = listNodeValue(ln);
 
                     if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
-                    ln = ln->next;
                 }
             } else if (o->type == REDIS_SET) {
                 /* Save a set value */
@@ -2336,6 +2372,7 @@ static void bgsaveCommand(redisClient *c) {
 
 static void shutdownCommand(redisClient *c) {
     redisLog(REDIS_WARNING,"User requested shutdown, saving DB...");
+    /* XXX: TODO kill the child if there is a bgsave in progress */
     if (rdbSave(server.dbfilename) == REDIS_OK) {
         if (server.daemonize) {
           unlink(server.pidfile);
@@ -2861,7 +2898,6 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, r
         deleteKey(c->db,dstkey);
         dictAdd(c->db->dict,dstkey,dstset);
         incrRefCount(dstkey);
-        server.dirty++;
     }
 
     /* Iterate all the elements of the first (smallest) set, and test
@@ -2886,15 +2922,16 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, r
         } else {
             dictAdd(dstset->ptr,ele,NULL);
             incrRefCount(ele);
-            server.dirty++;
         }
     }
     dictReleaseIterator(di);
 
-    if (!dstkey)
+    if (!dstkey) {
         lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
-    else
+    } else {
         addReply(c,shared.ok);
+        server.dirty++;
+    }
     zfree(dv);
 }
 
@@ -2906,6 +2943,100 @@ static void sinterstoreCommand(redisClient *c) {
     sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
 }
 
+static void sunionGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey) {
+    dict **dv = zmalloc(sizeof(dict*)*setsnum);
+    dictIterator *di;
+    dictEntry *de;
+    robj *lenobj = NULL, *dstset = NULL;
+    int j, cardinality = 0;
+
+    if (!dv) oom("sunionCommand");
+    for (j = 0; j < setsnum; j++) {
+        robj *setobj;
+
+        setobj = dstkey ?
+                    lookupKeyWrite(c->db,setskeys[j]) :
+                    lookupKeyRead(c->db,setskeys[j]);
+        if (!setobj) {
+            dv[j] = NULL;
+            continue;
+        }
+        if (setobj->type != REDIS_SET) {
+            zfree(dv);
+            addReply(c,shared.wrongtypeerr);
+            return;
+        }
+        dv[j] = setobj->ptr;
+    }
+
+    /* We need a temp set object to store our union. If the dstkey
+     * is not NULL (that is, we are inside an SUNIONSTORE operation) then
+     * this set object will be the resulting object to set into the target key*/
+    dstset = createSetObject();
+
+    /* The first thing we should output is the total number of elements...
+     * since this is a multi-bulk write, but at this stage we don't know
+     * the intersection set size, so we use a trick, append an empty object
+     * to the output list and save the pointer to later modify it with the
+     * right length */
+    if (!dstkey) {
+        lenobj = createObject(REDIS_STRING,NULL);
+        addReply(c,lenobj);
+        decrRefCount(lenobj);
+    } else {
+        /* If we have a target key where to store the resulting set
+         * create this key with an empty set inside */
+        deleteKey(c->db,dstkey);
+        dictAdd(c->db->dict,dstkey,dstset);
+        incrRefCount(dstkey);
+        server.dirty++;
+    }
+
+    /* Iterate all the elements of all the sets, add every element a single
+     * time to the result set */
+    for (j = 0; j < setsnum; j++) {
+        if (!dv[j]) continue; /* non existing keys are like empty sets */
+
+        di = dictGetIterator(dv[j]);
+        if (!di) oom("dictGetIterator");
+
+        while((de = dictNext(di)) != NULL) {
+            robj *ele;
+
+            /* dictAdd will not add the same element multiple times */
+            ele = dictGetEntryKey(de);
+            if (dictAdd(dstset->ptr,ele,NULL) == DICT_OK) {
+                incrRefCount(ele);
+                if (!dstkey) {
+                    addReplySds(c,sdscatprintf(sdsempty(),
+                            "$%d\r\n",sdslen(ele->ptr)));
+                    addReply(c,ele);
+                    addReply(c,shared.crlf);
+                    cardinality++;
+                }
+            }
+        }
+        dictReleaseIterator(di);
+    }
+
+    if (!dstkey) {
+        lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
+        decrRefCount(dstset);
+    } else {
+        addReply(c,shared.ok);
+        server.dirty++;
+    }
+    zfree(dv);
+}
+
+static void sunionCommand(redisClient *c) {
+    sunionGenericCommand(c,c->argv+1,c->argc-1,NULL);
+}
+
+static void sunionstoreCommand(redisClient *c) {
+    sunionGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
+}
+
 static void flushdbCommand(redisClient *c) {
     dictEmpty(c->db->dict);
     dictEmpty(c->db->expires);
@@ -3093,13 +3224,14 @@ static void sortCommand(redisClient *c) {
     j = 0;
     if (sortval->type == REDIS_LIST) {
         list *list = sortval->ptr;
-        listNode *ln = list->head;
-        while(ln) {
+        listNode *ln;
+
+        listRewind(list);
+        while((ln = listYield(list))) {
             robj *ele = ln->value;
             vector[j].obj = ele;
             vector[j].u.score = 0;
             vector[j].u.cmpobj = NULL;
-            ln = ln->next;
             j++;
         }
     } else {
@@ -3161,14 +3293,15 @@ static void sortCommand(redisClient *c) {
     outputlen = getop ? getop*(end-start+1) : end-start+1;
     addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",outputlen));
     for (j = start; j <= end; j++) {
-        listNode *ln = operations->head;
+        listNode *ln;
         if (!getop) {
             addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",
                 sdslen(vector[j].obj->ptr)));
             addReply(c,vector[j].obj);
             addReply(c,shared.crlf);
         }
-        while(ln) {
+        listRewind(operations);
+        while((ln = listYield(operations))) {
             redisSortOperation *sop = ln->value;
             robj *val = lookupKeyByPattern(c->db,sop->pattern,
                 vector[j].obj);
@@ -3185,7 +3318,6 @@ static void sortCommand(redisClient *c) {
             } else if (sop->type == REDIS_SORT_DEL) {
                 /* TODO */
             }
-            ln = ln->next;
         }
     }
 
@@ -3324,23 +3456,6 @@ static void expireCommand(redisClient *c) {
 
 /* =============================== Replication  ============================= */
 
-/* Send the whole output buffer syncronously to the slave. This a general operation in theory, but it is actually useful only for replication. */
-static int flushClientOutput(redisClient *c) {
-    int retval;
-    time_t start = time(NULL);
-
-    while(listLength(c->reply)) {
-        if (time(NULL)-start > 5) return REDIS_ERR; /* 5 seconds timeout */
-        retval = aeWait(c->fd,AE_WRITABLE,1000);
-        if (retval == -1) {
-            return REDIS_ERR;
-        } else if (retval & AE_WRITABLE) {
-            sendReplyToClient(NULL, c->fd, c, AE_WRITABLE);
-        }
-    }
-    return REDIS_OK;
-}
-
 static int syncWrite(int fd, char *ptr, ssize_t size, int timeout) {
     ssize_t nwritten, ret = size;
     time_t start = time(NULL);
@@ -3404,48 +3519,165 @@ static int syncReadLine(int fd, char *ptr, ssize_t size, int timeout) {
 }
 
 static void syncCommand(redisClient *c) {
-    struct stat sb;
-    int fd = -1, len;
-    time_t start = time(NULL);
-    char sizebuf[32];
-
     /* ignore SYNC if aleady slave or in monitor mode */
     if (c->flags & REDIS_SLAVE) return;
 
-    redisLog(REDIS_NOTICE,"Slave ask for syncronization");
-    if (flushClientOutput(c) == REDIS_ERR ||
-        rdbSave(server.dbfilename) != REDIS_OK)
-        goto closeconn;
-
-    fd = open(server.dbfilename, O_RDONLY);
-    if (fd == -1 || fstat(fd,&sb) == -1) goto closeconn;
-    len = sb.st_size;
-
-    snprintf(sizebuf,32,"$%d\r\n",len);
-    if (syncWrite(c->fd,sizebuf,strlen(sizebuf),5) == -1) goto closeconn;
-    while(len) {
-        char buf[1024];
-        int nread;
+    /* 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) {
+        addReplySds(c,sdsnew("-ERR SYNC is invalid with pending input\r\n"));
+        return;
+    }
 
-        if (time(NULL)-start > REDIS_MAX_SYNC_TIME) goto closeconn;
-        nread = read(fd,buf,1024);
-        if (nread == -1) goto closeconn;
-        len -= nread;
-        if (syncWrite(c->fd,buf,nread,5) == -1) goto closeconn;
+    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.bgsaveinprogress) {
+        /* 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;
+
+        listRewind(server.slaves);
+        while((ln = listYield(server.slaves))) {
+            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);
+            if (!c->reply) oom("listDup copying slave reply list");
+            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");
+            addReplySds(c,sdsnew("-ERR Unalbe to perform background save\r\n"));
+            return;
+        }
+        c->replstate = REDIS_REPL_WAIT_BGSAVE_END;
     }
-    if (syncWrite(c->fd,"\r\n",2,5) == -1) goto closeconn;
-    close(fd);
+    c->repldbfd = -1;
     c->flags |= REDIS_SLAVE;
     c->slaveseldb = 0;
     if (!listAddNodeTail(server.slaves,c)) oom("listAddNodeTail");
-    redisLog(REDIS_NOTICE,"Syncronization with slave succeeded");
     return;
+}
 
-closeconn:
-    if (fd != -1) close(fd);
-    c->flags |= REDIS_CLOSE;
-    redisLog(REDIS_WARNING,"Syncronization with slave failed");
-    return;
+static 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_DEBUG,"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, NULL) == AE_ERR) {
+            freeClient(slave);
+            return;
+        }
+        addReplySds(slave,sdsempty());
+        redisLog(REDIS_NOTICE,"Synchronization with slave succeeded");
+    }
+}
+
+static void updateSalvesWaitingBgsave(int bgsaveerr) {
+    listNode *ln;
+    int startbgsave = 0;
+
+    listRewind(server.slaves);
+    while((ln = listYield(server.slaves))) {
+        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 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 ||
+                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, NULL) == AE_ERR) {
+                freeClient(slave);
+                continue;
+            }
+        }
+    }
+    if (startbgsave) {
+        if (rdbSaveBackground(server.dbfilename) != REDIS_OK) {
+            listRewind(server.slaves);
+            redisLog(REDIS_WARNING,"SYNC failed. BGSAVE failed");
+            while((ln = listYield(server.slaves))) {
+                redisClient *slave = ln->value;
+
+                if (slave->replstate == REDIS_REPL_WAIT_BGSAVE_START)
+                    freeClient(slave);
+            }
+        }
+    }
 }
 
 static int syncWithMaster(void) {