X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/e1a586ee69a15dd15afbba8e25f8eb62bbb22da9..36c17a53b6aece050b79b667fd32064f6eb116c2:/src/redis.c?ds=inline diff --git a/src/redis.c b/src/redis.c index 14923bc8..2fd3ee39 100644 --- a/src/redis.c +++ b/src/redis.c @@ -74,10 +74,14 @@ struct redisCommand readonlyCommandTable[] = { {"setnx",setnxCommand,3,REDIS_CMD_DENYOOM,NULL,0,0,0}, {"setex",setexCommand,4,REDIS_CMD_DENYOOM,NULL,0,0,0}, {"append",appendCommand,3,REDIS_CMD_DENYOOM,NULL,1,1,1}, - {"substr",substrCommand,4,0,NULL,1,1,1}, {"strlen",strlenCommand,2,0,NULL,1,1,1}, {"del",delCommand,-2,0,NULL,0,0,0}, {"exists",existsCommand,2,0,NULL,1,1,1}, + {"setbit",setbitCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1}, + {"getbit",getbitCommand,3,0,NULL,1,1,1}, + {"setrange",setrangeCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1}, + {"getrange",getrangeCommand,4,0,NULL,1,1,1}, + {"substr",getrangeCommand,4,0,NULL,1,1,1}, {"incr",incrCommand,2,REDIS_CMD_DENYOOM,NULL,1,1,1}, {"decr",decrCommand,2,REDIS_CMD_DENYOOM,NULL,1,1,1}, {"mget",mgetCommand,-2,0,NULL,1,-1,1}, @@ -89,6 +93,7 @@ struct redisCommand readonlyCommandTable[] = { {"rpop",rpopCommand,2,0,NULL,1,1,1}, {"lpop",lpopCommand,2,0,NULL,1,1,1}, {"brpop",brpopCommand,-3,0,NULL,1,1,1}, + {"brpoplpush",brpoplpushCommand,4,REDIS_CMD_DENYOOM,NULL,1,2,1}, {"blpop",blpopCommand,-3,0,NULL,1,1,1}, {"llen",llenCommand,2,0,NULL,1,1,1}, {"lindex",lindexCommand,3,0,NULL,1,1,1}, @@ -96,7 +101,7 @@ struct redisCommand readonlyCommandTable[] = { {"lrange",lrangeCommand,4,0,NULL,1,1,1}, {"ltrim",ltrimCommand,4,0,NULL,1,1,1}, {"lrem",lremCommand,4,0,NULL,1,1,1}, - {"rpoplpush",rpoplpushcommand,3,REDIS_CMD_DENYOOM,NULL,1,2,1}, + {"rpoplpush",rpoplpushCommand,3,REDIS_CMD_DENYOOM,NULL,1,2,1}, {"sadd",saddCommand,3,REDIS_CMD_DENYOOM,NULL,1,1,1}, {"srem",sremCommand,3,0,NULL,1,1,1}, {"smove",smoveCommand,4,0,NULL,1,2,1}, @@ -340,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 */ @@ -575,7 +580,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { } /* Close connections of timedout clients */ - if ((server.maxidletime && !(loops % 100)) || server.blpop_blocked_clients) + if ((server.maxidletime && !(loops % 100)) || server.bpop_blocked_clients) closeTimedoutClients(); /* Check if a background saving or AOF rewrite in progress terminated */ @@ -584,14 +589,32 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { 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) { - backgroundSaveDoneHandler(statloc); + backgroundSaveDoneHandler(exitcode,bysignal); } else { - backgroundRewriteDoneHandler(statloc); + backgroundRewriteDoneHandler(exitcode,bysignal); } 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); @@ -613,28 +636,9 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { * in order to guarantee a strict consistency. */ if (server.masterhost == NULL) activeExpireCycle(); - /* Swap a few keys on disk if we are over the memory limit and VM - * is enbled. Try to free objects from the free list first. */ - if (vmCanSwapOut()) { - while (server.vm_enabled && zmalloc_used_memory() > - server.vm_max_memory) - { - int retval = (server.vm_max_threads == 0) ? - vmSwapOneObjectBlocking() : - vmSwapOneObjectThreaded(); - if (retval == REDIS_ERR && !(loops % 300) && - zmalloc_used_memory() > - (server.vm_max_memory+server.vm_max_memory/10)) - { - redisLog(REDIS_WARNING,"WARNING: vm-max-memory limit exceeded by more than 10%% but unable to swap more objects out!"); - } - /* Note that when using threade I/O we free just one object, - * because anyway when the I/O thread in charge to swap this - * object out will finish, the handler of completed jobs - * will try to swap more objects if we are still out of memory. */ - if (retval == REDIS_ERR || server.vm_max_threads > 0) break; - } - } + /* Remove a few cached objects from memory if we are over the + * configured memory limit */ + if (server.ds_enabled) cacheCron(); /* Replication cron function -- used to reconnect to master and * to detect transfer failures. */ @@ -648,21 +652,22 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { * for ready file descriptors. */ void beforeSleep(struct aeEventLoop *eventLoop) { REDIS_NOTUSED(eventLoop); + listNode *ln; + redisClient *c; - /* Awake clients that got all the swapped keys they requested */ - if (server.vm_enabled && listLength(server.io_ready_clients)) { + /* Awake clients that got all the on disk keys they requested */ + if (server.ds_enabled && listLength(server.io_ready_clients)) { listIter li; - listNode *ln; listRewind(server.io_ready_clients,&li); while((ln = listNext(&li))) { - redisClient *c = ln->value; + c = ln->value; struct redisCommand *cmd; /* Resume the client. */ listDelNode(server.io_ready_clients,ln); c->flags &= (~REDIS_IO_WAIT); - server.vm_blocked_clients--; + server.cache_blocked_clients--; aeCreateFileEvent(server.el, c->fd, AE_READABLE, readQueryFromClient, c); cmd = lookupCommand(c->argv[0]->ptr); @@ -674,6 +679,19 @@ void beforeSleep(struct aeEventLoop *eventLoop) { processInputBuffer(c); } } + + /* Try to process pending commands for clients that were just unblocked. */ + while (listLength(server.unblocked_clients)) { + ln = listFirst(server.unblocked_clients); + redisAssert(ln != NULL); + c = ln->value; + listDelNode(server.unblocked_clients,ln); + + /* Process remaining data in the input buffer. */ + if (c->querybuf && sdslen(c->querybuf) > 0) + processInputBuffer(c); + } + /* Write the AOF buffer on disk */ flushAppendOnlyFile(); } @@ -749,7 +767,6 @@ void initServerConfig() { 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; @@ -764,23 +781,21 @@ void initServerConfig() { server.rdbcompression = 1; server.activerehashing = 1; server.maxclients = 0; - server.blpop_blocked_clients = 0; + server.bpop_blocked_clients = 0; server.maxmemory = 0; server.maxmemory_policy = REDIS_MAXMEMORY_VOLATILE_LRU; server.maxmemory_samples = 3; - server.vm_enabled = 0; - server.vm_swap_file = zstrdup("/tmp/redis-%p.vm"); - server.vm_page_size = 256; /* 256 bytes per page */ - server.vm_pages = 1024*1024*100; /* 104 millions of pages */ - server.vm_max_memory = 1024LL*1024*1024*1; /* 1 GB of RAM */ - server.vm_max_threads = 4; - server.vm_blocked_clients = 0; + server.ds_enabled = 0; + 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.hash_max_zipmap_value = REDIS_HASH_MAX_ZIPMAP_VALUE; server.list_max_ziplist_entries = REDIS_LIST_MAX_ZIPLIST_ENTRIES; 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(); @@ -828,6 +843,9 @@ void initServer() { server.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); @@ -853,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); - if (server.vm_enabled) + if (server.ds_enabled) { 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); @@ -864,6 +885,8 @@ void initServer() { 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); @@ -871,6 +894,7 @@ void initServer() { server.stat_numcommands = 0; server.stat_numconnections = 0; server.stat_expiredkeys = 0; + server.stat_evictedkeys = 0; server.stat_starttime = time(NULL); server.stat_keyspace_misses = 0; server.stat_keyspace_hits = 0; @@ -890,7 +914,7 @@ void initServer() { } } - if (server.vm_enabled) vmInit(); + if (server.ds_enabled) dsInit(); } /* Populates the Redis Command Table starting from the hard coded list @@ -1029,8 +1053,8 @@ int processCommand(redisClient *c) { queueMultiCommand(c,cmd); addReply(c,shared.queued); } else { - if (server.vm_enabled && server.vm_max_threads > 0 && - blockClientOnSwappedKeys(c,cmd)) return REDIS_ERR; + if (server.ds_enabled && blockClientOnSwappedKeys(c,cmd)) + return REDIS_ERR; call(c,cmd); } return REDIS_OK; @@ -1048,10 +1072,11 @@ int prepareForShutdown() { 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); - if (server.vm_enabled) unlink(server.vm_swap_file); } else if (server.saveparamslen > 0) { /* Snapshotting. Perform a SYNC SAVE and exit */ if (rdbSave(server.dbfilename) != REDIS_OK) { @@ -1157,13 +1182,14 @@ sds genRedisInfoString(void) { "total_connections_received:%lld\r\n" "total_commands_processed:%lld\r\n" "expired_keys:%lld\r\n" + "evicted_keys:%lld\r\n" "keyspace_hits:%lld\r\n" "keyspace_misses:%lld\r\n" "hash_max_zipmap_entries:%zu\r\n" "hash_max_zipmap_value:%zu\r\n" "pubsub_channels:%ld\r\n" "pubsub_patterns:%u\r\n" - "vm_enabled:%d\r\n" + "ds_enabled:%d\r\n" "role:%s\r\n" ,REDIS_VERSION, redisGitSHA1(), @@ -1180,7 +1206,7 @@ sds genRedisInfoString(void) { (float)c_ru.ru_stime.tv_sec+(float)c_ru.ru_stime.tv_usec/1000000, listLength(server.clients)-listLength(server.slaves), listLength(server.slaves), - server.blpop_blocked_clients, + server.bpop_blocked_clients, zmalloc_used_memory(), hmem, zmalloc_get_rss(), @@ -1199,13 +1225,14 @@ sds genRedisInfoString(void) { server.stat_numconnections, server.stat_numcommands, server.stat_expiredkeys, + server.stat_evictedkeys, server.stat_keyspace_hits, server.stat_keyspace_misses, server.hash_max_zipmap_entries, server.hash_max_zipmap_value, dictSize(server.pubsub_channels), listLength(server.pubsub_patterns), - server.vm_enabled != 0, + server.ds_enabled != 0, server.masterhost == NULL ? "master" : "slave" ); if (server.masterhost) { @@ -1232,33 +1259,13 @@ sds genRedisInfoString(void) { ); } } - if (server.vm_enabled) { + if (server.ds_enabled) { lockThreadedIO(); info = sdscatprintf(info, - "vm_conf_max_memory:%llu\r\n" - "vm_conf_page_size:%llu\r\n" - "vm_conf_pages:%llu\r\n" - "vm_stats_used_pages:%llu\r\n" - "vm_stats_swapped_objects:%llu\r\n" - "vm_stats_swappin_count:%llu\r\n" - "vm_stats_swappout_count:%llu\r\n" - "vm_stats_io_newjobs_len:%lu\r\n" - "vm_stats_io_processing_len:%lu\r\n" - "vm_stats_io_processed_len:%lu\r\n" - "vm_stats_io_active_threads:%lu\r\n" - "vm_stats_blocked_clients:%lu\r\n" - ,(unsigned long long) server.vm_max_memory, - (unsigned long long) server.vm_page_size, - (unsigned long long) server.vm_pages, - (unsigned long long) server.vm_stats_used_pages, - (unsigned long long) server.vm_stats_swapped_objects, - (unsigned long long) server.vm_stats_swapins, - (unsigned long long) server.vm_stats_swapouts, - (unsigned long) listLength(server.io_newjobs), - (unsigned long) listLength(server.io_processing), - (unsigned long) listLength(server.io_processed), - (unsigned long) server.io_active_threads, - (unsigned long) server.vm_blocked_clients + "cache_max_memory:%llu\r\n" + "cache_blocked_clients:%lu\r\n" + ,(unsigned long long) server.cache_max_memory, + (unsigned long) server.cache_blocked_clients ); unlockThreadedIO(); } @@ -1416,44 +1423,13 @@ void freeMemoryIfNeeded(void) { if (bestkey) { robj *keyobj = createStringObject(bestkey,sdslen(bestkey)); dbDelete(db,keyobj); - server.stat_expiredkeys++; + server.stat_evictedkeys++; decrRefCount(keyobj); freed++; } } if (!freed) return; /* nothing to free... */ } - - while(0) { - int j, k, freed = 0; - for (j = 0; j < server.dbnum; j++) { - int minttl = -1; - sds minkey = NULL; - robj *keyobj = NULL; - struct dictEntry *de; - - if (dictSize(server.db[j].expires)) { - freed = 1; - /* From a sample of three keys drop the one nearest to - * the natural expire */ - for (k = 0; k < 3; k++) { - time_t t; - - de = dictGetRandomKey(server.db[j].expires); - t = (time_t) dictGetEntryVal(de); - if (minttl == -1 || t < minttl) { - minkey = dictGetEntryKey(de); - minttl = t; - } - } - keyobj = createStringObject(minkey,sdslen(minkey)); - dbDelete(server.db+j,keyobj); - server.stat_expiredkeys++; - decrRefCount(keyobj); - } - } - if (!freed) return; /* nothing to free... */ - } } /* =================================== Main! ================================ */ @@ -1484,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) { - fprintf(fp,"%d\n",getpid()); + fprintf(fp,"%d\n",(int)getpid()); fclose(fp); } } @@ -1541,7 +1517,9 @@ int main(int argc, char **argv) { 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 {