X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/64a13a36e685c318319a70b775f91f2c34bcc34f..c9d0c3623a7714bd41a35237f4ba927206a7adb6:/src/redis.c diff --git a/src/redis.c b/src/redis.c index bf9ced2c..89efffba 100644 --- a/src/redis.c +++ b/src/redis.c @@ -50,7 +50,6 @@ #include #include #include -#include #include /* Our shared "common" objects */ @@ -103,7 +102,7 @@ struct redisCommand redisCommandTable[] = { {"lrem",lremCommand,4,0,NULL,1,1,1,0,0}, {"rpoplpush",rpoplpushCommand,3,REDIS_CMD_DENYOOM,NULL,1,2,1,0,0}, {"sadd",saddCommand,-3,REDIS_CMD_DENYOOM,NULL,1,1,1,0,0}, - {"srem",sremCommand,3,0,NULL,1,1,1,0,0}, + {"srem",sremCommand,-3,0,NULL,1,1,1,0,0}, {"smove",smoveCommand,4,0,NULL,1,2,1,0,0}, {"sismember",sismemberCommand,3,0,NULL,1,1,1,0,0}, {"scard",scardCommand,2,0,NULL,1,1,1,0,0}, @@ -116,9 +115,9 @@ struct redisCommand redisCommandTable[] = { {"sdiff",sdiffCommand,-2,REDIS_CMD_DENYOOM,NULL,1,-1,1,0,0}, {"sdiffstore",sdiffstoreCommand,-3,REDIS_CMD_DENYOOM,NULL,2,-1,1,0,0}, {"smembers",sinterCommand,2,0,NULL,1,1,1,0,0}, - {"zadd",zaddCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1,0,0}, + {"zadd",zaddCommand,-4,REDIS_CMD_DENYOOM,NULL,1,1,1,0,0}, {"zincrby",zincrbyCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1,0,0}, - {"zrem",zremCommand,3,0,NULL,1,1,1,0,0}, + {"zrem",zremCommand,-3,0,NULL,1,1,1,0,0}, {"zremrangebyscore",zremrangebyscoreCommand,4,0,NULL,1,1,1,0,0}, {"zremrangebyrank",zremrangebyrankCommand,4,0,NULL,1,1,1,0,0}, {"zunionstore",zunionstoreCommand,-4,REDIS_CMD_DENYOOM,zunionInterGetKeys,0,0,0,0,0}, @@ -192,7 +191,8 @@ struct redisCommand redisCommandTable[] = { {"restore",restoreCommand,4,0,NULL,0,0,0,0,0}, {"migrate",migrateCommand,6,0,NULL,0,0,0,0,0}, {"dump",dumpCommand,2,0,NULL,0,0,0,0,0}, - {"object",objectCommand,-2,0,NULL,0,0,0,0,0} + {"object",objectCommand,-2,0,NULL,0,0,0,0,0}, + {"client",clientCommand,-2,0,NULL,0,0,0,0,0} }; /*============================ Utility functions ============================ */ @@ -585,6 +585,10 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { */ updateLRUClock(); + /* Record the max memory used since the server was started. */ + if (zmalloc_used_memory() > server.stat_peak_memory) + server.stat_peak_memory = zmalloc_used_memory(); + /* We received a SIGTERM, shutting down here in a safe way, as it is * not ok doing so inside the signal handler. */ if (server.shutdown_asap) { @@ -628,6 +632,14 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { if ((server.maxidletime && !(loops % 100)) || server.bpop_blocked_clients) closeTimedoutClients(); + /* Start a scheduled AOF rewrite if this was requested by the user while + * a BGSAVE was in progress. */ + if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1 && + server.aofrewrite_scheduled) + { + rewriteAppendOnlyFileBackground(); + } + /* Check if a background saving or AOF rewrite in progress terminated. */ if (server.bgsavechildpid != -1 || server.bgrewritechildpid != -1) { int statloc; @@ -646,25 +658,11 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { } updateDictResizePolicy(); } - } else if (server.bgsavethread != (pthread_t) -1) { - 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_THREAD_DONE_OK || - state == REDIS_BGSAVE_THREAD_DONE_ERR) - { - backgroundSaveDoneHandler( - (state == REDIS_BGSAVE_THREAD_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 */ + } else { time_t now = time(NULL); + + /* If there is not a background saving/rewrite in progress check if + * we have to save/rewrite now */ for (j = 0; j < server.saveparamslen; j++) { struct saveparam *sp = server.saveparams+j; @@ -676,6 +674,21 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { break; } } + + /* Trigger an AOF rewrite if needed */ + if (server.bgsavechildpid == -1 && + server.bgrewritechildpid == -1 && + server.auto_aofrewrite_perc && + server.appendonly_current_size > server.auto_aofrewrite_min_size) + { + int base = server.auto_aofrewrite_base_size ? + server.auto_aofrewrite_base_size : 1; + long long growth = (server.appendonly_current_size*100/base) - 100; + if (growth >= server.auto_aofrewrite_perc) { + redisLog(REDIS_NOTICE,"Starting automatic rewriting of AOF on %lld%% growth",growth); + rewriteAppendOnlyFileBackground(); + } + } } /* Expire a few keys per cycle, only if this is a master. @@ -683,10 +696,6 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { * in order to guarantee a strict consistency. */ if (server.masterhost == NULL) activeExpireCycle(); - /* 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. */ if (!(loops % 10)) replicationCron(); @@ -706,31 +715,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) { listNode *ln; redisClient *c; - /* Awake clients that got all the on disk keys they requested */ - if (server.ds_enabled && listLength(server.io_ready_clients)) { - listIter li; - - listRewind(server.io_ready_clients,&li); - while((ln = listNext(&li))) { - c = ln->value; - struct redisCommand *cmd; - - /* Resume the client. */ - listDelNode(server.io_ready_clients,ln); - c->flags &= (~REDIS_IO_WAIT); - server.cache_blocked_clients--; - aeCreateFileEvent(server.el, c->fd, AE_READABLE, - readQueryFromClient, c); - cmd = lookupCommand(c->argv[0]->ptr); - redisAssert(cmd != NULL); - call(c,cmd); - resetClient(c); - /* There may be more data to process in the input buffer. */ - if (c->querybuf && sdslen(c->querybuf) > 0) - processInputBuffer(c); - } - } - /* Try to process pending commands for clients that were just unblocked. */ while (listLength(server.unblocked_clients)) { ln = listFirst(server.unblocked_clients); @@ -823,6 +807,10 @@ void initServerConfig() { server.appendonly = 0; server.appendfsync = APPENDFSYNC_EVERYSEC; server.no_appendfsync_on_rewrite = 0; + server.auto_aofrewrite_perc = REDIS_AUTO_AOFREWRITE_PERC; + server.auto_aofrewrite_min_size = REDIS_AUTO_AOFREWRITE_MIN_SIZE; + server.auto_aofrewrite_base_size = 0; + server.aofrewrite_scheduled = 0; server.lastfsync = time(NULL); server.appendfd = -1; server.appendseldb = -1; /* Make sure the first time will not match */ @@ -837,10 +825,6 @@ void initServerConfig() { server.maxmemory = 0; server.maxmemory_policy = REDIS_MAXMEMORY_VOLATILE_LRU; server.maxmemory_samples = 3; - 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; @@ -849,7 +833,6 @@ void initServerConfig() { server.zset_max_ziplist_entries = REDIS_ZSET_MAX_ZIPLIST_ENTRIES; server.zset_max_ziplist_value = REDIS_ZSET_MAX_ZIPLIST_VALUE; server.shutdown_asap = 0; - server.cache_flush_delay = 0; server.cluster_enabled = 0; server.cluster.configfile = zstrdup("nodes.conf"); @@ -866,7 +849,9 @@ void initServerConfig() { server.masterport = 6379; server.master = NULL; server.replstate = REDIS_REPL_NONE; + server.repl_syncio_timeout = REDIS_REPL_SYNCIO_TIMEOUT; server.repl_serve_stale_data = 1; + server.repl_down_since = -1; /* Double constants initialization */ R_Zero = 0.0; @@ -895,12 +880,10 @@ void initServer() { server.syslog_facility); } - server.mainthread = pthread_self(); server.clients = listCreate(); server.slaves = listCreate(); server.monitors = listCreate(); server.unblocked_clients = listCreate(); - server.cache_io_queue = listCreate(); createSharedObjects(); server.el = aeCreateEventLoop(); @@ -930,11 +913,6 @@ 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.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); @@ -944,8 +922,6 @@ 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); @@ -957,6 +933,8 @@ void initServer() { server.stat_starttime = time(NULL); server.stat_keyspace_misses = 0; server.stat_keyspace_hits = 0; + server.stat_peak_memory = 0; + server.stat_fork_time = 0; server.unixtime = time(NULL); aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL); if (server.ipfd > 0 && aeCreateFileEvent(server.el,server.ipfd,AE_READABLE, @@ -973,7 +951,6 @@ void initServer() { } } - if (server.ds_enabled) dsInit(); if (server.cluster_enabled) clusterInit(); srand(time(NULL)^getpid()); } @@ -1090,13 +1067,15 @@ int processCommand(redisClient *c) { addReplyError(c,"The cluster is down. Check with CLUSTER INFO for more information"); return REDIS_OK; } else { - clusterNode *n = getNodeByQuery(c,cmd,c->argv,c->argc,&hashslot); + int ask; + clusterNode *n = getNodeByQuery(c,cmd,c->argv,c->argc,&hashslot,&ask); if (n == NULL) { - addReplyError(c,"Invalid cross-node request"); + addReplyError(c,"Multi keys request invalid in cluster"); return REDIS_OK; } else if (n != server.cluster.myself) { addReplySds(c,sdscatprintf(sdsempty(), - "-MOVED %d %s:%d\r\n",hashslot,n->ip,n->port)); + "-%s %d %s:%d\r\n", ask ? "ASK" : "MOVED", + hashslot,n->ip,n->port)); return REDIS_OK; } } @@ -1149,8 +1128,6 @@ int processCommand(redisClient *c) { queueMultiCommand(c,cmd); addReply(c,shared.queued); } else { - if (server.ds_enabled && blockClientOnSwappedKeys(c,cmd)) - return REDIS_ERR; call(c,cmd); } return REDIS_OK; @@ -1168,9 +1145,7 @@ int prepareForShutdown() { kill(server.bgsavechildpid,SIGKILL); rdbRemoveTempFile(server.bgsavechildpid); } - if (server.ds_enabled) { - /* FIXME: flush all objects on disk */ - } else if (server.appendonly) { + if (server.appendonly) { /* Append only file: fsync() the AOF and exit */ aof_fsync(server.appendfd); } else if (server.saveparamslen > 0) { @@ -1240,7 +1215,6 @@ sds genRedisInfoString(char *section) { sds info = sdsempty(); time_t uptime = time(NULL)-server.stat_starttime; int j, numcommands; - char hmem[64]; struct rusage self_ru, c_ru; unsigned long lol, bib; int allsections = 0, defsections = 0; @@ -1254,7 +1228,6 @@ sds genRedisInfoString(char *section) { getrusage(RUSAGE_SELF, &self_ru); getrusage(RUSAGE_CHILDREN, &c_ru); getClientsMaxBuffers(&lol,&bib); - bytesToHuman(hmem,zmalloc_used_memory()); /* Server */ if (allsections || defsections || !strcasecmp(section,"server")) { @@ -1299,23 +1272,28 @@ sds genRedisInfoString(char *section) { /* Memory */ if (allsections || defsections || !strcasecmp(section,"memory")) { + char hmem[64]; + char peak_hmem[64]; + + bytesToHuman(hmem,zmalloc_used_memory()); + bytesToHuman(peak_hmem,server.stat_peak_memory); if (sections++) info = sdscat(info,"\r\n"); info = sdscatprintf(info, "# Memory\r\n" "used_memory:%zu\r\n" "used_memory_human:%s\r\n" "used_memory_rss:%zu\r\n" + "used_memory_peak:%zu\r\n" + "used_memory_peak_human:%s\r\n" "mem_fragmentation_ratio:%.2f\r\n" - "use_tcmalloc:%d\r\n", + "mem_allocator:%s\r\n", zmalloc_used_memory(), hmem, zmalloc_get_rss(), + server.stat_peak_memory, + peak_hmem, zmalloc_get_fragmentation_ratio(), - #ifdef USE_TCMALLOC - 1 - #else - 0 - #endif + ZMALLOC_LIB ); } @@ -1349,11 +1327,20 @@ sds genRedisInfoString(char *section) { server.loading, server.appendonly, server.dirty, - server.bgsavechildpid != -1 || - server.bgsavethread != (pthread_t) -1, + server.bgsavechildpid != -1, server.lastsave, server.bgrewritechildpid != -1); + if (server.appendonly) { + info = sdscatprintf(info, + "aof_current_size:%lld\r\n" + "aof_base_size:%lld\r\n" + "aof_pending_rewrite:%d\r\n", + (long long) server.appendonly_current_size, + (long long) server.auto_aofrewrite_base_size, + server.aofrewrite_scheduled); + } + if (server.loading) { double perc; time_t eta, elapsed; @@ -1386,35 +1373,6 @@ sds genRedisInfoString(char *section) { } } - /* Diskstore */ - if (allsections || defsections || !strcasecmp(section,"diskstore")) { - if (sections++) info = sdscat(info,"\r\n"); - info = sdscatprintf(info, - "# Diskstore\r\n" - "ds_enabled:%d\r\n", - server.ds_enabled != 0); - if (server.ds_enabled) { - lockThreadedIO(); - info = sdscatprintf(info, - "cache_max_memory:%llu\r\n" - "cache_blocked_clients:%lu\r\n" - "cache_io_queue_len:%lu\r\n" - "cache_io_jobs_new:%lu\r\n" - "cache_io_jobs_processing:%lu\r\n" - "cache_io_jobs_processed:%lu\r\n" - "cache_io_ready_clients:%lu\r\n" - ,(unsigned long long) server.cache_max_memory, - (unsigned long) server.cache_blocked_clients, - (unsigned long) listLength(server.cache_io_queue), - (unsigned long) listLength(server.io_newjobs), - (unsigned long) listLength(server.io_processing), - (unsigned long) listLength(server.io_processed), - (unsigned long) listLength(server.io_ready_clients) - ); - unlockThreadedIO(); - } - } - /* Stats */ if (allsections || defsections || !strcasecmp(section,"stats")) { if (sections++) info = sdscat(info,"\r\n"); @@ -1427,7 +1385,8 @@ sds genRedisInfoString(char *section) { "keyspace_hits:%lld\r\n" "keyspace_misses:%lld\r\n" "pubsub_channels:%ld\r\n" - "pubsub_patterns:%u\r\n", + "pubsub_patterns:%u\r\n" + "latest_fork_usec:%lld\r\n", server.stat_numconnections, server.stat_numcommands, server.stat_expiredkeys, @@ -1435,7 +1394,8 @@ sds genRedisInfoString(char *section) { server.stat_keyspace_hits, server.stat_keyspace_misses, dictSize(server.pubsub_channels), - listLength(server.pubsub_patterns)); + listLength(server.pubsub_patterns), + server.stat_fork_time); } /* Replication */ @@ -1469,6 +1429,12 @@ sds genRedisInfoString(char *section) { (int)(time(NULL)-server.repl_transfer_lastio) ); } + + if (server.replstate != REDIS_REPL_CONNECTED) { + info = sdscatprintf(info, + "master_link_down_since_seconds:%ld\r\n", + (long)time(NULL)-server.repl_down_since); + } } info = sdscatprintf(info, "connected_slaves:%d\r\n", @@ -1764,9 +1730,7 @@ int main(int argc, char **argv) { linuxOvercommitMemoryWarning(); #endif start = ustime(); - if (server.ds_enabled) { - redisLog(REDIS_NOTICE,"DB not loaded (running with disk back end)"); - } else if (server.appendonly) { + if (server.appendonly) { if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK) redisLog(REDIS_NOTICE,"DB loaded from append only file: %.3f seconds",(float)(ustime()-start)/1000000); } else {