X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/913e9d6bcacfa28ace3b7ee1c75f8ec4e146974b..d0ccebcf46a2f726117b05f2a317b0a5e7113cae:/redis.c

diff --git a/redis.c b/redis.c
index 7158de60..21098e56 100644
--- a/redis.c
+++ b/redis.c
@@ -57,6 +57,11 @@
 #include <sys/time.h>
 #include <sys/resource.h>
 #include <limits.h>
+#include <math.h>
+
+#if defined(__sun)
+#include "solarisfixes.h"
+#endif
 
 #include "redis.h"
 #include "ae.h"     /* Event driven programming library */
@@ -167,11 +172,8 @@
 
 /* Sort operations */
 #define REDIS_SORT_GET 0
-#define REDIS_SORT_DEL 1
-#define REDIS_SORT_INCR 2
-#define REDIS_SORT_DECR 3
-#define REDIS_SORT_ASC 4
-#define REDIS_SORT_DESC 5
+#define REDIS_SORT_ASC 1
+#define REDIS_SORT_DESC 2
 #define REDIS_SORTKEY_MAX 1024
 
 /* Log levels */
@@ -185,6 +187,11 @@
 #define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^32 elements */
 #define ZSKIPLIST_P 0.25      /* Skiplist P = 1/4 */
 
+/* Append only defines */
+#define APPENDFSYNC_NO 0
+#define APPENDFSYNC_ALWAYS 1
+#define APPENDFSYNC_EVERYSEC 2
+
 /*================================= Data types ============================== */
 
 /* A redis object, that is a type able to hold a string / list / set */
@@ -256,6 +263,11 @@ struct redisServer {
     int maxidletime;
     int dbnum;
     int daemonize;
+    int appendonly;
+    int appendfsync;
+    time_t lastfsync;
+    int appendfd;
+    int appendseldb;
     char *pidfile;
     int bgsaveinprogress;
     pid_t bgsavechildpid;
@@ -264,10 +276,12 @@ struct redisServer {
     char *logfile;
     char *bindaddr;
     char *dbfilename;
+    char *appendfilename;
     char *requirepass;
     int shareobjects;
     /* Replication related */
     int isslave;
+    char *masterauth;
     char *masterhost;
     int masterport;
     redisClient *master;    /* client that is master for this slave */
@@ -311,13 +325,14 @@ typedef struct _redisSortOperation {
 
 typedef struct zskiplistNode {
     struct zskiplistNode **forward;
+    struct zskiplistNode *backward;
     double score;
     robj *obj;
 } zskiplistNode;
 
 typedef struct zskiplist {
-    struct zskiplistNode *header;
-    long length;
+    struct zskiplistNode *header, *tail;
+    unsigned long length;
     int level;
 } zskiplist;
 
@@ -337,6 +352,12 @@ struct sharedObjectsStruct {
     *select5, *select6, *select7, *select8, *select9;
 } shared;
 
+/* Global vars that are actally used as constants. The following double
+ * values are used for double on-disk serialization, and are initialized
+ * at runtime to avoid strange compiler optimizations. */
+
+static double R_Zero, R_PosInf, R_NegInf, R_Nan;
+
 /*================================ Prototypes =============================== */
 
 static void freeStringObject(robj *o);
@@ -352,6 +373,7 @@ static void incrRefCount(robj *o);
 static int rdbSaveBackground(char *filename);
 static robj *createStringObject(char *ptr, size_t len);
 static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int dictid, robj **argv, int argc);
+static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc);
 static int syncWithMaster(void);
 static robj *tryObjectSharing(robj *o);
 static int tryObjectEncoding(robj *o);
@@ -371,6 +393,7 @@ static size_t stringObjectLen(robj *o);
 static void processInputBuffer(redisClient *c);
 static zskiplist *zslCreate(void);
 static void zslFree(zskiplist *zsl);
+static void zslInsert(zskiplist *zsl, double score, robj *obj);
 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);
@@ -427,6 +450,7 @@ static void infoCommand(redisClient *c);
 static void mgetCommand(redisClient *c);
 static void monitorCommand(redisClient *c);
 static void expireCommand(redisClient *c);
+static void expireatCommand(redisClient *c);
 static void getsetCommand(redisClient *c);
 static void ttlCommand(redisClient *c);
 static void slaveofCommand(redisClient *c);
@@ -434,6 +458,13 @@ static void debugCommand(redisClient *c);
 static void msetCommand(redisClient *c);
 static void msetnxCommand(redisClient *c);
 static void zaddCommand(redisClient *c);
+static void zrangeCommand(redisClient *c);
+static void zrangebyscoreCommand(redisClient *c);
+static void zrevrangeCommand(redisClient *c);
+static void zcardCommand(redisClient *c);
+static void zremCommand(redisClient *c);
+static void zscoreCommand(redisClient *c);
+static void zremrangebyscoreCommand(redisClient *c);
 
 /*================================= Globals ================================= */
 
@@ -473,6 +504,13 @@ static struct redisCommand cmdTable[] = {
     {"sdiffstore",sdiffstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
     {"smembers",sinterCommand,2,REDIS_CMD_INLINE},
     {"zadd",zaddCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
+    {"zrem",zremCommand,3,REDIS_CMD_BULK},
+    {"zremrangebyscore",zremrangebyscoreCommand,4,REDIS_CMD_INLINE},
+    {"zrange",zrangeCommand,4,REDIS_CMD_INLINE},
+    {"zrangebyscore",zrangebyscoreCommand,4,REDIS_CMD_INLINE},
+    {"zrevrange",zrevrangeCommand,4,REDIS_CMD_INLINE},
+    {"zcard",zcardCommand,2,REDIS_CMD_INLINE},
+    {"zscore",zscoreCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
     {"incrby",incrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
     {"decrby",decrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
     {"getset",getsetCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
@@ -484,6 +522,7 @@ static struct redisCommand cmdTable[] = {
     {"rename",renameCommand,3,REDIS_CMD_INLINE},
     {"renamenx",renamenxCommand,3,REDIS_CMD_INLINE},
     {"expire",expireCommand,3,REDIS_CMD_INLINE},
+    {"expireat",expireatCommand,3,REDIS_CMD_INLINE},
     {"keys",keysCommand,2,REDIS_CMD_INLINE},
     {"dbsize",dbsizeCommand,1,REDIS_CMD_INLINE},
     {"auth",authCommand,2,REDIS_CMD_INLINE},
@@ -644,7 +683,7 @@ static void redisLog(int level, const char *fmt, ...) {
         time_t now;
 
         now = time(NULL);
-        strftime(buf,64,"%d %b %H:%M:%S",gmtime(&now));
+        strftime(buf,64,"%d %b %H:%M:%S",localtime(&now));
         fprintf(fp,"%s %c ",buf,c[level]);
         vfprintf(fp, fmt, ap);
         fprintf(fp,"\n");
@@ -994,8 +1033,14 @@ static void initServerConfig() {
     server.bindaddr = NULL;
     server.glueoutputbuf = 1;
     server.daemonize = 0;
+    server.appendonly = 0;
+    server.appendfsync = APPENDFSYNC_ALWAYS;
+    server.lastfsync = time(NULL);
+    server.appendfd = -1;
+    server.appendseldb = -1; /* Make sure the first time will not match */
     server.pidfile = "/var/run/redis.pid";
     server.dbfilename = "dump.rdb";
+    server.appendfilename = "appendonly.log";
     server.requirepass = NULL;
     server.shareobjects = 0;
     server.sharingpoolsize = 1024;
@@ -1008,10 +1053,17 @@ static void initServerConfig() {
     appendServerSaveParams(60,10000); /* save after 1 minute and 10000 changes */
     /* Replication related */
     server.isslave = 0;
+    server.masterauth = NULL;
     server.masterhost = NULL;
     server.masterport = 6379;
     server.master = NULL;
     server.replstate = REDIS_REPL_NONE;
+
+    /* Double constants initialization */
+    R_Zero = 0.0;
+    R_PosInf = 1.0/R_Zero;
+    R_NegInf = -1.0/R_Zero;
+    R_Nan = R_Zero/R_Zero;
 }
 
 static void initServer() {
@@ -1048,7 +1100,16 @@ static void initServer() {
     server.stat_numcommands = 0;
     server.stat_numconnections = 0;
     server.stat_starttime = time(NULL);
-    aeCreateTimeEvent(server.el, 1000, serverCron, NULL, NULL);
+    aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
+
+    if (server.appendonly) {
+        server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644);
+        if (server.appendfd == -1) {
+            redisLog(REDIS_WARNING, "Can't open the append-only file: %s",
+                strerror(errno));
+            exit(1);
+        }
+    }
 }
 
 /* Empty the whole database */
@@ -1171,6 +1232,8 @@ static void loadServerConfig(char *filename) {
             server.masterhost = sdsnew(argv[1]);
             server.masterport = atoi(argv[2]);
             server.replstate = REDIS_REPL_CONNECT;
+        } else if (!strcasecmp(argv[0],"masterauth") && argc == 2) {
+        	server.masterauth = zstrdup(argv[1]);
         } else if (!strcasecmp(argv[0],"glueoutputbuf") && argc == 2) {
             if ((server.glueoutputbuf = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
@@ -1188,6 +1251,21 @@ static void loadServerConfig(char *filename) {
             if ((server.daemonize = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }
+        } else if (!strcasecmp(argv[0],"appendonly") && argc == 2) {
+            if ((server.appendonly = yesnotoi(argv[1])) == -1) {
+                err = "argument must be 'yes' or 'no'"; goto loaderr;
+            }
+        } else if (!strcasecmp(argv[0],"appendfsync") && argc == 2) {
+            if (!strcasecmp(argv[1],"no")) {
+                server.appendfsync = APPENDFSYNC_NO;
+            } else if (!strcasecmp(argv[1],"always")) {
+                server.appendfsync = APPENDFSYNC_ALWAYS;
+            } else if (!strcasecmp(argv[1],"everysec")) {
+                server.appendfsync = APPENDFSYNC_EVERYSEC;
+            } else {
+                err = "argument must be 'no', 'always' or 'everysec'";
+                goto loaderr;
+            }
         } else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {
           server.requirepass = zstrdup(argv[1]);
         } else if (!strcasecmp(argv[0],"pidfile") && argc == 2) {
@@ -1503,7 +1581,9 @@ static int processCommand(redisClient *c) {
     /* Exec the command */
     dirty = server.dirty;
     cmd->proc(c);
-    if (server.dirty-dirty != 0 && listLength(server.slaves))
+    if (server.appendonly && server.dirty-dirty)
+        feedAppendOnlyFile(cmd,c->db->id,c->argv,c->argc);
+    if (server.dirty-dirty && listLength(server.slaves))
         replicationFeedSlaves(server.slaves,cmd,c->db->id,c->argv,c->argc);
     if (listLength(server.monitors))
         replicationFeedSlaves(server.monitors,cmd,c->db->id,c->argv,c->argc);
@@ -2047,23 +2127,31 @@ static robj *getDecodedObject(const robj *o) {
     }
 }
 
+/* Compare two string objects via strcmp() or alike.
+ * Note that the objects may be integer-encoded. In such a case we
+ * use snprintf() to get a string representation of the numbers on the stack
+ * and compare the strings, it's much faster than calling getDecodedObject(). */
 static int compareStringObjects(robj *a, robj *b) {
     assert(a->type == REDIS_STRING && b->type == REDIS_STRING);
-
-    if (a->encoding == REDIS_ENCODING_INT && b->encoding == REDIS_ENCODING_INT){
-        return (long)a->ptr - (long)b->ptr;
+    char bufa[128], bufb[128], *astr, *bstr;
+    int bothsds = 1;
+
+    if (a == b) return 0;
+    if (a->encoding != REDIS_ENCODING_RAW) {
+        snprintf(bufa,sizeof(bufa),"%ld",(long) a->ptr);
+        astr = bufa;
+        bothsds = 0;
     } else {
-        int retval;
-
-        incrRefCount(a);
-        incrRefCount(b);
-        if (a->encoding != REDIS_ENCODING_RAW) a = getDecodedObject(a);
-        if (b->encoding != REDIS_ENCODING_RAW) b = getDecodedObject(a);
-        retval = sdscmp(a->ptr,b->ptr);
-        decrRefCount(a);
-        decrRefCount(b);
-        return retval;
+        astr = a->ptr;
+    }
+    if (b->encoding != REDIS_ENCODING_RAW) {
+        snprintf(bufb,sizeof(bufb),"%ld",(long) b->ptr);
+        bstr = bufb;
+        bothsds = 0;
+    } else {
+        bstr = b->ptr;
     }
+    return bothsds ? sdscmp(astr,bstr) : strcmp(astr,bstr);
 }
 
 static size_t stringObjectLen(robj *o) {
@@ -2228,6 +2316,33 @@ static int rdbSaveStringObject(FILE *fp, robj *obj) {
     }
 }
 
+/* Save a double value. Doubles are saved as strings prefixed by an unsigned
+ * 8 bit integer specifing the length of the representation.
+ * This 8 bit integer has special values in order to specify the following
+ * conditions:
+ * 253: not a number
+ * 254: + inf
+ * 255: - inf
+ */
+static int rdbSaveDoubleValue(FILE *fp, double val) {
+    unsigned char buf[128];
+    int len;
+
+    if (isnan(val)) {
+        buf[0] = 253;
+        len = 1;
+    } else if (!isfinite(val)) {
+        len = 1;
+        buf[0] = (val < 0) ? 255 : 254;
+    } else {
+        snprintf((char*)buf+1,sizeof(buf)-1,"%.17g",val);
+        buf[0] = strlen((char*)buf);
+        len = buf[0]+1;
+    }
+    if (fwrite(buf,len,1,fp) == 0) return -1;
+    return 0;
+}
+
 /* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */
 static int rdbSave(char *filename) {
     dictIterator *di = NULL;
@@ -2302,6 +2417,21 @@ static int rdbSave(char *filename) {
                     if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
                 }
                 dictReleaseIterator(di);
+            } else if (o->type == REDIS_ZSET) {
+                /* Save a set value */
+                zset *zs = o->ptr;
+                dictIterator *di = dictGetIterator(zs->dict);
+                dictEntry *de;
+
+                if (rdbSaveLen(fp,dictSize(zs->dict)) == -1) goto werr;
+                while((de = dictNext(di)) != NULL) {
+                    robj *eleobj = dictGetEntryKey(de);
+                    double *score = dictGetEntryVal(de);
+
+                    if (rdbSaveStringObject(fp,eleobj) == -1) goto werr;
+                    if (rdbSaveDoubleValue(fp,*score) == -1) goto werr;
+                }
+                dictReleaseIterator(di);
             } else {
                 assert(0 != 0);
             }
@@ -2319,7 +2449,7 @@ static int rdbSave(char *filename) {
     /* Use RENAME to make sure the DB file is changed atomically only
      * if the generate DB file is ok. */
     if (rename(tmpfile,filename) == -1) {
-        redisLog(REDIS_WARNING,"Error moving temp DB file on the final destionation: %s", strerror(errno));
+        redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno));
         unlink(tmpfile);
         return REDIS_ERR;
     }
@@ -2490,6 +2620,23 @@ static robj *rdbLoadStringObject(FILE*fp, int rdbver) {
     return tryObjectSharing(createObject(REDIS_STRING,val));
 }
 
+/* For information about double serialization check rdbSaveDoubleValue() */
+static int rdbLoadDoubleValue(FILE *fp, double *val) {
+    char buf[128];
+    unsigned char len;
+
+    if (fread(&len,1,1,fp) == 0) return -1;
+    switch(len) {
+    case 255: *val = R_NegInf; return 0;
+    case 254: *val = R_PosInf; return 0;
+    case 253: *val = R_Nan; return 0;
+    default:
+        if (fread(buf,len,1,fp) == 0) return -1;
+        sscanf(buf, "%lg", val);
+        return 0;
+    }
+}
+
 static int rdbLoad(char *filename) {
     FILE *fp;
     robj *keyobj = NULL;
@@ -2564,6 +2711,27 @@ static int rdbLoad(char *filename) {
                     dictAdd((dict*)o->ptr,ele,NULL);
                 }
             }
+        } else if (type == REDIS_ZSET) {
+            /* Read list/set value */
+            uint32_t zsetlen;
+            zset *zs;
+
+            if ((zsetlen = rdbLoadLen(fp,rdbver,NULL)) == REDIS_RDB_LENERR)
+                goto eoferr;
+            o = createZsetObject();
+            zs = o->ptr;
+            /* Load every single element of the list/set */
+            while(zsetlen--) {
+                robj *ele;
+                double *score = zmalloc(sizeof(double));
+
+                if ((ele = rdbLoadStringObject(fp,rdbver)) == NULL) goto eoferr;
+                tryObjectEncoding(ele);
+                if (rdbLoadDoubleValue(fp,score) == -1) goto eoferr;
+                dictAdd(zs->dict,ele,score);
+                zslInsert(zs->zsl,*score,ele);
+                incrRefCount(ele); /* added to skiplist */
+            }
         } else {
             assert(0 != 0);
         }
@@ -2587,7 +2755,7 @@ static int rdbLoad(char *filename) {
 
 eoferr: /* unexpected end of file is handled here with a fatal exit */
     if (keyobj) decrRefCount(keyobj);
-    redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, exiting now.");
+    redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
     exit(1);
     return REDIS_ERR; /* Just to avoid warning */
 }
@@ -2600,7 +2768,7 @@ static void authCommand(redisClient *c) {
       addReply(c,shared.ok);
     } else {
       c->authenticated = 0;
-      addReply(c,shared.err);
+      addReplySds(c,sdscatprintf(sdsempty(),"-ERR invalid password\r\n"));
     }
 }
 
@@ -2858,6 +3026,7 @@ static void typeCommand(redisClient *c) {
         case REDIS_STRING: type = "+string"; break;
         case REDIS_LIST: type = "+list"; break;
         case REDIS_SET: type = "+set"; break;
+        case REDIS_ZSET: type = "+zset"; break;
         default: type = "unknown"; break;
         }
     }
@@ -3710,25 +3879,32 @@ static zskiplist *zslCreate(void) {
     
     zsl = zmalloc(sizeof(*zsl));
     zsl->level = 1;
+    zsl->length = 0;
     zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
     for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++)
         zsl->header->forward[j] = NULL;
+    zsl->header->backward = NULL;
+    zsl->tail = NULL;
     return zsl;
 }
 
 static void zslFreeNode(zskiplistNode *node) {
     decrRefCount(node->obj);
+    zfree(node->forward);
     zfree(node);
 }
 
 static void zslFree(zskiplist *zsl) {
-    zskiplistNode *node = zsl->header->forward[1], *next;
+    zskiplistNode *node = zsl->header->forward[0], *next;
 
+    zfree(zsl->header->forward);
+    zfree(zsl->header);
     while(node) {
-        next = node->forward[1];
+        next = node->forward[0];
         zslFreeNode(node);
         node = next;
     }
+    zfree(zsl);
 }
 
 static int zslRandomLevel(void) {
@@ -3744,11 +3920,13 @@ static void zslInsert(zskiplist *zsl, double score, robj *obj) {
 
     x = zsl->header;
     for (i = zsl->level-1; i >= 0; i--) {
-        while (x->forward[i] && x->forward[i]->score < score)
+        while (x->forward[i] &&
+            (x->forward[i]->score < score ||
+                (x->forward[i]->score == score &&
+                compareStringObjects(x->forward[i]->obj,obj) < 0)))
             x = x->forward[i];
         update[i] = x;
     }
-    x = x->forward[1];
     /* we assume the key is not already inside, since we allow duplicated
      * scores, and the re-insertion of score and redis object should never
      * happpen since the caller of zslInsert() should test in the hash table
@@ -3764,10 +3942,110 @@ static void zslInsert(zskiplist *zsl, double score, robj *obj) {
         x->forward[i] = update[i]->forward[i];
         update[i]->forward[i] = x;
     }
+    x->backward = (update[0] == zsl->header) ? NULL : update[0];
+    if (x->forward[0])
+        x->forward[0]->backward = x;
+    else
+        zsl->tail = x;
+    zsl->length++;
 }
 
+/* Delete an element with matching score/object from the skiplist. */
 static int zslDelete(zskiplist *zsl, double score, robj *obj) {
-    return 1;
+    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
+    int i;
+
+    x = zsl->header;
+    for (i = zsl->level-1; i >= 0; i--) {
+        while (x->forward[i] &&
+            (x->forward[i]->score < score ||
+                (x->forward[i]->score == score &&
+                compareStringObjects(x->forward[i]->obj,obj) < 0)))
+            x = x->forward[i];
+        update[i] = x;
+    }
+    /* We may have multiple elements with the same score, what we need
+     * is to find the element with both the right score and object. */
+    x = x->forward[0];
+    if (x && score == x->score && compareStringObjects(x->obj,obj) == 0) {
+        for (i = 0; i < zsl->level; i++) {
+            if (update[i]->forward[i] != x) break;
+            update[i]->forward[i] = x->forward[i];
+        }
+        if (x->forward[0]) {
+            x->forward[0]->backward = (x->backward == zsl->header) ?
+                                        NULL : x->backward;
+        } else {
+            zsl->tail = x->backward;
+        }
+        zslFreeNode(x);
+        while(zsl->level > 1 && zsl->header->forward[zsl->level-1] == NULL)
+            zsl->level--;
+        zsl->length--;
+        return 1;
+    } else {
+        return 0; /* not found */
+    }
+    return 0; /* not found */
+}
+
+/* Delete all the elements with score between min and max from the skiplist.
+ * Min and mx are inclusive, so a score >= min || score <= max is deleted.
+ * Note that this function takes the reference to the hash table view of the
+ * sorted set, in order to remove the elements from the hash table too. */
+static unsigned long zslDeleteRange(zskiplist *zsl, double min, double max, dict *dict) {
+    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
+    unsigned long removed = 0;
+    int i;
+
+    x = zsl->header;
+    for (i = zsl->level-1; i >= 0; i--) {
+        while (x->forward[i] && x->forward[i]->score < min)
+            x = x->forward[i];
+        update[i] = x;
+    }
+    /* We may have multiple elements with the same score, what we need
+     * is to find the element with both the right score and object. */
+    x = x->forward[0];
+    while (x && x->score <= max) {
+        zskiplistNode *next;
+
+        for (i = 0; i < zsl->level; i++) {
+            if (update[i]->forward[i] != x) break;
+            update[i]->forward[i] = x->forward[i];
+        }
+        if (x->forward[0]) {
+            x->forward[0]->backward = (x->backward == zsl->header) ?
+                                        NULL : x->backward;
+        } else {
+            zsl->tail = x->backward;
+        }
+        next = x->forward[0];
+        dictDelete(dict,x->obj);
+        zslFreeNode(x);
+        while(zsl->level > 1 && zsl->header->forward[zsl->level-1] == NULL)
+            zsl->level--;
+        zsl->length--;
+        removed++;
+        x = next;
+    }
+    return removed; /* not found */
+}
+
+/* Find the first node having a score equal or greater than the specified one.
+ * Returns NULL if there is no match. */
+static zskiplistNode *zslFirstWithScore(zskiplist *zsl, double score) {
+    zskiplistNode *x;
+    int i;
+
+    x = zsl->header;
+    for (i = zsl->level-1; i >= 0; i--) {
+        while (x->forward[i] && x->forward[i]->score < score)
+            x = x->forward[i];
+    }
+    /* We may have multiple elements with the same score, what we need
+     * is to find the element with both the right score and object. */
+    return x->forward[0];
 }
 
 /* The actual Z-commands implementations */
@@ -3809,13 +4087,235 @@ static void zaddCommand(redisClient *c) {
         if (*score != *oldscore) {
             int deleted;
 
-            deleted = zslDelete(zs->zsl,*score,c->argv[3]);
+            deleted = zslDelete(zs->zsl,*oldscore,c->argv[3]);
             assert(deleted != 0);
             zslInsert(zs->zsl,*score,c->argv[3]);
             incrRefCount(c->argv[3]);
+            dictReplace(zs->dict,c->argv[3],score);
             server.dirty++;
+        } else {
+            zfree(score);
+        }
+        addReply(c,shared.czero);
+    }
+}
+
+static void zremCommand(redisClient *c) {
+    robj *zsetobj;
+    zset *zs;
+
+    zsetobj = lookupKeyWrite(c->db,c->argv[1]);
+    if (zsetobj == NULL) {
+        addReply(c,shared.czero);
+    } else {
+        dictEntry *de;
+        double *oldscore;
+        int deleted;
+
+        if (zsetobj->type != REDIS_ZSET) {
+            addReply(c,shared.wrongtypeerr);
+            return;
+        }
+        zs = zsetobj->ptr;
+        de = dictFind(zs->dict,c->argv[2]);
+        if (de == NULL) {
+            addReply(c,shared.czero);
+            return;
+        }
+        /* Delete from the skiplist */
+        oldscore = dictGetEntryVal(de);
+        deleted = zslDelete(zs->zsl,*oldscore,c->argv[2]);
+        assert(deleted != 0);
+
+        /* Delete from the hash table */
+        dictDelete(zs->dict,c->argv[2]);
+        if (htNeedsResize(zs->dict)) dictResize(zs->dict);
+        server.dirty++;
+        addReply(c,shared.cone);
+    }
+}
+
+static void zremrangebyscoreCommand(redisClient *c) {
+    double min = strtod(c->argv[2]->ptr,NULL);
+    double max = strtod(c->argv[3]->ptr,NULL);
+    robj *zsetobj;
+    zset *zs;
+
+    zsetobj = lookupKeyWrite(c->db,c->argv[1]);
+    if (zsetobj == NULL) {
+        addReply(c,shared.czero);
+    } else {
+        long deleted;
+
+        if (zsetobj->type != REDIS_ZSET) {
+            addReply(c,shared.wrongtypeerr);
+            return;
+        }
+        zs = zsetobj->ptr;
+        deleted = zslDeleteRange(zs->zsl,min,max,zs->dict);
+        if (htNeedsResize(zs->dict)) dictResize(zs->dict);
+        server.dirty += deleted;
+        addReplySds(c,sdscatprintf(sdsempty(),":%lu\r\n",deleted));
+    }
+}
+
+static void zrangeGenericCommand(redisClient *c, int reverse) {
+    robj *o;
+    int start = atoi(c->argv[2]->ptr);
+    int end = atoi(c->argv[3]->ptr);
+
+    o = lookupKeyRead(c->db,c->argv[1]);
+    if (o == NULL) {
+        addReply(c,shared.nullmultibulk);
+    } else {
+        if (o->type != REDIS_ZSET) {
+            addReply(c,shared.wrongtypeerr);
+        } else {
+            zset *zsetobj = o->ptr;
+            zskiplist *zsl = zsetobj->zsl;
+            zskiplistNode *ln;
+
+            int llen = zsl->length;
+            int rangelen, j;
+            robj *ele;
+
+            /* convert negative indexes */
+            if (start < 0) start = llen+start;
+            if (end < 0) end = llen+end;
+            if (start < 0) start = 0;
+            if (end < 0) end = 0;
+
+            /* indexes sanity checks */
+            if (start > end || start >= llen) {
+                /* Out of range start or start > end result in empty list */
+                addReply(c,shared.emptymultibulk);
+                return;
+            }
+            if (end >= llen) end = llen-1;
+            rangelen = (end-start)+1;
+
+            /* Return the result in form of a multi-bulk reply */
+            if (reverse) {
+                ln = zsl->tail;
+                while (start--)
+                    ln = ln->backward;
+            } else {
+                ln = zsl->header->forward[0];
+                while (start--)
+                    ln = ln->forward[0];
+            }
+
+            addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",rangelen));
+            for (j = 0; j < rangelen; j++) {
+                ele = ln->obj;
+                addReplyBulkLen(c,ele);
+                addReply(c,ele);
+                addReply(c,shared.crlf);
+                ln = reverse ? ln->backward : ln->forward[0];
+            }
+        }
+    }
+}
+
+static void zrangeCommand(redisClient *c) {
+    zrangeGenericCommand(c,0);
+}
+
+static void zrevrangeCommand(redisClient *c) {
+    zrangeGenericCommand(c,1);
+}
+
+static void zrangebyscoreCommand(redisClient *c) {
+    robj *o;
+    double min = strtod(c->argv[2]->ptr,NULL);
+    double max = strtod(c->argv[3]->ptr,NULL);
+
+    o = lookupKeyRead(c->db,c->argv[1]);
+    if (o == NULL) {
+        addReply(c,shared.nullmultibulk);
+    } else {
+        if (o->type != REDIS_ZSET) {
+            addReply(c,shared.wrongtypeerr);
+        } else {
+            zset *zsetobj = o->ptr;
+            zskiplist *zsl = zsetobj->zsl;
+            zskiplistNode *ln;
+            robj *ele, *lenobj;
+            unsigned int rangelen = 0;
+
+            /* Get the first node with the score >= min */
+            ln = zslFirstWithScore(zsl,min);
+            if (ln == NULL) {
+                /* No element matching the speciifed interval */
+                addReply(c,shared.emptymultibulk);
+                return;
+            }
+
+            /* We don't know in advance how many matching elements there
+             * are in the list, so we push this object that will represent
+             * the multi-bulk length in the output buffer, and will "fix"
+             * it later */
+            lenobj = createObject(REDIS_STRING,NULL);
+            addReply(c,lenobj);
+
+            while(ln && ln->score <= max) {
+                ele = ln->obj;
+                addReplyBulkLen(c,ele);
+                addReply(c,ele);
+                addReply(c,shared.crlf);
+                ln = ln->forward[0];
+                rangelen++;
+            }
+            lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",rangelen);
+        }
+    }
+}
+
+static void zcardCommand(redisClient *c) {
+    robj *o;
+    zset *zs;
+    
+    o = lookupKeyRead(c->db,c->argv[1]);
+    if (o == NULL) {
+        addReply(c,shared.czero);
+        return;
+    } else {
+        if (o->type != REDIS_ZSET) {
+            addReply(c,shared.wrongtypeerr);
+        } else {
+            zs = o->ptr;
+            addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",zs->zsl->length));
         }
+    }
+}
+
+static void zscoreCommand(redisClient *c) {
+    robj *o;
+    zset *zs;
+    
+    o = lookupKeyRead(c->db,c->argv[1]);
+    if (o == NULL) {
         addReply(c,shared.czero);
+        return;
+    } else {
+        if (o->type != REDIS_ZSET) {
+            addReply(c,shared.wrongtypeerr);
+        } else {
+            dictEntry *de;
+
+            zs = o->ptr;
+            de = dictFind(zs->dict,c->argv[2]);
+            if (!de) {
+                addReply(c,shared.nullbulk);
+            } else {
+                char buf[128];
+                double *score = dictGetEntryVal(de);
+
+                snprintf(buf,sizeof(buf),"%.17g",*score);
+                addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n%s\r\n",
+                    strlen(buf),buf));
+            }
+        }
     }
 }
 
@@ -3948,7 +4448,7 @@ static void sortCommand(redisClient *c) {
     int limit_start = 0, limit_count = -1, start, end;
     int j, dontsort = 0, vectorlen;
     int getop = 0; /* GET operation counter */
-    robj *sortval, *sortby = NULL;
+    robj *sortval, *sortby = NULL, *storekey = NULL;
     redisSortObject *vector; /* Resulting vector to sort */
 
     /* Lookup the key to sort. It must be of the right types */
@@ -3986,6 +4486,9 @@ static void sortCommand(redisClient *c) {
             limit_start = atoi(c->argv[j+1]->ptr);
             limit_count = atoi(c->argv[j+2]->ptr);
             j+=2;
+        } else if (!strcasecmp(c->argv[j]->ptr,"store") && leftargs >= 1) {
+            storekey = c->argv[j+1];
+            j++;
         } else if (!strcasecmp(c->argv[j]->ptr,"by") && leftargs >= 1) {
             sortby = c->argv[j+1];
             /* If the BY pattern does not contain '*', i.e. it is constant,
@@ -3997,18 +4500,6 @@ static void sortCommand(redisClient *c) {
                 REDIS_SORT_GET,c->argv[j+1]));
             getop++;
             j++;
-        } else if (!strcasecmp(c->argv[j]->ptr,"del") && leftargs >= 1) {
-            listAddNodeTail(operations,createSortOperation(
-                REDIS_SORT_DEL,c->argv[j+1]));
-            j++;
-        } else if (!strcasecmp(c->argv[j]->ptr,"incr") && leftargs >= 1) {
-            listAddNodeTail(operations,createSortOperation(
-                REDIS_SORT_INCR,c->argv[j+1]));
-            j++;
-        } else if (!strcasecmp(c->argv[j]->ptr,"get") && leftargs >= 1) {
-            listAddNodeTail(operations,createSortOperation(
-                REDIS_SORT_DECR,c->argv[j+1]));
-            j++;
         } else {
             decrRefCount(sortval);
             listRelease(operations);
@@ -4115,32 +4606,70 @@ static void sortCommand(redisClient *c) {
     /* Send command output to the output buffer, performing the specified
      * GET/DEL/INCR/DECR operations if any. */
     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;
-        if (!getop) {
-            addReplyBulkLen(c,vector[j].obj);
-            addReply(c,vector[j].obj);
-            addReply(c,shared.crlf);
+    if (storekey == NULL) {
+        /* STORE option not specified, sent the sorting result to client */
+        addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",outputlen));
+        for (j = start; j <= end; j++) {
+            listNode *ln;
+            if (!getop) {
+                addReplyBulkLen(c,vector[j].obj);
+                addReply(c,vector[j].obj);
+                addReply(c,shared.crlf);
+            }
+            listRewind(operations);
+            while((ln = listYield(operations))) {
+                redisSortOperation *sop = ln->value;
+                robj *val = lookupKeyByPattern(c->db,sop->pattern,
+                    vector[j].obj);
+
+                if (sop->type == REDIS_SORT_GET) {
+                    if (!val || val->type != REDIS_STRING) {
+                        addReply(c,shared.nullbulk);
+                    } else {
+                        addReplyBulkLen(c,val);
+                        addReply(c,val);
+                        addReply(c,shared.crlf);
+                    }
+                } else {
+                    assert(sop->type == REDIS_SORT_GET); /* always fails */
+                }
+            }
         }
-        listRewind(operations);
-        while((ln = listYield(operations))) {
-            redisSortOperation *sop = ln->value;
-            robj *val = lookupKeyByPattern(c->db,sop->pattern,
-                vector[j].obj);
+    } else {
+        robj *listObject = createListObject();
+        list *listPtr = (list*) listObject->ptr;
 
-            if (sop->type == REDIS_SORT_GET) {
-                if (!val || val->type != REDIS_STRING) {
-                    addReply(c,shared.nullbulk);
+        /* STORE option specified, set the sorting result as a List object */
+        for (j = start; j <= end; j++) {
+            listNode *ln;
+            if (!getop) {
+                listAddNodeTail(listPtr,vector[j].obj);
+                incrRefCount(vector[j].obj);
+            }
+            listRewind(operations);
+            while((ln = listYield(operations))) {
+                redisSortOperation *sop = ln->value;
+                robj *val = lookupKeyByPattern(c->db,sop->pattern,
+                    vector[j].obj);
+
+                if (sop->type == REDIS_SORT_GET) {
+                    if (!val || val->type != REDIS_STRING) {
+                        listAddNodeTail(listPtr,createStringObject("",0));
+                    } else {
+                        listAddNodeTail(listPtr,val);
+                        incrRefCount(val);
+                    }
                 } else {
-                    addReplyBulkLen(c,val);
-                    addReply(c,val);
-                    addReply(c,shared.crlf);
+                    assert(sop->type == REDIS_SORT_GET); /* always fails */
                 }
-            } else if (sop->type == REDIS_SORT_DEL) {
-                /* TODO */
             }
         }
+        dictReplace(c->db->dict,storekey,listObject);
+        /* Note: we add 1 because the DB is dirty anyway since even if the
+         * SORT result is empty a new key is set and maybe the old content
+         * replaced. */
+        server.dirty += 1+outputlen;
+        addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",outputlen));
     }
 
     /* Cleanup */
@@ -4196,7 +4725,7 @@ static void infoCommand(redisClient *c) {
             server.masterport,
             (server.replstate == REDIS_REPL_CONNECTED) ?
                 "up" : "down",
-            (int)(time(NULL)-server.master->lastinteraction)
+            server.master ? ((int)(time(NULL)-server.master->lastinteraction)) : -1
         );
     }
     for (j = 0; j < server.dbnum; j++) {
@@ -4284,21 +4813,21 @@ static int deleteIfVolatile(redisDb *db, robj *key) {
     return dictDelete(db->dict,key) == DICT_OK;
 }
 
-static void expireCommand(redisClient *c) {
+static void expireGenericCommand(redisClient *c, robj *key, time_t seconds) {
     dictEntry *de;
-    int seconds = atoi(c->argv[2]->ptr);
 
-    de = dictFind(c->db->dict,c->argv[1]);
+    de = dictFind(c->db->dict,key);
     if (de == NULL) {
         addReply(c,shared.czero);
         return;
     }
-    if (seconds <= 0) {
-        addReply(c, shared.czero);
+    if (seconds < 0) {
+        if (deleteKey(c->db,key)) server.dirty++;
+        addReply(c, shared.cone);
         return;
     } else {
         time_t when = time(NULL)+seconds;
-        if (setExpire(c->db,c->argv[1],when)) {
+        if (setExpire(c->db,key,when)) {
             addReply(c,shared.cone);
             server.dirty++;
         } else {
@@ -4308,6 +4837,14 @@ static void expireCommand(redisClient *c) {
     }
 }
 
+static void expireCommand(redisClient *c) {
+    expireGenericCommand(c,c->argv[1],strtol(c->argv[2]->ptr,NULL,10));
+}
+
+static void expireatCommand(redisClient *c) {
+    expireGenericCommand(c,c->argv[1],strtol(c->argv[2]->ptr,NULL,10)-time(NULL));
+}
+
 static void ttlCommand(redisClient *c) {
     time_t expire;
     int ttl = -1;
@@ -4595,7 +5132,7 @@ static void updateSlavesWaitingBgsave(int bgsaveerr) {
 }
 
 static int syncWithMaster(void) {
-    char buf[1024], tmpfile[256];
+    char buf[1024], tmpfile[256], authcmd[1024];
     int dumpsize;
     int fd = anetTcpConnect(NULL,server.masterhost,server.masterport);
     int dfd;
@@ -4605,6 +5142,30 @@ static int syncWithMaster(void) {
             strerror(errno));
         return REDIS_ERR;
     }
+
+    /* AUTH with the master if required. */
+    if(server.masterauth) {
+    	snprintf(authcmd, 1024, "AUTH %s\r\n", server.masterauth);
+    	if (syncWrite(fd, authcmd, strlen(server.masterauth)+7, 5) == -1) {
+            close(fd);
+            redisLog(REDIS_WARNING,"Unable to AUTH to MASTER: %s",
+                strerror(errno));
+            return REDIS_ERR;
+    	}
+        /* Read the AUTH result.  */
+        if (syncReadLine(fd,buf,1024,3600) == -1) {
+            close(fd);
+            redisLog(REDIS_WARNING,"I/O error reading auth result from MASTER: %s",
+                strerror(errno));
+            return REDIS_ERR;
+        }
+        if (buf[0] != '+') {
+            close(fd);
+            redisLog(REDIS_WARNING,"Cannot AUTH to MASTER, is the masterauth password correct?");
+            return REDIS_ERR;
+        }
+    }
+
     /* Issue the SYNC command */
     if (syncWrite(fd,"SYNC \r\n",7,5) == -1) {
         close(fd);
@@ -4619,6 +5180,11 @@ static int syncWithMaster(void) {
             strerror(errno));
         return REDIS_ERR;
     }
+    if (buf[0] != '$') {
+        close(fd);
+        redisLog(REDIS_WARNING,"Bad protocol from MASTER, the first byte is not '$', are you sure the host and port are right?");
+        return REDIS_ERR;
+    }
     dumpsize = atoi(buf+1);
     redisLog(REDIS_NOTICE,"Receiving %d bytes data dump from MASTER",dumpsize);
     /* Read the bulk write data on a temp file */
@@ -4742,6 +5308,198 @@ static void freeMemoryIfNeeded(void) {
     }
 }
 
+/* ============================== Append Only file ========================== */
+
+static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc) {
+    sds buf = sdsempty();
+    int j;
+    ssize_t nwritten;
+    time_t now;
+    robj *tmpargv[3];
+
+    /* The DB this command was targetting is not the same as the last command
+     * we appendend. To issue a SELECT command is needed. */
+    if (dictid != server.appendseldb) {
+        char seldb[64];
+
+        snprintf(seldb,sizeof(seldb),"%d",dictid);
+        buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n",
+            strlen(seldb),seldb);
+        server.appendseldb = dictid;
+    }
+
+    /* "Fix" the argv vector if the command is EXPIRE. We want to translate
+     * EXPIREs into EXPIREATs calls */
+    if (cmd->proc == expireCommand) {
+        long when;
+
+        tmpargv[0] = createStringObject("EXPIREAT",8);
+        tmpargv[1] = argv[1];
+        incrRefCount(argv[1]);
+        when = time(NULL)+strtol(argv[2]->ptr,NULL,10);
+        tmpargv[2] = createObject(REDIS_STRING,
+            sdscatprintf(sdsempty(),"%ld",when));
+        argv = tmpargv;
+    }
+
+    /* Append the actual command */
+    buf = sdscatprintf(buf,"*%d\r\n",argc);
+    for (j = 0; j < argc; j++) {
+        robj *o = argv[j];
+
+        if (o->encoding != REDIS_ENCODING_RAW)
+            o = getDecodedObject(o);
+        buf = sdscatprintf(buf,"$%d\r\n",sdslen(o->ptr));
+        buf = sdscatlen(buf,o->ptr,sdslen(o->ptr));
+        buf = sdscatlen(buf,"\r\n",2);
+        if (o != argv[j])
+            decrRefCount(o);
+    }
+
+    /* Free the objects from the modified argv for EXPIREAT */
+    if (cmd->proc == expireCommand) {
+        for (j = 0; j < 3; j++)
+            decrRefCount(argv[j]);
+    }
+
+    /* We want to perform a single write. This should be guaranteed atomic
+     * at least if the filesystem we are writing is a real physical one.
+     * While this will save us against the server being killed I don't think
+     * there is much to do about the whole server stopping for power problems
+     * or alike */
+     nwritten = write(server.appendfd,buf,sdslen(buf));
+     if (nwritten != (signed)sdslen(buf)) {
+        /* Ooops, we are in troubles. The best thing to do for now is
+         * to simply exit instead to give the illusion that everything is
+         * working as expected. */
+         if (nwritten == -1) {
+            redisLog(REDIS_WARNING,"Exiting on error writing to the append-only file: %s",strerror(errno));
+         } else {
+            redisLog(REDIS_WARNING,"Exiting on short write while writing to the append-only file: %s",strerror(errno));
+         }
+         exit(1);
+    }
+    now = time(NULL);
+    if (server.appendfsync == APPENDFSYNC_ALWAYS ||
+        (server.appendfsync == APPENDFSYNC_EVERYSEC &&
+         now-server.lastfsync > 1))
+    {
+        fsync(server.appendfd); /* Let's try to get this data on the disk */
+        server.lastfsync = now;
+    }
+}
+
+/* In Redis commands are always executed in the context of a client, so in
+ * order to load the append only file we need to create a fake client. */
+static struct redisClient *createFakeClient(void) {
+    struct redisClient *c = zmalloc(sizeof(*c));
+
+    selectDb(c,0);
+    c->fd = -1;
+    c->querybuf = sdsempty();
+    c->argc = 0;
+    c->argv = NULL;
+    c->flags = 0;
+    /* We set the fake client as a slave waiting for the synchronization
+     * so that Redis will not try to send replies to this client. */
+    c->replstate = REDIS_REPL_WAIT_BGSAVE_START;
+    c->reply = listCreate();
+    listSetFreeMethod(c->reply,decrRefCount);
+    listSetDupMethod(c->reply,dupClientReplyValue);
+    return c;
+}
+
+static void freeFakeClient(struct redisClient *c) {
+    sdsfree(c->querybuf);
+    listRelease(c->reply);
+    zfree(c);
+}
+
+/* Replay the append log file. On error REDIS_OK is returned. On non fatal
+ * error (the append only file is zero-length) REDIS_ERR is returned. On
+ * fatal error an error message is logged and the program exists. */
+int loadAppendOnlyFile(char *filename) {
+    struct redisClient *fakeClient;
+    FILE *fp = fopen(filename,"r");
+    struct redis_stat sb;
+
+    if (redis_fstat(fileno(fp),&sb) != -1 && sb.st_size == 0)
+        return REDIS_ERR;
+
+    if (fp == NULL) {
+        redisLog(REDIS_WARNING,"Fatal error: can't open the append log file for reading: %s",strerror(errno));
+        exit(1);
+    }
+
+    fakeClient = createFakeClient();
+    while(1) {
+        int argc, j;
+        unsigned long len;
+        robj **argv;
+        char buf[128];
+        sds argsds;
+        struct redisCommand *cmd;
+
+        if (fgets(buf,sizeof(buf),fp) == NULL) {
+            if (feof(fp))
+                break;
+            else
+                goto readerr;
+        }
+        if (buf[0] != '*') goto fmterr;
+        argc = atoi(buf+1);
+        argv = zmalloc(sizeof(robj*)*argc);
+        for (j = 0; j < argc; j++) {
+            if (fgets(buf,sizeof(buf),fp) == NULL) goto readerr;
+            if (buf[0] != '$') goto fmterr;
+            len = strtol(buf+1,NULL,10);
+            argsds = sdsnewlen(NULL,len);
+            if (fread(argsds,len,1,fp) == 0) goto fmterr;
+            argv[j] = createObject(REDIS_STRING,argsds);
+            if (fread(buf,2,1,fp) == 0) goto fmterr; /* discard CRLF */
+        }
+
+        /* Command lookup */
+        cmd = lookupCommand(argv[0]->ptr);
+        if (!cmd) {
+            redisLog(REDIS_WARNING,"Unknown command '%s' reading the append only file", argv[0]->ptr);
+            exit(1);
+        }
+        /* Try object sharing and encoding */
+        if (server.shareobjects) {
+            int j;
+            for(j = 1; j < argc; j++)
+                argv[j] = tryObjectSharing(argv[j]);
+        }
+        if (cmd->flags & REDIS_CMD_BULK)
+            tryObjectEncoding(argv[argc-1]);
+        /* Run the command in the context of a fake client */
+        fakeClient->argc = argc;
+        fakeClient->argv = argv;
+        cmd->proc(fakeClient);
+        /* Discard the reply objects list from the fake client */
+        while(listLength(fakeClient->reply))
+            listDelNode(fakeClient->reply,listFirst(fakeClient->reply));
+        /* Clean up, ready for the next command */
+        for (j = 0; j < argc; j++) decrRefCount(argv[j]);
+        zfree(argv);
+    }
+    fclose(fp);
+    freeFakeClient(fakeClient);
+    return REDIS_OK;
+
+readerr:
+    if (feof(fp)) {
+        redisLog(REDIS_WARNING,"Unexpected end of file reading the append only file");
+    } else {
+        redisLog(REDIS_WARNING,"Unrecoverable error reading the append only file: %s", strerror(errno));
+    }
+    exit(1);
+fmterr:
+    redisLog(REDIS_WARNING,"Bad file format reading the append only file");
+    exit(1);
+}
+
 /* ================================= Debugging ============================== */
 
 static void debugCommand(redisClient *c) {
@@ -4855,6 +5613,7 @@ static struct redisFunctionSym symsTable[] = {
 {"mgetCommand", (unsigned long)mgetCommand},
 {"monitorCommand", (unsigned long)monitorCommand},
 {"expireCommand", (unsigned long)expireCommand},
+{"expireatCommand", (unsigned long)expireatCommand},
 {"getsetCommand", (unsigned long)getsetCommand},
 {"ttlCommand", (unsigned long)ttlCommand},
 {"slaveofCommand", (unsigned long)slaveofCommand},
@@ -4875,6 +5634,13 @@ static struct redisFunctionSym symsTable[] = {
 {"zslDelete",(unsigned long)zslDelete},
 {"createZsetObject",(unsigned long)createZsetObject},
 {"zaddCommand",(unsigned long)zaddCommand},
+{"zrangeGenericCommand",(unsigned long)zrangeGenericCommand},
+{"zrangeCommand",(unsigned long)zrangeCommand},
+{"zrevrangeCommand",(unsigned long)zrevrangeCommand},
+{"zremCommand",(unsigned long)zremCommand},
+{"rdbSaveDoubleValue",(unsigned long)rdbSaveDoubleValue},
+{"rdbLoadDoubleValue",(unsigned long)rdbLoadDoubleValue},
+{"feedAppendOnlyFile",(unsigned long)feedAppendOnlyFile},
 {NULL,0}
 };
 
@@ -5066,8 +5832,13 @@ int main(int argc, char **argv) {
 #ifdef __linux__
     linuxOvercommitMemoryWarning();
 #endif
-    if (rdbLoad(server.dbfilename) == REDIS_OK)
-        redisLog(REDIS_NOTICE,"DB loaded from disk");
+    if (server.appendonly) {
+        if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK)
+            redisLog(REDIS_NOTICE,"DB loaded from append only file");
+    } else {
+        if (rdbLoad(server.dbfilename) == REDIS_OK)
+            redisLog(REDIS_NOTICE,"DB loaded from disk");
+    }
     if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
         acceptHandler, NULL, NULL) == AE_ERR) oom("creating file event");
     redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);