X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/ca734d17ad9b7ea97a87bd048ad1c1e253ef67a6..d934e1e85b1ec1c6d2cd07292cced53770f4a626:/src/redis.c diff --git a/src/redis.c b/src/redis.c index 99ff1898..65cf3ace 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}, @@ -124,7 +129,7 @@ struct redisCommand readonlyCommandTable[] = { {"zcount",zcountCommand,4,0,NULL,1,1,1}, {"zrevrange",zrevrangeCommand,-4,0,NULL,1,1,1}, {"zcard",zcardCommand,2,0,NULL,1,1,1}, - {"zscore",zscoreCommand,3,REDIS_CMD_DENYOOM,NULL,1,1,1}, + {"zscore",zscoreCommand,3,0,NULL,1,1,1}, {"zrank",zrankCommand,3,0,NULL,1,1,1}, {"zrevrank",zrevrankCommand,3,0,NULL,1,1,1}, {"hset",hsetCommand,4,REDIS_CMD_DENYOOM,NULL,1,1,1}, @@ -188,11 +193,13 @@ struct redisCommand readonlyCommandTable[] = { /*============================ Utility functions ============================ */ void redisLog(int level, const char *fmt, ...) { + const int syslogLevelMap[] = { LOG_DEBUG, LOG_INFO, LOG_NOTICE, LOG_WARNING }; + const char *c = ".-*#"; + time_t now = time(NULL); va_list ap; FILE *fp; - char *c = ".-*#"; char buf[64]; - time_t now; + char msg[REDIS_MAX_LOGMSG_LEN]; if (level < server.verbosity) return; @@ -200,15 +207,16 @@ void redisLog(int level, const char *fmt, ...) { if (!fp) return; va_start(ap, fmt); - now = time(NULL); - strftime(buf,64,"%d %b %H:%M:%S",localtime(&now)); - fprintf(fp,"[%d] %s %c ",(int)getpid(),buf,c[level]); - vfprintf(fp, fmt, ap); - fprintf(fp,"\n"); - fflush(fp); + vsnprintf(msg, sizeof(msg), fmt, ap); va_end(ap); + strftime(buf,sizeof(buf),"%d %b %H:%M:%S",localtime(&now)); + fprintf(fp,"[%d] %s %c %s\n",(int)getpid(),buf,c[level],msg); + fflush(fp); + if (server.logfile) fclose(fp); + + if (server.syslog_enabled) syslog(syslogLevelMap[level], "%s", msg); } /* Redis generally does not try to recover from out of memory conditions @@ -252,6 +260,15 @@ int dictSdsKeyCompare(void *privdata, const void *key1, return memcmp(key1, key2, l1) == 0; } +/* A case insensitive version used for the command lookup table. */ +int dictSdsKeyCaseCompare(void *privdata, const void *key1, + const void *key2) +{ + DICT_NOTUSED(privdata); + + return strcasecmp(key1, key2) == 0; +} + void dictRedisObjectDestructor(void *privdata, void *val) { DICT_NOTUSED(privdata); @@ -283,6 +300,10 @@ unsigned int dictSdsHash(const void *key) { return dictGenHashFunction((unsigned char*)key, sdslen((char*)key)); } +unsigned int dictSdsCaseHash(const void *key) { + return dictGenCaseHashFunction((unsigned char*)key, sdslen((char*)key)); +} + int dictEncObjKeyCompare(void *privdata, const void *key1, const void *key2) { @@ -324,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 */ @@ -364,6 +385,16 @@ dictType keyptrDictType = { NULL /* val destructor */ }; +/* Command table. sds string -> command struct pointer. */ +dictType commandTableDictType = { + dictSdsCaseHash, /* hash function */ + NULL, /* key dup */ + NULL, /* val dup */ + dictSdsKeyCaseCompare, /* key compare */ + dictSdsDestructor, /* key destructor */ + NULL /* val destructor */ +}; + /* Hash type hash table (note that small hashes are represented with zimpaps) */ dictType hashDictType = { dictEncObjHash, /* hash function */ @@ -549,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 */ @@ -565,7 +596,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { } updateDictResizePolicy(); } - } else { + } 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); @@ -587,43 +618,14 @@ 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; - - if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue; - 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(); - /* Check if we should connect to a MASTER */ - if (server.replstate == REDIS_REPL_CONNECT && !(loops % 10)) { - redisLog(REDIS_NOTICE,"Connecting to MASTER..."); - if (syncWithMaster() == REDIS_OK) { - redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync succeeded"); - if (server.appendonly) rewriteAppendOnlyFileBackground(); - } - } + /* Replication cron function -- used to reconnect to master and + * to detect transfer failures. */ + if (!(loops % 10)) replicationCron(); - /* Update fragmentation info, used for fast RSS estimation */ - if (!(loops % 10)) server.fragmentation = zmalloc_get_fragmentation_ratio(); return 100; } @@ -632,21 +634,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); @@ -658,6 +661,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(); } @@ -689,6 +705,8 @@ void createSharedObjects(void) { "-ERR source and destination objects are the same\r\n")); shared.outofrangeerr = createObject(REDIS_STRING,sdsnew( "-ERR index out of range\r\n")); + shared.loadingerr = createObject(REDIS_STRING,sdsnew( + "-LOADING Redis is loading the dataset in memory\r\n")); shared.space = createObject(REDIS_STRING,sdsnew(" ")); shared.colon = createObject(REDIS_STRING,sdsnew(":")); shared.plus = createObject(REDIS_STRING,sdsnew("+")); @@ -717,13 +735,20 @@ void createSharedObjects(void) { } void initServerConfig() { - server.dbnum = REDIS_DEFAULT_DBNUM; server.port = REDIS_SERVERPORT; + server.bindaddr = NULL; + server.unixsocket = NULL; + server.ipfd = -1; + server.sofd = -1; + server.dbnum = REDIS_DEFAULT_DBNUM; server.verbosity = REDIS_VERBOSE; server.maxidletime = REDIS_MAXIDLETIME; server.saveparams = NULL; + server.loading = 0; server.logfile = NULL; /* NULL = log on standard output */ - server.bindaddr = NULL; + server.syslog_enabled = 0; + server.syslog_ident = zstrdup("redis"); + server.syslog_facility = LOG_LOCAL0; server.glueoutputbuf = 1; server.daemonize = 0; server.appendonly = 0; @@ -738,19 +763,15 @@ void initServerConfig() { server.requirepass = NULL; server.rdbcompression = 1; server.activerehashing = 1; - server.fragmentation = 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; @@ -771,12 +792,21 @@ void initServerConfig() { server.masterport = 6379; server.master = NULL; server.replstate = REDIS_REPL_NONE; + server.repl_serve_stale_data = 1; /* Double constants initialization */ R_Zero = 0.0; R_PosInf = 1.0/R_Zero; R_NegInf = -1.0/R_Zero; R_Nan = R_Zero/R_Zero; + + /* Command table -- we intiialize it here as it is part of the + * initial configuration, since command names may be changed via + * redis.conf using the rename-command directive. */ + server.commands = dictCreate(&commandTableDictType,NULL); + populateCommandTable(); + server.delCommand = lookupCommandByCString("del"); + server.multiCommand = lookupCommandByCString("multi"); } void initServer() { @@ -786,22 +816,37 @@ void initServer() { signal(SIGPIPE, SIG_IGN); setupSigSegvAction(); - server.mainthread = pthread_self(); - server.devnull = fopen("/dev/null","w"); - if (server.devnull == NULL) { - redisLog(REDIS_WARNING, "Can't open /dev/null: %s", server.neterr); - exit(1); + if (server.syslog_enabled) { + openlog(server.syslog_ident, LOG_PID | LOG_NDELAY | LOG_NOWAIT, + server.syslog_facility); } + + server.mainthread = pthread_self(); server.clients = listCreate(); server.slaves = listCreate(); server.monitors = listCreate(); - server.objfreelist = listCreate(); + server.unblocked_clients = listCreate(); + server.cache_flush_queue = listCreate(); + server.cache_flush_delay = 0; + createSharedObjects(); server.el = aeCreateEventLoop(); server.db = zmalloc(sizeof(redisDb)*server.dbnum); - server.fd = anetTcpServer(server.neterr, server.port, server.bindaddr); - if (server.fd == -1) { - redisLog(REDIS_WARNING, "Opening TCP port: %s", server.neterr); + server.ipfd = anetTcpServer(server.neterr,server.port,server.bindaddr); + if (server.ipfd == ANET_ERR) { + redisLog(REDIS_WARNING, "Opening port: %s", server.neterr); + exit(1); + } + if (server.unixsocket != NULL) { + unlink(server.unixsocket); /* don't care if this fails */ + server.sofd = anetUnixServer(server.neterr,server.unixsocket); + if (server.sofd == ANET_ERR) { + redisLog(REDIS_WARNING, "Opening socket: %s", server.neterr); + exit(1); + } + } + if (server.ipfd < 0 && server.sofd < 0) { + redisLog(REDIS_WARNING, "Configured to not listen anywhere, exiting."); exit(1); } for (j = 0; j < server.dbnum; j++) { @@ -809,8 +854,10 @@ 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].id = j; } server.pubsub_channels = dictCreate(&keylistDictType,NULL); @@ -827,13 +874,16 @@ 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; server.unixtime = time(NULL); aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL); - if (aeCreateFileEvent(server.el, server.fd, AE_READABLE, - acceptHandler, NULL) == AE_ERR) oom("creating file event"); + if (server.ipfd > 0 && aeCreateFileEvent(server.el,server.ipfd,AE_READABLE, + acceptTcpHandler,NULL) == AE_ERR) oom("creating file event"); + if (server.sofd > 0 && aeCreateFileEvent(server.el,server.sofd,AE_READABLE, + acceptUnixHandler,NULL) == AE_ERR) oom("creating file event"); if (server.appendonly) { server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644); @@ -844,34 +894,37 @@ void initServer() { } } - if (server.vm_enabled) vmInit(); + if (server.ds_enabled) dsInit(); } -int qsortRedisCommands(const void *r1, const void *r2) { - return strcasecmp( - ((struct redisCommand*)r1)->name, - ((struct redisCommand*)r2)->name); -} +/* Populates the Redis Command Table starting from the hard coded list + * we have on top of redis.c file. */ +void populateCommandTable(void) { + int j; + int numcommands = sizeof(readonlyCommandTable)/sizeof(struct redisCommand); + + for (j = 0; j < numcommands; j++) { + struct redisCommand *c = readonlyCommandTable+j; + int retval; -void sortCommandTable() { - /* Copy and sort the read-only version of the command table */ - commandTable = (struct redisCommand*)zmalloc(sizeof(readonlyCommandTable)); - memcpy(commandTable,readonlyCommandTable,sizeof(readonlyCommandTable)); - qsort(commandTable, - sizeof(readonlyCommandTable)/sizeof(struct redisCommand), - sizeof(struct redisCommand),qsortRedisCommands); + retval = dictAdd(server.commands, sdsnew(c->name), c); + assert(retval == DICT_OK); + } } /* ====================== Commands lookup and execution ===================== */ -struct redisCommand *lookupCommand(char *name) { - struct redisCommand tmp = {name,NULL,0,0,NULL,0,0,0}; - return bsearch( - &tmp, - commandTable, - sizeof(readonlyCommandTable)/sizeof(struct redisCommand), - sizeof(struct redisCommand), - qsortRedisCommands); +struct redisCommand *lookupCommand(sds name) { + return dictFetchValue(server.commands, name); +} + +struct redisCommand *lookupCommandByCString(char *s) { + struct redisCommand *cmd; + sds name = sdsnew(s); + + cmd = dictFetchValue(server.commands, name); + sdsfree(name); + return cmd; } /* Call() is the core of Redis execution of a command */ @@ -955,6 +1008,23 @@ int processCommand(redisClient *c) { return REDIS_OK; } + /* Only allow INFO and SLAVEOF when slave-serve-stale-data is no and + * we are a slave with a broken link with master. */ + if (server.masterhost && server.replstate != REDIS_REPL_CONNECTED && + server.repl_serve_stale_data == 0 && + cmd->proc != infoCommand && cmd->proc != slaveofCommand) + { + addReplyError(c, + "link with MASTER is down and slave-serve-stale-data is set to no"); + return REDIS_OK; + } + + /* Loading DB? Return an error if the command is not INFO */ + if (server.loading && cmd->proc != infoCommand) { + addReply(c, shared.loadingerr); + return REDIS_OK; + } + /* Exec the command */ if (c->flags & REDIS_MULTI && cmd->proc != execCommand && cmd->proc != discardCommand && @@ -963,8 +1033,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; @@ -982,10 +1052,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) { @@ -1080,9 +1151,10 @@ sds genRedisInfoString(void) { "used_memory:%zu\r\n" "used_memory_human:%s\r\n" "used_memory_rss:%zu\r\n" - "used_memory_estimated_rss:%zu\r\n" "mem_fragmentation_ratio:%.2f\r\n" "use_tcmalloc:%d\r\n" + "loading:%d\r\n" + "aof_enabled:%d\r\n" "changes_since_last_save:%lld\r\n" "bgsave_in_progress:%d\r\n" "last_save_time:%ld\r\n" @@ -1090,13 +1162,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(), @@ -1113,17 +1186,18 @@ 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(), - redisEstimateRSS(), zmalloc_get_fragmentation_ratio(), #ifdef USE_TCMALLOC 1, #else 0, #endif + server.loading, + server.appendonly, server.dirty, server.bgsavechildpid != -1, server.lastsave, @@ -1131,13 +1205,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) { @@ -1146,43 +1221,63 @@ sds genRedisInfoString(void) { "master_port:%d\r\n" "master_link_status:%s\r\n" "master_last_io_seconds_ago:%d\r\n" + "master_sync_in_progress:%d\r\n" ,server.masterhost, server.masterport, (server.replstate == REDIS_REPL_CONNECTED) ? "up" : "down", - server.master ? ((int)(time(NULL)-server.master->lastinteraction)) : -1 + server.master ? ((int)(time(NULL)-server.master->lastinteraction)) : -1, + server.replstate == REDIS_REPL_TRANSFER ); + + if (server.replstate == REDIS_REPL_TRANSFER) { + info = sdscatprintf(info, + "master_sync_left_bytes:%ld\r\n" + "master_sync_last_io_seconds_ago:%d\r\n" + ,(long)server.repl_transfer_left, + (int)(time(NULL)-server.repl_transfer_lastio) + ); + } } - 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(); } + if (server.loading) { + double perc; + time_t eta, elapsed; + off_t remaining_bytes = server.loading_total_bytes- + server.loading_loaded_bytes; + + perc = ((double)server.loading_loaded_bytes / + server.loading_total_bytes) * 100; + + elapsed = time(NULL)-server.loading_start_time; + if (elapsed == 0) { + eta = 1; /* A fake 1 second figure if we don't have enough info */ + } else { + eta = (elapsed*remaining_bytes)/server.loading_loaded_bytes; + } + + info = sdscatprintf(info, + "loading_start_time:%ld\r\n" + "loading_total_bytes:%llu\r\n" + "loading_loaded_bytes:%llu\r\n" + "loading_loaded_perc:%.2f\r\n" + "loading_eta_seconds:%ld\r\n" + ,(unsigned long) server.loading_start_time, + (unsigned long long) server.loading_total_bytes, + (unsigned long long) server.loading_loaded_bytes, + perc, + eta + ); + } for (j = 0; j < server.dbnum; j++) { long long keys, vkeys; @@ -1216,61 +1311,6 @@ void monitorCommand(redisClient *c) { /* ============================ Maxmemory directive ======================== */ -/* Try to free one object form the pre-allocated objects free list. - * This is useful under low mem conditions as by default we take 1 million - * free objects allocated. On success REDIS_OK is returned, otherwise - * REDIS_ERR. */ -int tryFreeOneObjectFromFreelist(void) { - robj *o; - - if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex); - if (listLength(server.objfreelist)) { - listNode *head = listFirst(server.objfreelist); - o = listNodeValue(head); - listDelNode(server.objfreelist,head); - if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex); - zfree(o); - return REDIS_OK; - } else { - if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex); - return REDIS_ERR; - } -} - -/* A fast RSS sampling function. - * - * The function is reasonably accurate while fast, since it uses the trick of - * using the server.fragmentation ratio that is computed every second and - * is the ratio between the RSS and our zmalloc() count of allocated bytes. - * - * So in order to compute the current RSS used we just need to multiply - * the zmalloc() memory reporting, that is as fast as reading a counter, - * for the latest estimation of fragmentation. - * - * The behavior of this function is also very desirable because it is - * very responsive to memory changes: while the real RSS is actually measured - * in pages, the RSS estimation will actually change even if just a few bytes - * are freed, and this is a good property when the function is used in order - * to evict keys for Virtual Memory of for 'maxmemory' directive. - * - * Note that when the memory reported by zmalloc is smaller than the RSS - * (that is, fragmentation < 1) this means that something is odd (many pages - * swapped since the Redis instance is idle) and we consider the fragmentation - * ratio 1. */ -size_t redisEstimateRSS(void) { - size_t used = zmalloc_used_memory(); - float maxfrag; - - if (server.fragmentation < 1) return used; - maxfrag = (float)SIZE_MAX / used; - - /* If multiplying memory usage reported by zmalloc per fragmentation - * ratio will lead to an overflow we just return SIZE_MAX. */ - if (maxfrag < server.fragmentation) return SIZE_MAX; - - return (size_t)((used * server.fragmentation)); -} - /* This function gets called when 'maxmemory' is set on the config file to limit * the max memory used by the server, and we are out of memory. * This function will try to, in order: @@ -1285,12 +1325,11 @@ size_t redisEstimateRSS(void) { void freeMemoryIfNeeded(void) { /* Remove keys accordingly to the active policy as long as we are * over the memory limit. */ + if (server.maxmemory_policy == REDIS_MAXMEMORY_NO_EVICTION) return; + while (server.maxmemory && zmalloc_used_memory() > server.maxmemory) { int j, k, freed = 0; - /* Basic strategy -- remove objects from the free list. */ - if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue; - for (j = 0; j < server.dbnum; j++) { long bestval = 0; /* just to prevent warning */ sds bestkey = NULL; @@ -1326,6 +1365,10 @@ void freeMemoryIfNeeded(void) { de = dictGetRandomKey(dict); thiskey = dictGetEntryKey(de); + /* When policy is volatile-lru we need an additonal 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); o = dictGetEntryVal(de); thisval = estimateObjectIdleTime(o); @@ -1360,44 +1403,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! ================================ */ @@ -1466,7 +1478,6 @@ int main(int argc, char **argv) { time_t start; initServerConfig(); - sortCommandTable(); if (argc == 2) { if (strcmp(argv[1], "-v") == 0 || strcmp(argv[1], "--version") == 0) version(); @@ -1486,14 +1497,19 @@ 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 { if (rdbLoad(server.dbfilename) == REDIS_OK) redisLog(REDIS_NOTICE,"DB loaded from disk: %ld seconds",time(NULL)-start); } - redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port); + if (server.ipfd > 0) + redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port); + if (server.sofd > 0) + redisLog(REDIS_NOTICE,"The server is now ready to accept connections at %s", server.unixsocket); aeSetBeforeSleepProc(server.el,beforeSleep); aeMain(server.el); aeDeleteEventLoop(server.el);