]> git.saurik.com Git - redis.git/blobdiff - redis.c
append only file loading fixed
[redis.git] / redis.c
diff --git a/redis.c b/redis.c
index c730c04cdb108b48a15b223deff0195505fd726a..2d233311772b0f01be3b7429a68fb6ea98c30808 100644 (file)
--- a/redis.c
+++ b/redis.c
 #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 */
@@ -261,6 +266,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;
@@ -269,6 +279,7 @@ struct redisServer {
     char *logfile;
     char *bindaddr;
     char *dbfilename;
+    char *appendfilename;
     char *requirepass;
     int shareobjects;
     /* Replication related */
@@ -364,6 +375,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);
@@ -440,6 +452,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);
@@ -511,6 +524,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},
@@ -1021,8 +1035,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;
@@ -1082,6 +1102,15 @@ static void initServer() {
     server.stat_numconnections = 0;
     server.stat_starttime = time(NULL);
     aeCreateTimeEvent(server.el, 1000, 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 */
@@ -1221,6 +1250,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) {
@@ -1536,6 +1580,8 @@ static int processCommand(redisClient *c) {
     /* Exec the command */
     dirty = server.dirty;
     cmd->proc(c);
+    if (server.appendonly != 0)
+        feedAppendOnlyFile(cmd,c->db->id,c->argv,c->argc);
     if (server.dirty-dirty != 0 && listLength(server.slaves))
         replicationFeedSlaves(server.slaves,cmd,c->db->id,c->argv,c->argc);
     if (listLength(server.monitors))
@@ -2708,7 +2754,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 */
 }
@@ -2721,7 +2767,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"));
     }
 }
 
@@ -2979,6 +3025,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;
         }
     }
@@ -4736,21 +4783,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 {
@@ -4760,6 +4807,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;
@@ -5199,6 +5254,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) {
@@ -5312,6 +5559,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},
@@ -5338,6 +5586,7 @@ static struct redisFunctionSym symsTable[] = {
 {"zremCommand",(unsigned long)zremCommand},
 {"rdbSaveDoubleValue",(unsigned long)rdbSaveDoubleValue},
 {"rdbLoadDoubleValue",(unsigned long)rdbLoadDoubleValue},
+{"feedAppendOnlyFile",(unsigned long)feedAppendOnlyFile},
 {NULL,0}
 };
 
@@ -5529,8 +5778,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);