Further integrate MDB key archival into config.

[redis.git] / src / redis.c

diff --git a/src/redis.c b/src/redis.c
index aa5a73f2abae195632463461416acf60c5c3f887..eed212635dc3882626fe76043f64f54c92fdd02e 100644 (file)
--- a/src/redis.c
+++ b/src/redis.c
@@ -1,5 +1,5 @@
 /*
 /*

- * Copyright (c) 2009-2010, Salvatore Sanfilippo <antirez at gmail dot com>
+ * Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>

  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * All rights reserved.
  *
  * Redistribution and use in source and binary forms, with or without


@@ -49,6 +49,7 @@
 #include <math.h>
 #include <sys/resource.h>
 #include <sys/utsname.h>
 #include <math.h>
 #include <sys/resource.h>
 #include <sys/utsname.h>

+#include <lmdb.h>

 
 /* Our shared "common" objects */
 
 
 /* Our shared "common" objects */
 


@@ -110,6 +111,7 @@ struct redisCommand *commandTable;
  * t: Allow command while a slave has stale data but is not allowed to
  *    server this data. Normally no command is accepted in this condition
  *    but just a few.
  * t: Allow command while a slave has stale data but is not allowed to
  *    server this data. Normally no command is accepted in this condition
  *    but just a few.

+ * M: Do not automatically propagate the command on MONITOR.

  */
 struct redisCommand redisCommandTable[] = {
     {"get",getCommand,2,"r",0,NULL,1,1,1,0,0},
  */
 struct redisCommand redisCommandTable[] = {
     {"get",getCommand,2,"r",0,NULL,1,1,1,0,0},


@@ -216,7 +218,7 @@ struct redisCommand redisCommandTable[] = {
     {"lastsave",lastsaveCommand,1,"r",0,NULL,0,0,0,0,0},
     {"type",typeCommand,2,"r",0,NULL,1,1,1,0,0},
     {"multi",multiCommand,1,"rs",0,NULL,0,0,0,0,0},
     {"lastsave",lastsaveCommand,1,"r",0,NULL,0,0,0,0,0},
     {"type",typeCommand,2,"r",0,NULL,1,1,1,0,0},
     {"multi",multiCommand,1,"rs",0,NULL,0,0,0,0,0},

-    {"exec",execCommand,1,"s",0,NULL,0,0,0,0,0},
+    {"exec",execCommand,1,"sM",0,NULL,0,0,0,0,0},

     {"discard",discardCommand,1,"rs",0,NULL,0,0,0,0,0},
     {"sync",syncCommand,1,"ars",0,NULL,0,0,0,0,0},
     {"replconf",replconfCommand,-1,"ars",0,NULL,0,0,0,0,0},
     {"discard",discardCommand,1,"rs",0,NULL,0,0,0,0,0},
     {"sync",syncCommand,1,"ars",0,NULL,0,0,0,0,0},
     {"replconf",replconfCommand,-1,"ars",0,NULL,0,0,0,0,0},


@@ -390,7 +392,8 @@ int dictSdsKeyCompare(void *privdata, const void *key1,
     return memcmp(key1, key2, l1) == 0;
 }
 
     return memcmp(key1, key2, l1) == 0;
 }
 

-/* A case insensitive version used for the command lookup table. */
+/* A case insensitive version used for the command lookup table and other
+ * places where case insensitive non binary-safe comparison is needed. */

 int dictSdsKeyCaseCompare(void *privdata, const void *key1,
         const void *key2)
 {
 int dictSdsKeyCaseCompare(void *privdata, const void *key1,
         const void *key2)
 {


@@ -505,6 +508,16 @@ dictType dbDictType = {
     dictRedisObjectDestructor   /* val destructor */
 };
 
     dictRedisObjectDestructor   /* val destructor */
 };
 

+/* server.lua_scripts sha (as sds string) -> scripts (as robj) cache. */
+dictType shaScriptObjectDictType = {
+    dictSdsCaseHash,            /* hash function */
+    NULL,                       /* key dup */
+    NULL,                       /* val dup */
+    dictSdsKeyCaseCompare,      /* key compare */
+    dictSdsDestructor,          /* key destructor */
+    dictRedisObjectDestructor   /* val destructor */
+};
+

 /* Db->expires */
 dictType keyptrDictType = {
     dictSdsHash,               /* hash function */
 /* Db->expires */
 dictType keyptrDictType = {
     dictSdsHash,               /* hash function */


@@ -901,8 +914,12 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 
             if (pid == server.rdb_child_pid) {
                 backgroundSaveDoneHandler(exitcode,bysignal);
 
             if (pid == server.rdb_child_pid) {
                 backgroundSaveDoneHandler(exitcode,bysignal);

-            } else {
+            } else if (pid == server.aof_child_pid) {

                 backgroundRewriteDoneHandler(exitcode,bysignal);
                 backgroundRewriteDoneHandler(exitcode,bysignal);

+            } else {
+                redisLog(REDIS_WARNING,
+                    "Warning, detected child with unmatched pid: %ld",
+                    (long)pid);

             }
             updateDictResizePolicy();
         }
             }
             updateDictResizePolicy();
         }


@@ -1032,6 +1049,8 @@ void createSharedObjects(void) {
         "-READONLY You can't write against a read only slave.\r\n"));
     shared.oomerr = createObject(REDIS_STRING,sdsnew(
         "-OOM command not allowed when used memory > 'maxmemory'.\r\n"));
         "-READONLY You can't write against a read only slave.\r\n"));
     shared.oomerr = createObject(REDIS_STRING,sdsnew(
         "-OOM command not allowed when used memory > 'maxmemory'.\r\n"));

+    shared.execaborterr = createObject(REDIS_STRING,sdsnew(
+        "-EXECABORT Transaction discarded because of previous errors.\r\n"));

     shared.space = createObject(REDIS_STRING,sdsnew(" "));
     shared.colon = createObject(REDIS_STRING,sdsnew(":"));
     shared.plus = createObject(REDIS_STRING,sdsnew("+"));
     shared.space = createObject(REDIS_STRING,sdsnew(" "));
     shared.colon = createObject(REDIS_STRING,sdsnew(":"));
     shared.plus = createObject(REDIS_STRING,sdsnew("+"));


@@ -1101,6 +1120,9 @@ void initServerConfig() {
     server.pidfile = zstrdup("/var/run/redis.pid");
     server.rdb_filename = zstrdup("dump.rdb");
     server.aof_filename = zstrdup("appendonly.aof");
     server.pidfile = zstrdup("/var/run/redis.pid");
     server.rdb_filename = zstrdup("dump.rdb");
     server.aof_filename = zstrdup("appendonly.aof");

+    server.mdb_state = REDIS_MDB_OFF;
+    server.mdb_environment = zstrdup("archive");
+    server.mdb_mapsize = 10485760;

     server.requirepass = NULL;
     server.rdb_compression = 1;
     server.rdb_checksum = 1;
     server.requirepass = NULL;
     server.rdb_compression = 1;
     server.rdb_checksum = 1;


@@ -1326,13 +1348,22 @@ void initServer() {
         }
     }
 
         }
     }
 

+    if (server.aof_state == REDIS_MDB_ON) {
+        int retval = startKeyArchive();
+        if (retval != 0) {
+            redisLog(REDIS_WARNING, "Can't open the key-archive environment: %s",
+                mdb_strerror(retval));
+            exit(1);
+        }
+    }
+

     /* 32 bit instances are limited to 4GB of address space, so if there is
      * no explicit limit in the user provided configuration we set a limit
     /* 32 bit instances are limited to 4GB of address space, so if there is
      * no explicit limit in the user provided configuration we set a limit

-     * at 3.5GB using maxmemory with 'noeviction' policy'. This saves
-     * useless crashes of the Redis instance. */
+     * at 3 GB using maxmemory with 'noeviction' policy'. This avoids
+     * useless crashes of the Redis instance for out of memory. */

     if (server.arch_bits == 32 && server.maxmemory == 0) {
     if (server.arch_bits == 32 && server.maxmemory == 0) {

-        redisLog(REDIS_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3.5 GB maxmemory limit with 'noeviction' policy now.");
-        server.maxmemory = 3584LL*(1024*1024); /* 3584 MB = 3.5 GB */
+        redisLog(REDIS_WARNING,"Warning: 32 bit instance detected but no memory limit set. Setting 3 GB maxmemory limit with 'noeviction' policy now.");
+        server.maxmemory = 3072LL*(1024*1024); /* 3 GB */

         server.maxmemory_policy = REDIS_MAXMEMORY_NO_EVICTION;
     }
 
         server.maxmemory_policy = REDIS_MAXMEMORY_NO_EVICTION;
     }
 


@@ -1365,6 +1396,7 @@ void populateCommandTable(void) {
             case 'S': c->flags |= REDIS_CMD_SORT_FOR_SCRIPT; break;
             case 'l': c->flags |= REDIS_CMD_LOADING; break;
             case 't': c->flags |= REDIS_CMD_STALE; break;
             case 'S': c->flags |= REDIS_CMD_SORT_FOR_SCRIPT; break;
             case 'l': c->flags |= REDIS_CMD_LOADING; break;
             case 't': c->flags |= REDIS_CMD_STALE; break;

+            case 'M': c->flags |= REDIS_CMD_SKIP_MONITOR; break;

             default: redisPanic("Unsupported command flag"); break;
             }
             f++;
             default: redisPanic("Unsupported command flag"); break;
             }
             f++;


@@ -1440,7 +1472,7 @@ struct redisCommand *lookupCommandByCString(char *s) {
 }
 
 /* Propagate the specified command (in the context of the specified database id)
 }
 
 /* Propagate the specified command (in the context of the specified database id)

- * to AOF, Slaves and Monitors.
+ * to AOF and Slaves.

  *
  * flags are an xor between:
  * + REDIS_PROPAGATE_NONE (no propagation of command at all)
  *
  * flags are an xor between:
  * + REDIS_PROPAGATE_NONE (no propagation of command at all)


@@ -1470,8 +1502,12 @@ void call(redisClient *c, int flags) {
 
     /* Sent the command to clients in MONITOR mode, only if the commands are
      * not geneated from reading an AOF. */
 
     /* Sent the command to clients in MONITOR mode, only if the commands are
      * not geneated from reading an AOF. */

-    if (listLength(server.monitors) && !server.loading)
+    if (listLength(server.monitors) &&
+        !server.loading &&
+        !(c->cmd->flags & REDIS_CMD_SKIP_MONITOR))
+    {

         replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);
         replicationFeedMonitors(c,server.monitors,c->db->id,c->argv,c->argc);

+    }

 
     /* Call the command. */
     redisOpArrayInit(&server.also_propagate);
 
     /* Call the command. */
     redisOpArrayInit(&server.also_propagate);


@@ -1521,7 +1557,7 @@ void call(redisClient *c, int flags) {
 }
 
 /* If this function gets called we already read a whole
 }
 
 /* If this function gets called we already read a whole

- * command, argments are in the client argv/argc fields.
+ * command, arguments are in the client argv/argc fields.

  * processCommand() execute the command or prepare the
  * server for a bulk read from the client.
  *
  * processCommand() execute the command or prepare the
  * server for a bulk read from the client.
  *


@@ -1543,11 +1579,13 @@ int processCommand(redisClient *c) {
      * such as wrong arity, bad command name and so forth. */
     c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
     if (!c->cmd) {
      * such as wrong arity, bad command name and so forth. */
     c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
     if (!c->cmd) {

+        flagTransaction(c);

         addReplyErrorFormat(c,"unknown command '%s'",
             (char*)c->argv[0]->ptr);
         return REDIS_OK;
     } else if ((c->cmd->arity > 0 && c->cmd->arity != c->argc) ||
                (c->argc < -c->cmd->arity)) {
         addReplyErrorFormat(c,"unknown command '%s'",
             (char*)c->argv[0]->ptr);
         return REDIS_OK;
     } else if ((c->cmd->arity > 0 && c->cmd->arity != c->argc) ||
                (c->argc < -c->cmd->arity)) {

+        flagTransaction(c);

         addReplyErrorFormat(c,"wrong number of arguments for '%s' command",
             c->cmd->name);
         return REDIS_OK;
         addReplyErrorFormat(c,"wrong number of arguments for '%s' command",
             c->cmd->name);
         return REDIS_OK;


@@ -1556,6 +1594,7 @@ int processCommand(redisClient *c) {
     /* Check if the user is authenticated */
     if (server.requirepass && !c->authenticated && c->cmd->proc != authCommand)
     {
     /* Check if the user is authenticated */
     if (server.requirepass && !c->authenticated && c->cmd->proc != authCommand)
     {

+        flagTransaction(c);

         addReplyError(c,"operation not permitted");
         return REDIS_OK;
     }
         addReplyError(c,"operation not permitted");
         return REDIS_OK;
     }


@@ -1568,6 +1607,7 @@ int processCommand(redisClient *c) {
     if (server.maxmemory) {
         int retval = freeMemoryIfNeeded();
         if ((c->cmd->flags & REDIS_CMD_DENYOOM) && retval == REDIS_ERR) {
     if (server.maxmemory) {
         int retval = freeMemoryIfNeeded();
         if ((c->cmd->flags & REDIS_CMD_DENYOOM) && retval == REDIS_ERR) {

+            flagTransaction(c);

             addReply(c, shared.oomerr);
             return REDIS_OK;
         }
             addReply(c, shared.oomerr);
             return REDIS_OK;
         }


@@ -1579,6 +1619,7 @@ int processCommand(redisClient *c) {
         && server.lastbgsave_status == REDIS_ERR &&
         c->cmd->flags & REDIS_CMD_WRITE)
     {
         && server.lastbgsave_status == REDIS_ERR &&
         c->cmd->flags & REDIS_CMD_WRITE)
     {

+        flagTransaction(c);

         addReply(c, shared.bgsaveerr);
         return REDIS_OK;
     }
         addReply(c, shared.bgsaveerr);
         return REDIS_OK;
     }


@@ -1610,6 +1651,7 @@ int processCommand(redisClient *c) {
         server.repl_serve_stale_data == 0 &&
         !(c->cmd->flags & REDIS_CMD_STALE))
     {
         server.repl_serve_stale_data == 0 &&
         !(c->cmd->flags & REDIS_CMD_STALE))
     {

+        flagTransaction(c);

         addReply(c, shared.masterdownerr);
         return REDIS_OK;
     }
         addReply(c, shared.masterdownerr);
         return REDIS_OK;
     }


@@ -1623,6 +1665,7 @@ int processCommand(redisClient *c) {
 
     /* Lua script too slow? Only allow commands with REDIS_CMD_STALE flag. */
     if (server.lua_timedout &&
 
     /* Lua script too slow? Only allow commands with REDIS_CMD_STALE flag. */
     if (server.lua_timedout &&

+          c->cmd->proc != authCommand &&

         !(c->cmd->proc == shutdownCommand &&
           c->argc == 2 &&
           tolower(((char*)c->argv[1]->ptr)[0]) == 'n') &&
         !(c->cmd->proc == shutdownCommand &&
           c->argc == 2 &&
           tolower(((char*)c->argv[1]->ptr)[0]) == 'n') &&


@@ -1630,6 +1673,7 @@ int processCommand(redisClient *c) {
           c->argc == 2 &&
           tolower(((char*)c->argv[1]->ptr)[0]) == 'k'))
     {
           c->argc == 2 &&
           tolower(((char*)c->argv[1]->ptr)[0]) == 'k'))
     {

+        flagTransaction(c);

         addReply(c, shared.slowscripterr);
         return REDIS_OK;
     }
         addReply(c, shared.slowscripterr);
         return REDIS_OK;
     }


@@ -2063,7 +2107,10 @@ sds genRedisInfoString(char *section) {
                     (long)server.unixtime-server.repl_down_since);
             }
             info = sdscatprintf(info,
                     (long)server.unixtime-server.repl_down_since);
             }
             info = sdscatprintf(info,

-                "slave_priority:%d\r\n", server.slave_priority);
+                "slave_priority:%d\r\n"
+                "slave_read_only:%d\r\n",
+                server.slave_priority,
+                server.repl_slave_ro);

         }
         info = sdscatprintf(info,
             "connected_slaves:%lu\r\n",
         }
         info = sdscatprintf(info,
             "connected_slaves:%lu\r\n",


@@ -2165,7 +2212,7 @@ void infoCommand(redisClient *c) {
 }
 
 void monitorCommand(redisClient *c) {
 }
 
 void monitorCommand(redisClient *c) {

-    /* ignore MONITOR if aleady slave or in monitor mode */
+    /* ignore MONITOR if already slave or in monitor mode */

     if (c->flags & REDIS_SLAVE) return;
 
     c->flags |= (REDIS_SLAVE|REDIS_MONITOR);
     if (c->flags & REDIS_SLAVE) return;
 
     c->flags |= (REDIS_SLAVE|REDIS_MONITOR);


@@ -2264,7 +2311,7 @@ int freeMemoryIfNeeded(void) {
 
                     de = dictGetRandomKey(dict);
                     thiskey = dictGetKey(de);
 
                     de = dictGetRandomKey(dict);
                     thiskey = dictGetKey(de);

-                    /* When policy is volatile-lru we need an additonal lookup
+                    /* When policy is volatile-lru we need an additional lookup

                      * to locate the real key, as dict is set to db->expires. */
                     if (server.maxmemory_policy == REDIS_MAXMEMORY_VOLATILE_LRU)
                         de = dictFind(db->dict, thiskey);
                      * to locate the real key, as dict is set to db->expires. */
                     if (server.maxmemory_policy == REDIS_MAXMEMORY_VOLATILE_LRU)
                         de = dictFind(db->dict, thiskey);


@@ -2303,6 +2350,8 @@ int freeMemoryIfNeeded(void) {
                 long long delta;
 
                 robj *keyobj = createStringObject(bestkey,sdslen(bestkey));
                 long long delta;
 
                 robj *keyobj = createStringObject(bestkey,sdslen(bestkey));

+                int archived = archive(db, keyobj);
+                redisAssert(archived != 0);

                 propagateExpire(db,keyobj);
                 /* We compute the amount of memory freed by dbDelete() alone.
                  * It is possible that actually the memory needed to propagate
                 propagateExpire(db,keyobj);
                 /* We compute the amount of memory freed by dbDelete() alone.
                  * It is possible that actually the memory needed to propagate


@@ -2566,7 +2615,7 @@ int main(int argc, char **argv) {
     redisAsciiArt();
 
     if (!server.sentinel_mode) {
     redisAsciiArt();
 
     if (!server.sentinel_mode) {

-        /* Things only needed when not runnign in Sentinel mode. */
+        /* Things only needed when not running in Sentinel mode. */

         redisLog(REDIS_WARNING,"Server started, Redis version " REDIS_VERSION);
     #ifdef __linux__
         linuxOvercommitMemoryWarning();
         redisLog(REDIS_WARNING,"Server started, Redis version " REDIS_VERSION);
     #ifdef __linux__
         linuxOvercommitMemoryWarning();