X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/3a401464e9c42fafeaa9405db60d10b3256aca51..1a3e9d951be14fecf29cb7ad3d5f068e151b66f1:/src/redis.c diff --git a/src/redis.c b/src/redis.c index 4cc378e2..46915c13 100644 --- a/src/redis.c +++ b/src/redis.c @@ -242,7 +242,9 @@ struct redisCommand redisCommandTable[] = { {"evalsha",evalShaCommand,-3,"s",0,zunionInterGetKeys,0,0,0,0,0}, {"slowlog",slowlogCommand,-2,"r",0,NULL,0,0,0,0,0}, {"script",scriptCommand,-2,"ras",0,NULL,0,0,0,0,0}, - {"time",timeCommand,1,"rR",0,NULL,0,0,0,0,0} + {"time",timeCommand,1,"rR",0,NULL,0,0,0,0,0}, + {"bitop",bitopCommand,-4,"wm",0,NULL,2,-1,1,0,0}, + {"bitcount",bitcountCommand,-2,"r",0,NULL,1,1,1,0,0} }; /*============================ Utility functions ============================ */ @@ -581,10 +583,16 @@ void incrementallyRehash(void) { int j; for (j = 0; j < server.dbnum; j++) { + /* Keys dictionary */ if (dictIsRehashing(server.db[j].dict)) { dictRehashMilliseconds(server.db[j].dict,1); break; /* already used our millisecond for this loop... */ } + /* Expires */ + if (dictIsRehashing(server.db[j].expires)) { + dictRehashMilliseconds(server.db[j].expires,1); + break; /* already used our millisecond for this loop... */ + } } } @@ -608,19 +616,35 @@ void updateDictResizePolicy(void) { * it will get more aggressive to avoid that too much memory is used by * keys that can be removed from the keyspace. */ void activeExpireCycle(void) { - int j; - long long start = mstime(); + int j, iteration = 0; + long long start = ustime(), timelimit; + + /* We can use at max REDIS_EXPIRELOOKUPS_TIME_PERC percentage of CPU time + * per iteration. Since this function gets called with a frequency of + * REDIS_HZ times per second, the following is the max amount of + * microseconds we can spend in this function. */ + timelimit = 1000000*REDIS_EXPIRELOOKUPS_TIME_PERC/REDIS_HZ/100; + if (timelimit <= 0) timelimit = 1; for (j = 0; j < server.dbnum; j++) { - int expired, iteration = 0; + int expired; redisDb *db = server.db+j; /* Continue to expire if at the end of the cycle more than 25% * of the keys were expired. */ do { - long num = dictSize(db->expires); + unsigned long num = dictSize(db->expires); + unsigned long slots = dictSlots(db->expires); long long now = mstime(); + /* When there are less than 1% filled slots getting random + * keys is expensive, so stop here waiting for better times... + * The dictionary will be resized asap. */ + if (num && slots > DICT_HT_INITIAL_SIZE && + (num*100/slots < 1)) break; + + /* The main collection cycle. Sample random keys among keys + * with an expire set, checking for expired ones. */ expired = 0; if (num > REDIS_EXPIRELOOKUPS_PER_CRON) num = REDIS_EXPIRELOOKUPS_PER_CRON; @@ -645,8 +669,8 @@ void activeExpireCycle(void) { * expire. So after a given amount of milliseconds return to the * caller waiting for the other active expire cycle. */ iteration++; - if ((iteration & 0xff) == 0 && /* & 0xff is the same as % 255 */ - (mstime()-start) > REDIS_EXPIRELOOKUPS_TIME_LIMIT) return; + if ((iteration & 0xf) == 0 && /* check once every 16 cycles. */ + (ustime()-start) > timelimit) return; } while (expired > REDIS_EXPIRELOOKUPS_PER_CRON/4); } } @@ -732,13 +756,13 @@ int clientsCronResizeQueryBuffer(redisClient *c) { } void clientsCron(void) { - /* Make sure to process at least 1/100 of clients per call. - * Since this function is called 10 times per second we are sure that + /* Make sure to process at least 1/(REDIS_HZ*10) of clients per call. + * Since this function is called REDIS_HZ times per second we are sure that * in the worst case we process all the clients in 10 seconds. * In normal conditions (a reasonable number of clients) we process * all the clients in a shorter time. */ int numclients = listLength(server.clients); - int iterations = numclients/100; + int iterations = numclients/(REDIS_HZ*10); if (iterations < 50) iterations = (numclients < 50) ? numclients : 50; @@ -760,6 +784,30 @@ void clientsCron(void) { } } +/* This is our timer interrupt, called REDIS_HZ times per second. + * Here is where we do a number of things that need to be done asynchronously. + * For instance: + * + * - Active expired keys collection (it is also performed in a lazy way on + * lookup). + * - Software watchdong. + * - Update some statistic. + * - Incremental rehashing of the DBs hash tables. + * - Triggering BGSAVE / AOF rewrite, and handling of terminated children. + * - Clients timeout of differnet kinds. + * - Replication reconnection. + * - Many more... + * + * Everything directly called here will be called REDIS_HZ times per second, + * so in order to throttle execution of things we want to do less frequently + * a macro is used: run_with_period(milliseconds) { .... } + */ + +/* Using the following macro you can run code inside serverCron() with the + * specified period, specified in milliseconds. + * The actual resolution depends on REDIS_HZ. */ +#define run_with_period(_ms_) if (!(loops % ((_ms_)/(1000/REDIS_HZ)))) + int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { int j, loops = server.cronloops; REDIS_NOTUSED(eventLoop); @@ -776,7 +824,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { * To access a global var is faster than calling time(NULL) */ server.unixtime = time(NULL); - trackOperationsPerSecond(); + run_with_period(100) trackOperationsPerSecond(); /* We have just 22 bits per object for LRU information. * So we use an (eventually wrapping) LRU clock with 10 seconds resolution. @@ -804,15 +852,17 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { } /* Show some info about non-empty databases */ - for (j = 0; j < server.dbnum; j++) { - long long size, used, vkeys; - - size = dictSlots(server.db[j].dict); - used = dictSize(server.db[j].dict); - vkeys = dictSize(server.db[j].expires); - if (!(loops % 50) && (used || vkeys)) { - redisLog(REDIS_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size); - /* dictPrintStats(server.dict); */ + run_with_period(5000) { + for (j = 0; j < server.dbnum; j++) { + long long size, used, vkeys; + + size = dictSlots(server.db[j].dict); + used = dictSize(server.db[j].dict); + vkeys = dictSize(server.db[j].expires); + if (used || vkeys) { + redisLog(REDIS_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size); + /* dictPrintStats(server.dict); */ + } } } @@ -823,12 +873,12 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { * a lot of memory movements in the parent will cause a lot of pages * copied. */ if (server.rdb_child_pid == -1 && server.aof_child_pid == -1) { - if (!(loops % 10)) tryResizeHashTables(); + tryResizeHashTables(); if (server.activerehashing) incrementallyRehash(); } /* Show information about connected clients */ - if (!(loops % 50)) { + run_with_period(5000) { redisLog(REDIS_VERBOSE,"%d clients connected (%d slaves), %zu bytes in use", listLength(server.clients)-listLength(server.slaves), listLength(server.slaves), @@ -910,10 +960,10 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) { /* Replication cron function -- used to reconnect to master and * to detect transfer failures. */ - if (!(loops % 10)) replicationCron(); + run_with_period(1000) replicationCron(); server.cronloops++; - return 100; + return 1000/REDIS_HZ; } /* This function gets called every time Redis is entering the @@ -1043,6 +1093,8 @@ void initServerConfig() { server.aof_rewrite_base_size = 0; server.aof_rewrite_scheduled = 0; server.aof_last_fsync = time(NULL); + server.aof_rewrite_time_last = -1; + server.aof_rewrite_time_start = -1; server.aof_delayed_fsync = 0; server.aof_fd = -1; server.aof_selected_db = -1; /* Make sure the first time will not match */ @@ -1231,9 +1283,11 @@ void initServer() { server.cronloops = 0; server.rdb_child_pid = -1; server.aof_child_pid = -1; - server.aof_rewrite_buf = sdsempty(); + aofRewriteBufferReset(); server.aof_buf = sdsempty(); server.lastsave = time(NULL); + server.rdb_save_time_last = -1; + server.rdb_save_time_start = -1; server.dirty = 0; server.stat_numcommands = 0; server.stat_numconnections = 0; @@ -1806,21 +1860,31 @@ sds genRedisInfoString(char *section) { info = sdscatprintf(info, "# Persistence\r\n" "loading:%d\r\n" + "rdb_changes_since_last_save:%lld\r\n" + "rdb_bgsave_in_progress:%d\r\n" + "rdb_last_save_time:%ld\r\n" + "rdb_last_bgsave_status:%s\r\n" + "rdb_last_bgsave_time_sec:%ld\r\n" + "rdb_current_bgsave_time_sec:%ld\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" - "last_bgsave_status:%s\r\n" - "bgrewriteaof_in_progress:%d\r\n" - "bgrewriteaof_scheduled:%d\r\n", + "aof_rewrite_in_progress:%d\r\n" + "aof_rewrite_scheduled:%d\r\n" + "aof_last_rewrite_time_sec:%ld\r\n" + "aof_current_rewrite_time_sec:%ld\r\n", server.loading, - server.aof_state != REDIS_AOF_OFF, server.dirty, server.rdb_child_pid != -1, server.lastsave, server.lastbgsave_status == REDIS_OK ? "ok" : "err", + server.rdb_save_time_last, + (server.rdb_child_pid == -1) ? + -1 : time(NULL)-server.rdb_save_time_start, + server.aof_state != REDIS_AOF_OFF, server.aof_child_pid != -1, - server.aof_rewrite_scheduled); + server.aof_rewrite_scheduled, + server.aof_rewrite_time_last, + (server.aof_child_pid == -1) ? + -1 : time(NULL)-server.aof_rewrite_time_start); if (server.aof_state != REDIS_AOF_OFF) { info = sdscatprintf(info, @@ -1828,12 +1892,14 @@ sds genRedisInfoString(char *section) { "aof_base_size:%lld\r\n" "aof_pending_rewrite:%d\r\n" "aof_buffer_length:%zu\r\n" + "aof_rewrite_buffer_length:%zu\r\n" "aof_pending_bio_fsync:%llu\r\n" "aof_delayed_fsync:%lu\r\n", (long long) server.aof_current_size, (long long) server.aof_rewrite_base_size, server.aof_rewrite_scheduled, sdslen(server.aof_buf), + aofRewriteBufferSize(), bioPendingJobsOfType(REDIS_BIO_AOF_FSYNC), server.aof_delayed_fsync); } @@ -2086,7 +2152,7 @@ int freeMemoryIfNeeded(void) { } if (server.aof_state != REDIS_AOF_OFF) { mem_used -= sdslen(server.aof_buf); - mem_used -= sdslen(server.aof_rewrite_buf); + mem_used -= aofRewriteBufferSize(); } /* Check if we are over the memory limit. */