redis 精讲系列介绍八 - 淘汰策略

typedef struct redisObject {    unsigned type:4;    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 access time). */    int refcount;    void *ptr;} robj;

/* Low level key lookup API, not actually called directly from commands * implementations that should instead rely on lookupKeyRead(), * lookupKeyWrite() and lookupKeyReadWithFlags(). */robj *lookupKey(redisDb *db, robj *key, int flags) {    dictEntry *de = dictFind(db->dict,key->ptr);    if (de) {        robj *val = dictGetVal(de);
        /* Update the access time for the ageing algorithm.         * Don't do it if we have a saving child, as this will trigger         * a copy on write madness. */        if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH)){            if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {                // 这个是后面一节的内容                updateLFU(val);            } else {                //  对于这个分支，访问时就会去更新 lru 值                val->lru = LRU_CLOCK();            }        }        return val;    } else {        return NULL;    }}/* This function is used to obtain the current LRU clock. * If the current resolution is lower than the frequency we refresh the * LRU clock (as it should be in production servers) we return the * precomputed value, otherwise we need to resort to a system call. */unsigned int LRU_CLOCK(void) {    unsigned int lruclock;    if (1000/server.hz <= LRU_CLOCK_RESOLUTION) {        // 如果服务器的频率server.hz大于 1 时就是用系统预设的 lruclock        lruclock = server.lruclock;    } else {        lruclock = getLRUClock();    }    return lruclock;}/* Return the LRU clock, based on the clock resolution. This is a time * in a reduced-bits format that can be used to set and check the * object->lru field of redisObject structures. */unsigned int getLRUClock(void) {    return (mstime()/LRU_CLOCK_RESOLUTION) & LRU_CLOCK_MAX;}

/* If this function gets called we already read a whole * command, arguments are in the client argv/argc fields. * processCommand() execute the command or prepare the * server for a bulk read from the client. * * If C_OK is returned the client is still alive and valid and * other operations can be performed by the caller. Otherwise * if C_ERR is returned the client was destroyed (i.e. after QUIT). */int processCommand(client *c) {    moduleCallCommandFilters(c);
    
    /* Handle the maxmemory directive.     *     * Note that we do not want to reclaim memory if we are here re-entering     * the event loop since there is a busy Lua script running in timeout     * condition, to avoid mixing the propagation of scripts with the     * propagation of DELs due to eviction. */    if (server.maxmemory && !server.lua_timedout) {        int out_of_memory = freeMemoryIfNeededAndSafe() == C_ERR;        /* freeMemoryIfNeeded may flush slave output buffers. This may result         * into a slave, that may be the active client, to be freed. */        if (server.current_client == NULL) return C_ERR;
        /* It was impossible to free enough memory, and the command the client         * is trying to execute is denied during OOM conditions or the client         * is in MULTI/EXEC context? Error. */        if (out_of_memory &&            (c->cmd->flags & CMD_DENYOOM ||             (c->flags & CLIENT_MULTI &&              c->cmd->proc != execCommand &&              c->cmd->proc != discardCommand)))        {            flagTransaction(c);            addReply(c, shared.oomerr);            return C_OK;        }    }}

/* This is a wrapper for freeMemoryIfNeeded() that only really calls the * function if right now there are the conditions to do so safely: * * - There must be no script in timeout condition. * - Nor we are loading data right now. * */int freeMemoryIfNeededAndSafe(void) {    if (server.lua_timedout || server.loading) return C_OK;    return freeMemoryIfNeeded();}/* This function is periodically called to see if there is memory to free * according to the current "maxmemory" settings. In case we are over the * memory limit, the function will try to free some memory to return back * under the limit. * * The function returns C_OK if we are under the memory limit or if we * were over the limit, but the attempt to free memory was successful. * Otehrwise if we are over the memory limit, but not enough memory * was freed to return back under the limit, the function returns C_ERR. */int freeMemoryIfNeeded(void) {    int keys_freed = 0;    /* By default replicas should ignore maxmemory     * and just be masters exact copies. */    if (server.masterhost && server.repl_slave_ignore_maxmemory) return C_OK;
    size_t mem_reported, mem_tofree, mem_freed;    mstime_t latency, eviction_latency;    long long delta;    int slaves = listLength(server.slaves);
    /* When clients are paused the dataset should be static not just from the     * POV of clients not being able to write, but also from the POV of     * expires and evictions of keys not being performed. */    if (clientsArePaused()) return C_OK;    if (getMaxmemoryState(&mem_reported,NULL,&mem_tofree,NULL) == C_OK)        return C_OK;
    mem_freed = 0;
    if (server.maxmemory_policy == MAXMEMORY_NO_EVICTION)        goto cant_free; /* We need to free memory, but policy forbids. */
    latencyStartMonitor(latency);    while (mem_freed < mem_tofree) {        int j, k, i;        static unsigned int next_db = 0;        sds bestkey = NULL;        int bestdbid;        redisDb *db;        dict *dict;        dictEntry *de;
        if (server.maxmemory_policy & (MAXMEMORY_FLAG_LRU|MAXMEMORY_FLAG_LFU) ||            server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL)        {            struct evictionPoolEntry *pool = EvictionPoolLRU;
            while(bestkey == NULL) {                unsigned long total_keys = 0, keys;
                /* We don't want to make local-db choices when expiring keys,                 * so to start populate the eviction pool sampling keys from                 * every DB. */                for (i = 0; i < server.dbnum; i++) {                    db = server.db+i;                    dict = (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) ?                            db->dict : db->expires;                    if ((keys = dictSize(dict)) != 0) {                        evictionPoolPopulate(i, dict, db->dict, pool);                        total_keys += keys;                    }                }                if (!total_keys) break; /* No keys to evict. */
                /* Go backward from best to worst element to evict. */                for (k = EVPOOL_SIZE-1; k >= 0; k--) {                    if (pool[k].key == NULL) continue;                    bestdbid = pool[k].dbid;
                    if (server.maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) {                        de = dictFind(server.db[pool[k].dbid].dict,                            pool[k].key);                    } else {                        de = dictFind(server.db[pool[k].dbid].expires,                            pool[k].key);                    }
                    /* Remove the entry from the pool. */                    if (pool[k].key != pool[k].cached)                        sdsfree(pool[k].key);                    pool[k].key = NULL;                    pool[k].idle = 0;
                    /* If the key exists, is our pick. Otherwise it is                     * a ghost and we need to try the next element. */                    if (de) {                        bestkey = dictGetKey(de);                        break;                    } else {                        /* Ghost... Iterate again. */                    }                }            }        }
        /* volatile-random and allkeys-random policy */        else if (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM ||                 server.maxmemory_policy == MAXMEMORY_VOLATILE_RANDOM)        {            /* When evicting a random key, we try to evict a key for             * each DB, so we use the static 'next_db' variable to             * incrementally visit all DBs. */            for (i = 0; i < server.dbnum; i++) {                j = (++next_db) % server.dbnum;                db = server.db+j;                dict = (server.maxmemory_policy == MAXMEMORY_ALLKEYS_RANDOM) ?                        db->dict : db->expires;                if (dictSize(dict) != 0) {                    de = dictGetRandomKey(dict);                    bestkey = dictGetKey(de);                    bestdbid = j;                    break;                }            }        }
        /* Finally remove the selected key. */        if (bestkey) {            db = server.db+bestdbid;            robj *keyobj = createStringObject(bestkey,sdslen(bestkey));            propagateExpire(db,keyobj,server.lazyfree_lazy_eviction);            /* We compute the amount of memory freed by db*Delete() alone.             * It is possible that actually the memory needed to propagate             * the DEL in AOF and replication link is greater than the one             * we are freeing removing the key, but we can't account for             * that otherwise we would never exit the loop.             *             * AOF and Output buffer memory will be freed eventually so             * we only care about memory used by the key space. */            delta = (long long) zmalloc_used_memory();            latencyStartMonitor(eviction_latency);            if (server.lazyfree_lazy_eviction)                dbAsyncDelete(db,keyobj);            else                dbSyncDelete(db,keyobj);            latencyEndMonitor(eviction_latency);            latencyAddSampleIfNeeded("eviction-del",eviction_latency);            latencyRemoveNestedEvent(latency,eviction_latency);            delta -= (long long) zmalloc_used_memory();            mem_freed += delta;            server.stat_evictedkeys++;            notifyKeyspaceEvent(NOTIFY_EVICTED, "evicted",                keyobj, db->id);            decrRefCount(keyobj);            keys_freed++;
            /* When the memory to free starts to be big enough, we may             * start spending so much time here that is impossible to             * deliver data to the slaves fast enough, so we force the             * transmission here inside the loop. */            if (slaves) flushSlavesOutputBuffers();
            /* Normally our stop condition is the ability to release             * a fixed, pre-computed amount of memory. However when we             * are deleting objects in another thread, it's better to             * check, from time to time, if we already reached our target             * memory, since the "mem_freed" amount is computed only             * across the dbAsyncDelete() call, while the thread can             * release the memory all the time. */            if (server.lazyfree_lazy_eviction && !(keys_freed % 16)) {                if (getMaxmemoryState(NULL,NULL,NULL,NULL) == C_OK) {                    /* Let's satisfy our stop condition. */                    mem_freed = mem_tofree;                }            }        } else {            latencyEndMonitor(latency);            latencyAddSampleIfNeeded("eviction-cycle",latency);            goto cant_free; /* nothing to free... */        }    }    latencyEndMonitor(latency);    latencyAddSampleIfNeeded("eviction-cycle",latency);    return C_OK;
cant_free:    /* We are here if we are not able to reclaim memory. There is only one     * last thing we can try: check if the lazyfree thread has jobs in queue     * and wait... */    while(bioPendingJobsOfType(BIO_LAZY_FREE)) {        if (((mem_reported - zmalloc_used_memory()) + mem_freed) >= mem_tofree)            break;        usleep(1000);    }    return C_ERR;}

void evictionPoolPopulate(int dbid, dict *sampledict, dict *keydict, struct evictionPoolEntry *pool) {    int j, k, count;    dictEntry *samples[server.maxmemory_samples];
    count = dictGetSomeKeys(sampledict,samples,server.maxmemory_samples);    for (j = 0; j < count; j++) {        unsigned long long idle;        sds key;        robj *o;        dictEntry *de;
        de = samples[j];        key = dictGetKey(de);
        /* If the dictionary we are sampling from is not the main         * dictionary (but the expires one) we need to lookup the key         * again in the key dictionary to obtain the value object. */        if (server.maxmemory_policy != MAXMEMORY_VOLATILE_TTL) {            if (sampledict != keydict) de = dictFind(keydict, key);            o = dictGetVal(de);        }
        /* Calculate the idle time according to the policy. This is called         * idle just because the code initially handled LRU, but is in fact         * just a score where an higher score means better candidate. */        if (server.maxmemory_policy & MAXMEMORY_FLAG_LRU) {            idle = estimateObjectIdleTime(o);        } else if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {            /* When we use an LRU policy, we sort the keys by idle time             * so that we expire keys starting from greater idle time.             * However when the policy is an LFU one, we have a frequency             * estimation, and we want to evict keys with lower frequency             * first. So inside the pool we put objects using the inverted             * frequency subtracting the actual frequency to the maximum             * frequency of 255. */            idle = 255-LFUDecrAndReturn(o);        } else if (server.maxmemory_policy == MAXMEMORY_VOLATILE_TTL) {            /* In this case the sooner the expire the better. */            idle = ULLONG_MAX - (long)dictGetVal(de);        } else {            serverPanic("Unknown eviction policy in evictionPoolPopulate()");        }
        /* Insert the element inside the pool.         * First, find the first empty bucket or the first populated         * bucket that has an idle time smaller than our idle time. */        k = 0;        while (k < EVPOOL_SIZE &&               pool[k].key &&               pool[k].idle < idle) k++;        if (k == 0 && pool[EVPOOL_SIZE-1].key != NULL) {            /* Can't insert if the element is < the worst element we have             * and there are no empty buckets. */            continue;        } else if (k < EVPOOL_SIZE && pool[k].key == NULL) {            /* Inserting into empty position. No setup needed before insert. */        } else {            /* Inserting in the middle. Now k points to the first element             * greater than the element to insert.  */            if (pool[EVPOOL_SIZE-1].key == NULL) {                /* Free space on the right? Insert at k shifting                 * all the elements from k to end to the right. */
                /* Save SDS before overwriting. */                sds cached = pool[EVPOOL_SIZE-1].cached;                memmove(pool+k+1,pool+k,                    sizeof(pool[0])*(EVPOOL_SIZE-k-1));                pool[k].cached = cached;            } else {                /* No free space on right? Insert at k-1 */                k--;                /* Shift all elements on the left of k (included) to the                 * left, so we discard the element with smaller idle time. */                sds cached = pool[0].cached; /* Save SDS before overwriting. */                if (pool[0].key != pool[0].cached) sdsfree(pool[0].key);                memmove(pool,pool+1,sizeof(pool[0])*k);                pool[k].cached = cached;            }        }
        /* Try to reuse the cached SDS string allocated in the pool entry,         * because allocating and deallocating this object is costly         * (according to the profiler, not my fantasy. Remember:         * premature optimizbla bla bla bla. */        int klen = sdslen(key);        if (klen > EVPOOL_CACHED_SDS_SIZE) {            pool[k].key = sdsdup(key);        } else {            memcpy(pool[k].cached,key,klen+1);            sdssetlen(pool[k].cached,klen);            pool[k].key = pool[k].cached;        }        pool[k].idle = idle;        pool[k].dbid = dbid;    }}

/* Given an object returns the min number of milliseconds the object was never * requested, using an approximated LRU algorithm. */unsigned long long estimateObjectIdleTime(robj *o) {    unsigned long long lruclock = LRU_CLOCK();    if (lruclock >= o->lru) {        return (lruclock - o->lru) * LRU_CLOCK_RESOLUTION;    } else {        return (lruclock + (LRU_CLOCK_MAX - o->lru)) *                    LRU_CLOCK_RESOLUTION;    }}

~~~~~A~~~~~A~~~~~A~~~~A~~~~~A~~~~~A~~|~~B~~B~~B~~B~~B~~B~~B~~B~~B~~B~~B~~B~|~~~~~~~~~~C~~~~~~~~~C~~~~~~~~~C~~~~~~|~~~~~D~~~~~~~~~~D~~~~~~~~~D~~~~~~~~~D|

        16 bits      8 bits  +----------------+--------+  + Last decr time | LOG_C  |  +----------------+--------+

lfu-log-factor 10lfu-decay-time 1

/* Low level key lookup API, not actually called directly from commands * implementations that should instead rely on lookupKeyRead(), * lookupKeyWrite() and lookupKeyReadWithFlags(). */robj *lookupKey(redisDb *db, robj *key, int flags) {    dictEntry *de = dictFind(db->dict,key->ptr);    if (de) {        robj *val = dictGetVal(de);
        /* Update the access time for the ageing algorithm.         * Don't do it if we have a saving child, as this will trigger         * a copy on write madness. */        if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH)){            if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {                // 当淘汰策略是 LFU 时，就会调用这个updateLFU                updateLFU(val);            } else {                val->lru = LRU_CLOCK();            }        }        return val;    } else {        return NULL;    }}

/* 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;}

/* 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. */unsigned long LFUDecrAndReturn(robj *o) {    // 右移 8 位，拿到上次衰减时间    unsigned long ldt = o->lru >> 8;    // 对 255 做与操作，拿到 counter 值    unsigned long counter = o->lru & 255;    // 根据lfu_decay_time来算出过了多少个衰减周期    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;}

/* Logarithmically increment a counter. The greater is the current counter value * the less likely is that it gets really implemented. Saturate it at 255. */uint8_t LFULogIncr(uint8_t counter) {    // 最大值就是 255，到顶了就不加了    if (counter == 255) return 255;    // 生成个随机小数    double r = (double)rand()/RAND_MAX;    // 减去个基础值，LFU_INIT_VAL = 5，防止刚进来就被逐出    double baseval = counter - LFU_INIT_VAL;    // 如果是小于 0，    if (baseval < 0) baseval = 0;    // 如果 baseval 是 0，那么 p 就是 1了，后面 counter 直接加一，如果不是的话，得看系统参数lfu_log_factor，这个越大，除出来的 p 越小，那么 counter++的可能性也越小，这样子就把前面的疑问给解决了，不是直接+1 的    double p = 1.0/(baseval*server.lfu_log_factor+1);    if (r < p) counter++;    return counter;}

+--------+------------+------------+------------+------------+------------+| factor | 100 hits   | 1000 hits  | 100K hits  | 1M hits    | 10M hits   |+--------+------------+------------+------------+------------+------------+| 0      | 104        | 255        | 255        | 255        | 255        |+--------+------------+------------+------------+------------+------------+| 1      | 18         | 49         | 255        | 255        | 255        |+--------+------------+------------+------------+------------+------------+| 10     | 10         | 18         | 142        | 255        | 255        |+--------+------------+------------+------------+------------+------------+| 100    | 8          | 11         | 49         | 143        | 255        |+--------+------------+------------+------------+------------+------------+