source reshaped a bit to play well with a bgsaving thread, still work to do, does...

[redis.git] / src / redis.c

diff --git a/src/redis.c b/src/redis.c
index 8e82ee032ce8ecffe5f68809aff8dcfcbd7c3132..2fd3ee39e3c8843f1f5e56c991e195c11b8aaf10 100644 (file)
--- a/src/redis.c
+++ b/src/redis.c
@@ -345,7 +345,7 @@ unsigned int dictEncObjHash(const void *key) {
     }
 }
 
     }
 }
 

-/* Sets type */
+/* Sets type and diskstore negative caching hash table */

 dictType setDictType = {
     dictEncObjHash,            /* hash function */
     NULL,                      /* key dup */
 dictType setDictType = {
     dictEncObjHash,            /* hash function */
     NULL,                      /* key dup */


@@ -589,14 +589,32 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
         pid_t pid;
 
         if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
         pid_t pid;
 
         if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {

+            int exitcode = WEXITSTATUS(statloc);
+            int bysignal = 0;
+            
+            if (WIFSIGNALED(statloc)) bysignal = WTERMSIG(statloc);
+

             if (pid == server.bgsavechildpid) {
             if (pid == server.bgsavechildpid) {

-                backgroundSaveDoneHandler(statloc);
+                backgroundSaveDoneHandler(exitcode,bysignal);

             } else {
             } else {

-                backgroundRewriteDoneHandler(statloc);
+                backgroundRewriteDoneHandler(exitcode,bysignal);

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

-    } else {
+        if (server.bgsavethread != (pthread_t) -1) {
+            int state;
+
+            pthread_mutex_lock(&server.bgsavethread_mutex);
+            state = server.bgsavethread_state;
+            pthread_mutex_unlock(&server.bgsavethread_mutex);
+
+            if (state == REDIS_BGSAVE_DONE_OK || state == REDIS_BGSAVE_DONE_ERR)
+            {
+                backgroundSaveDoneHandler(
+                    (state == REDIS_BGSAVE_DONE_OK) ? 0 : 1, 0);
+            }
+        }
+    } else if (!server.ds_enabled) {

         /* If there is not a background saving in progress check if
          * we have to save now */
          time_t now = time(NULL);
         /* If there is not a background saving in progress check if
          * we have to save now */
          time_t now = time(NULL);


@@ -620,11 +638,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 
     /* Remove a few cached objects from memory if we are over the
      * configured memory limit */
 
     /* Remove a few cached objects from memory if we are over the
      * configured memory limit */

-    while (server.ds_enabled && zmalloc_used_memory() >
-            server.cache_max_memory)
-    {
-        cacheFreeOneEntry();
-    }
+    if (server.ds_enabled) cacheCron();

 
     /* Replication cron function -- used to reconnect to master and
      * to detect transfer failures. */
 
     /* Replication cron function -- used to reconnect to master and
      * to detect transfer failures. */


@@ -753,7 +767,6 @@ void initServerConfig() {
     server.syslog_enabled = 0;
     server.syslog_ident = zstrdup("redis");
     server.syslog_facility = LOG_LOCAL0;
     server.syslog_enabled = 0;
     server.syslog_ident = zstrdup("redis");
     server.syslog_facility = LOG_LOCAL0;

-    server.glueoutputbuf = 1;

     server.daemonize = 0;
     server.appendonly = 0;
     server.appendfsync = APPENDFSYNC_EVERYSEC;
     server.daemonize = 0;
     server.appendonly = 0;
     server.appendfsync = APPENDFSYNC_EVERYSEC;


@@ -773,7 +786,7 @@ void initServerConfig() {
     server.maxmemory_policy = REDIS_MAXMEMORY_VOLATILE_LRU;
     server.maxmemory_samples = 3;
     server.ds_enabled = 0;
     server.maxmemory_policy = REDIS_MAXMEMORY_VOLATILE_LRU;
     server.maxmemory_samples = 3;
     server.ds_enabled = 0;

-    server.ds_path = zstrdup("/tmp/redis.ds");
+    server.ds_path = sdsnew("/tmp/redis.ds");

     server.cache_max_memory = 64LL*1024*1024; /* 64 MB of RAM */
     server.cache_blocked_clients = 0;
     server.hash_max_zipmap_entries = REDIS_HASH_MAX_ZIPMAP_ENTRIES;
     server.cache_max_memory = 64LL*1024*1024; /* 64 MB of RAM */
     server.cache_blocked_clients = 0;
     server.hash_max_zipmap_entries = REDIS_HASH_MAX_ZIPMAP_ENTRIES;


@@ -782,6 +795,7 @@ void initServerConfig() {
     server.list_max_ziplist_value = REDIS_LIST_MAX_ZIPLIST_VALUE;
     server.set_max_intset_entries = REDIS_SET_MAX_INTSET_ENTRIES;
     server.shutdown_asap = 0;
     server.list_max_ziplist_value = REDIS_LIST_MAX_ZIPLIST_VALUE;
     server.set_max_intset_entries = REDIS_SET_MAX_INTSET_ENTRIES;
     server.shutdown_asap = 0;

+    server.cache_flush_delay = 0;

 
     updateLRUClock();
     resetServerSaveParams();
 
     updateLRUClock();
     resetServerSaveParams();


@@ -830,6 +844,8 @@ void initServer() {
     server.slaves = listCreate();
     server.monitors = listCreate();
     server.unblocked_clients = listCreate();
     server.slaves = listCreate();
     server.monitors = listCreate();
     server.unblocked_clients = listCreate();

+    server.cache_io_queue = listCreate();
+

     createSharedObjects();
     server.el = aeCreateEventLoop();
     server.db = zmalloc(sizeof(redisDb)*server.dbnum);
     createSharedObjects();
     server.el = aeCreateEventLoop();
     server.db = zmalloc(sizeof(redisDb)*server.dbnum);


@@ -855,8 +871,11 @@ void initServer() {
         server.db[j].expires = dictCreate(&keyptrDictType,NULL);
         server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
         server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
         server.db[j].expires = dictCreate(&keyptrDictType,NULL);
         server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
         server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);

-        if (server.ds_enabled)
+        if (server.ds_enabled) {

             server.db[j].io_keys = dictCreate(&keylistDictType,NULL);
             server.db[j].io_keys = dictCreate(&keylistDictType,NULL);

+            server.db[j].io_negcache = dictCreate(&setDictType,NULL);
+            server.db[j].io_queued = dictCreate(&setDictType,NULL);
+        }

         server.db[j].id = j;
     }
     server.pubsub_channels = dictCreate(&keylistDictType,NULL);
         server.db[j].id = j;
     }
     server.pubsub_channels = dictCreate(&keylistDictType,NULL);


@@ -866,6 +885,8 @@ void initServer() {
     server.cronloops = 0;
     server.bgsavechildpid = -1;
     server.bgrewritechildpid = -1;
     server.cronloops = 0;
     server.bgsavechildpid = -1;
     server.bgrewritechildpid = -1;

+    server.bgsavethread_state = REDIS_BGSAVE_THREAD_UNACTIVE;
+    server.bgsavethread = (pthread_t) -1;

     server.bgrewritebuf = sdsempty();
     server.aofbuf = sdsempty();
     server.lastsave = time(NULL);
     server.bgrewritebuf = sdsempty();
     server.aofbuf = sdsempty();
     server.lastsave = time(NULL);


@@ -1051,7 +1072,9 @@ int prepareForShutdown() {
         kill(server.bgsavechildpid,SIGKILL);
         rdbRemoveTempFile(server.bgsavechildpid);
     }
         kill(server.bgsavechildpid,SIGKILL);
         rdbRemoveTempFile(server.bgsavechildpid);
     }

-    if (server.appendonly) {
+    if (server.ds_enabled) {
+        /* FIXME: flush all objects on disk */
+    } else if (server.appendonly) {

         /* Append only file: fsync() the AOF and exit */
         aof_fsync(server.appendfd);
     } else if (server.saveparamslen > 0) {
         /* Append only file: fsync() the AOF and exit */
         aof_fsync(server.appendfd);
     } else if (server.saveparamslen > 0) {


@@ -1437,7 +1460,7 @@ void createPidFile(void) {
     /* Try to write the pid file in a best-effort way. */
     FILE *fp = fopen(server.pidfile,"w");
     if (fp) {
     /* Try to write the pid file in a best-effort way. */
     FILE *fp = fopen(server.pidfile,"w");
     if (fp) {

-        fprintf(fp,"%d\n",getpid());
+        fprintf(fp,"%d\n",(int)getpid());

         fclose(fp);
     }
 }
         fclose(fp);
     }
 }


@@ -1494,7 +1517,9 @@ int main(int argc, char **argv) {
     linuxOvercommitMemoryWarning();
 #endif
     start = time(NULL);
     linuxOvercommitMemoryWarning();
 #endif
     start = time(NULL);

-    if (server.appendonly) {
+    if (server.ds_enabled) {
+        redisLog(REDIS_NOTICE,"DB not loaded (running with disk back end)");
+    } else if (server.appendonly) {

         if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK)
             redisLog(REDIS_NOTICE,"DB loaded from append only file: %ld seconds",time(NULL)-start);
     } else {
         if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK)
             redisLog(REDIS_NOTICE,"DB loaded from append only file: %ld seconds",time(NULL)-start);
     } else {