Merge branch 'lfu-fixes' into unstable

src/config.c

@@ -330,13 +330,13 @@ void loadServerConfigFromString(char *config) {
             }
         } else if (!strcasecmp(argv[0],"lfu-log-factor") && argc == 2) {
             server.lfu_log_factor = atoi(argv[1]);
-            if (server.maxmemory_samples < 0) {
+            if (server.lfu_log_factor < 0) {
                 err = "lfu-log-factor must be 0 or greater";
                 goto loaderr;
             }
         } else if (!strcasecmp(argv[0],"lfu-decay-time") && argc == 2) {
             server.lfu_decay_time = atoi(argv[1]);
-            if (server.maxmemory_samples < 1) {
+            if (server.lfu_decay_time < 0) {
                 err = "lfu-decay-time must be 0 or greater";
                 goto loaderr;
             }
@@ -1221,6 +1221,8 @@ void configGetCommand(client *c) {
     /* Numerical values */
     config_get_numerical_field("maxmemory",server.maxmemory);
     config_get_numerical_field("maxmemory-samples",server.maxmemory_samples);
+    config_get_numerical_field("lfu-log-factor",server.lfu_log_factor);
+    config_get_numerical_field("lfu-decay-time",server.lfu_decay_time);
     config_get_numerical_field("timeout",server.maxidletime);
     config_get_numerical_field("active-defrag-threshold-lower",server.active_defrag_threshold_lower);
     config_get_numerical_field("active-defrag-threshold-upper",server.active_defrag_threshold_upper);
@@ -1992,6 +1994,8 @@ int rewriteConfig(char *path) {
     rewriteConfigBytesOption(state,"maxmemory",server.maxmemory,CONFIG_DEFAULT_MAXMEMORY);
     rewriteConfigEnumOption(state,"maxmemory-policy",server.maxmemory_policy,maxmemory_policy_enum,CONFIG_DEFAULT_MAXMEMORY_POLICY);
     rewriteConfigNumericalOption(state,"maxmemory-samples",server.maxmemory_samples,CONFIG_DEFAULT_MAXMEMORY_SAMPLES);
+    rewriteConfigNumericalOption(state,"lfu-log-factor",server.lfu_log_factor,CONFIG_DEFAULT_LFU_LOG_FACTOR);
+    rewriteConfigNumericalOption(state,"lfu-decay-time",server.lfu_decay_time,CONFIG_DEFAULT_LFU_DECAY_TIME);
     rewriteConfigNumericalOption(state,"active-defrag-threshold-lower",server.active_defrag_threshold_lower,CONFIG_DEFAULT_DEFRAG_THRESHOLD_LOWER);
     rewriteConfigNumericalOption(state,"active-defrag-threshold-upper",server.active_defrag_threshold_upper,CONFIG_DEFAULT_DEFRAG_THRESHOLD_UPPER);
     rewriteConfigBytesOption(state,"active-defrag-ignore-bytes",server.active_defrag_ignore_bytes,CONFIG_DEFAULT_DEFRAG_IGNORE_BYTES);

src/db.c

@@ -38,6 +38,15 @@
  * C-level DB API
  *----------------------------------------------------------------------------*/
 
+/* Update LFU when an object is accessed.
+ * Firstly, decrement the counter if the decrement time is reached.
+ * Then logarithmically increment the counter, and update the access time. */
+void updateLFU(robj *val) {
+    unsigned long counter = LFUDecrAndReturn(val);
+    counter = LFULogIncr(counter);
+    val->lru = (LFUGetTimeInMinutes()<<8) | counter;
+}
+
 /* Low level key lookup API, not actually called directly from commands
  * implementations that should instead rely on lookupKeyRead(),
  * lookupKeyWrite() and lookupKeyReadWithFlags(). */
@@ -54,9 +63,7 @@ robj *lookupKey(redisDb *db, robj *key, int flags) {
             !(flags & LOOKUP_NOTOUCH))
         {
             if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
-                unsigned long ldt = val->lru >> 8;
-                unsigned long counter = LFULogIncr(val->lru & 255);
-                val->lru = (ldt << 8) | counter;
+                updateLFU(val);
             } else {
                 val->lru = LRU_CLOCK();
             }
@@ -180,6 +187,9 @@ void dbOverwrite(redisDb *db, robj *key, robj *val) {
         int saved_lru = old->lru;
         dictReplace(db->dict, key->ptr, val);
         val->lru = saved_lru;
+        /* LFU should be not only copied but also updated
+         * when a key is overwritten. */
+        updateLFU(val);
     } else {
         dictReplace(db->dict, key->ptr, val);
     }

src/evict.c

@@ -60,8 +60,6 @@ struct evictionPoolEntry {
 
 static struct evictionPoolEntry *EvictionPoolLRU;
 
-unsigned long LFUDecrAndReturn(robj *o);
-
 /* ----------------------------------------------------------------------------
  * Implementation of eviction, aging and LRU
  * --------------------------------------------------------------------------*/
@@ -302,8 +300,8 @@ unsigned long LFUGetTimeInMinutes(void) {
     return (server.unixtime/60) & 65535;
 }
 
-/* Given an object last decrement time, compute the minimum number of minutes
- * that elapsed since the last decrement. Handle overflow (ldt greater than
+/* Given an object last access time, compute the minimum number of minutes
+ * that elapsed since the last access. Handle overflow (ldt greater than
  * the current 16 bits minutes time) considering the time as wrapping
  * exactly once. */
 unsigned long LFUTimeElapsed(unsigned long ldt) {
@@ -324,25 +322,22 @@ uint8_t LFULogIncr(uint8_t counter) {
     return counter;
 }
 
-/* If the object decrement time is reached, decrement the LFU counter and
- * update the decrement time field. Return the object frequency counter.
+/* If the object decrement time is reached decrement the LFU counter but
+ * do not update LFU fields of the object, we update the access time
+ * and counter in an explicit way when the object is really accessed.
+ * And we will times halve the counter according to the times of
+ * elapsed time than server.lfu_decay_time.
+ * Return the object frequency counter.
  *
  * This function is used in order to scan the dataset for the best object
  * to fit: as we check for the candidate, we incrementally decrement the
  * counter of the scanned objects if needed. */
-#define LFU_DECR_INTERVAL 1
 unsigned long LFUDecrAndReturn(robj *o) {
     unsigned long ldt = o->lru >> 8;
     unsigned long counter = o->lru & 255;
-    if (LFUTimeElapsed(ldt) >= server.lfu_decay_time && counter) {
-        if (counter > LFU_INIT_VAL*2) {
-            counter /= 2;
-            if (counter < LFU_INIT_VAL*2) counter = LFU_INIT_VAL*2;
-        } else {
-            counter--;
-        }
-        o->lru = (LFUGetTimeInMinutes()<<8) | counter;
-    }
+    unsigned long num_periods = server.lfu_decay_time ? LFUTimeElapsed(ldt) / server.lfu_decay_time : 0;
+    if (num_periods)
+        counter = (num_periods > counter) ? 0 : counter - num_periods;
     return counter;
 }
 

src/object.c

@@ -1050,10 +1050,14 @@ void objectCommand(client *c) {
         if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nullbulk))
                 == NULL) return;
         if (!(server.maxmemory_policy & MAXMEMORY_FLAG_LFU)) {
-            addReplyError(c,"A non-LFU maxmemory policy is selected, access frequency not tracked. Please note that when switching between policies at runtime LRU and LFU data will take some time to adjust.");
+            addReplyError(c,"An LFU maxmemory policy is not selected, access frequency not tracked. Please note that when switching between policies at runtime LRU and LFU data will take some time to adjust.");
             return;
         }
-        addReplyLongLong(c,o->lru&255);
+        /* LFUDecrAndReturn should be called
+         * in case of the key has not been accessed for a long time,
+         * because we update the access time only
+         * when the key is read or overwritten. */
+        addReplyLongLong(c,LFUDecrAndReturn(o));
     } else {
         addReplyErrorFormat(c, "Unknown subcommand or wrong number of arguments for '%s'. Try OBJECT help",
             (char *)c->argv[1]->ptr);

src/redis-cli.c

@@ -107,6 +107,7 @@ static struct config {
     char *pattern;
     char *rdb_filename;
     int bigkeys;
+    int hotkeys;
     int stdinarg; /* get last arg from stdin. (-x option) */
     char *auth;
     int output; /* output mode, see OUTPUT_* defines */
@@ -1129,6 +1130,8 @@ static int parseOptions(int argc, char **argv) {
             config.pipe_timeout = atoi(argv[++i]);
         } else if (!strcmp(argv[i],"--bigkeys")) {
             config.bigkeys = 1;
+        } else if (!strcmp(argv[i],"--hotkeys")) {
+            config.hotkeys = 1;
         } else if (!strcmp(argv[i],"--eval") && !lastarg) {
             config.eval = argv[++i];
         } else if (!strcmp(argv[i],"--ldb")) {
@@ -1229,6 +1232,8 @@ static void usage(void) {
 "                     no reply is received within <n> seconds.\n"
 "                     Default timeout: %d. Use 0 to wait forever.\n"
 "  --bigkeys          Sample Redis keys looking for big keys.\n"
+"  --hotkeys          Sample Redis keys looking for hot keys.\n"
+"                     only works when maxmemory-policy is *lfu.\n"
 "  --scan             List all keys using the SCAN command.\n"
 "  --pattern <pat>    Useful with --scan to specify a SCAN pattern.\n"
 "  --intrinsic-latency <sec> Run a test to measure intrinsic system latency.\n"
@@ -2343,6 +2348,129 @@ static void findBigKeys(void) {
     exit(0);
 }
 
+static void getKeyFreqs(redisReply *keys, unsigned long long *freqs) {
+    redisReply *reply;
+    unsigned int i;
+
+    /* Pipeline OBJECT freq commands */
+    for(i=0;i<keys->elements;i++) {
+        redisAppendCommand(context, "OBJECT freq %s", keys->element[i]->str);
+    }
+
+    /* Retrieve freqs */
+    for(i=0;i<keys->elements;i++) {
+        if(redisGetReply(context, (void**)&reply)!=REDIS_OK) {
+            fprintf(stderr, "Error getting freq for key '%s' (%d: %s)\n",
+                keys->element[i]->str, context->err, context->errstr);
+            exit(1);
+        } else if(reply->type != REDIS_REPLY_INTEGER) {
+            if(reply->type == REDIS_REPLY_ERROR) {
+                fprintf(stderr, "Error: %s\n", reply->str);
+                exit(1);
+            } else {
+                fprintf(stderr, "Warning: OBJECT freq on '%s' failed (may have been deleted)\n", keys->element[i]->str);
+                freqs[i] = 0;
+            }
+        } else {
+            freqs[i] = reply->integer;
+        }
+        freeReplyObject(reply);
+    }
+}
+
+#define HOTKEYS_SAMPLE 16
+static void findHotKeys(void) {
+    redisReply *keys, *reply;
+    unsigned long long counters[HOTKEYS_SAMPLE] = {0};
+    sds hotkeys[HOTKEYS_SAMPLE] = {NULL};
+    unsigned long long sampled = 0, total_keys, *freqs = NULL, it = 0;
+    unsigned int arrsize = 0, i, k;
+    double pct;
+
+    /* Total keys pre scanning */
+    total_keys = getDbSize();
+
+    /* Status message */
+    printf("\n# Scanning the entire keyspace to find hot keys as well as\n");
+    printf("# average sizes per key type.  You can use -i 0.1 to sleep 0.1 sec\n");
+    printf("# per 100 SCAN commands (not usually needed).\n\n");
+
+    /* SCAN loop */
+    do {
+        /* Calculate approximate percentage completion */
+        pct = 100 * (double)sampled/total_keys;
+
+        /* Grab some keys and point to the keys array */
+        reply = sendScan(&it);
+        keys  = reply->element[1];
+
+        /* Reallocate our freqs array if we need to */
+        if(keys->elements > arrsize) {
+            freqs = zrealloc(freqs, sizeof(unsigned long long)*keys->elements);
+
+            if(!freqs) {
+                fprintf(stderr, "Failed to allocate storage for keys!\n");
+                exit(1);
+            }
+
+            arrsize = keys->elements;
+        }
+
+        getKeyFreqs(keys, freqs);
+
+        /* Now update our stats */
+        for(i=0;i<keys->elements;i++) {
+            sampled++;
+            /* Update overall progress */
+            if(sampled % 1000000 == 0) {
+                printf("[%05.2f%%] Sampled %llu keys so far\n", pct, sampled);
+            }
+
+            /* Use eviction pool here */
+            k = 0;
+            while (k < HOTKEYS_SAMPLE && freqs[i] > counters[k]) k++;
+            if (k == 0) continue;
+            k--;
+            if (k == 0 || counters[k] == 0) {
+                sdsfree(hotkeys[k]);
+            } else {
+                sdsfree(hotkeys[0]);
+                memmove(counters,counters+1,sizeof(counters[0])*k);
+                memmove(hotkeys,hotkeys+1,sizeof(hotkeys[0])*k);
+            }
+            counters[k] = freqs[i];
+            hotkeys[k] = sdsnew(keys->element[i]->str);
+            printf(
+               "[%05.2f%%] Hot key '%s' found so far with counter %llu\n",
+               pct, keys->element[i]->str, freqs[i]);
+        }
+
+        /* Sleep if we've been directed to do so */
+        if(sampled && (sampled %100) == 0 && config.interval) {
+            usleep(config.interval);
+        }
+
+        freeReplyObject(reply);
+    } while(it != 0);
+
+    if (freqs) zfree(freqs);
+
+    /* We're done */
+    printf("\n-------- summary -------\n\n");
+
+    printf("Sampled %llu keys in the keyspace!\n", sampled);
+
+    for (i=1; i<= HOTKEYS_SAMPLE; i++) {
+        k = HOTKEYS_SAMPLE - i;
+        if(counters[k]>0) {
+            printf("hot key found with counter: %llu\tkeyname: %s\n", counters[k], hotkeys[k]);
+            sdsfree(hotkeys[k]);
+        }
+    }
+
+    exit(0);
+}
+
 /*------------------------------------------------------------------------------
  * Stats mode
  *--------------------------------------------------------------------------- */
@@ -2720,6 +2848,7 @@ int main(int argc, char **argv) {
     config.pipe_mode = 0;
     config.pipe_timeout = REDIS_CLI_DEFAULT_PIPE_TIMEOUT;
     config.bigkeys = 0;
+    config.hotkeys = 0;
     config.stdinarg = 0;
     config.auth = NULL;
     config.eval = NULL;
@@ -2780,6 +2909,12 @@ int main(int argc, char **argv) {
         findBigKeys();
     }
 
+    /* Find hot keys */
+    if (config.hotkeys) {
+        if (cliConnect(0) == REDIS_ERR) exit(1);
+        findHotKeys();
+    }
+
     /* Stat mode */
     if (config.stat_mode) {
         if (cliConnect(0) == REDIS_ERR) exit(1);

src/server.h

@@ -586,7 +586,7 @@ typedef struct redisObject {
     unsigned encoding:4;
     unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
                             * LFU data (least significant 8 bits frequency
-                            * and most significant 16 bits decreas time). */
+                            * and most significant 16 bits access time). */
     int refcount;
     void *ptr;
 } robj;
@@ -1118,8 +1118,8 @@ struct redisServer {
     unsigned long long maxmemory;   /* Max number of memory bytes to use */
     int maxmemory_policy;           /* Policy for key eviction */
     int maxmemory_samples;          /* Pricision of random sampling */
-    unsigned int lfu_log_factor;    /* LFU logarithmic counter factor. */
-    unsigned int lfu_decay_time;    /* LFU counter decay factor. */
+    int lfu_log_factor;             /* LFU logarithmic counter factor. */
+    int lfu_decay_time;             /* LFU counter decay factor. */
     /* Blocked clients */
     unsigned int bpop_blocked_clients; /* Number of clients blocked by lists */
     list *unblocked_clients; /* list of clients to unblock before next loop */
@@ -1802,6 +1802,7 @@ void evictionPoolAlloc(void);
 #define LFU_INIT_VAL 5
 unsigned long LFUGetTimeInMinutes(void);
 uint8_t LFULogIncr(uint8_t value);
+unsigned long LFUDecrAndReturn(robj *o);
 
 /* Keys hashing / comparison functions for dict.c hash tables. */
 uint64_t dictSdsHash(const void *key);