Redis db 的实现 · Analyze

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

typedef struct redisDb {
    // 数据库键空间
    dict *dict;                 /* The keyspace for this DB */
    // 键的过期时间
    dict *expires;              /* Timeout of keys with a timeout set */
    // 正处于阻塞状态的键
    dict *blocking_keys;        /* Keys with clients waiting for data (BLPOP)*/
    // 可以解除阻塞的键
    dict *ready_keys;           /* Blocked keys that received a PUSH */
    // 正在被 WATCH 命令监视的键
    dict *watched_keys;         /* WATCHED keys for MULTI/EXEC CAS */
    // 数据库 ID 
    int id;                     /* Database ID */
    // 数据库的键的平均 TTL，统计信息
    long long avg_ttl;          /* Average TTL, just for stats */
    // 仍存活的过期键游标
    unsigned long expires_cursor; /* Cursor of the active expire cycle. */
    // 尝试进行碎片整理的键名称列表
    list *defrag_later;         /* List of key names to attempt to defrag one by one, gradually. */
} redisDb;

1
2
3
4
5
6
7
8

struct redisServer{
    ...
    // 当前使用的数据库
    redisDb *db;
    // 创建的数据库数量
    int dbnum;
    ...
}

1
2
3
4
5
6
7

int selectDb(client *c, int id) {
    if (id < 0 || id >= server.dbnum)
        return C_ERR;
    // 设置当前 client 使用的数据库
    c->db = &server.db[id];
    return C_OK;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

robj *lookupKeyRead(redisDb *db, robj *key) {
    robj *val;
    expireIfNeeded(db,key);
    val = lookupKey(db,key);
    ...
    return val;
}

int expireIfNeeded(redisDb *db, robj *key) {
    // 如果键没有过期返回 0 
    if (!keyIsExpired(db,key)) return 0;
    // 如果当前节点不是主节点返回 1
    // 返回 1 并不是一个确切的信息，但作者觉得要比返回 0 更准确一些
    if (server.masterhost != NULL) return 1;
    // 将过期键的数量标记值加 1
    server.stat_expiredkeys++;
    propagateExpire(db,key,server.lazyfree_lazy_expire);
    notifyKeyspaceEvent(NOTIFY_EXPIRED,
        "expired",key,db->id);
    // 删除键
    return server.lazyfree_lazy_expire ? dbAsyncDelete(db,key) :
                                         dbSyncDelete(db,key);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

void databasesCron(void) {
    // 如果服务器不是从服务器，那么执行过期键清除
    if (server.active_expire_enabled) {
        if (server.masterhost == NULL) {
            // 默认慢循环，这个模式会尽量多清除过期键
            activeExpireCycle(ACTIVE_EXPIRE_CYCLE_SLOW);
        } else {
            expireSlaveKeys();
        }
    }
    ...
}

1
2
3
4
5
6

typedef struct redisObject {
	...
    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). */
} robj;

1
2
3

#define LRU_BITS 24
#define LRU_CLOCK_MAX ((1<<LRU_BITS)-1) /* Max value of obj->lru */
#define LRU_CLOCK_RESOLUTION 1000 /* LRU clock resolution in ms */

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

#define RAND_MAX 0x7fff
#define LFU_INIT_VAL 5
uint8_t LFULogIncr(uint8_t counter) {
    if (counter == 255) return 255;
    double r = (double)rand()/RAND_MAX;
    double baseval = counter - LFU_INIT_VAL;
    if (baseval < 0) baseval = 0;
    double p = 1.0/(baseval*server.lfu_log_factor+1);
    if (r < p) counter++;
    return counter;
}

1
2
3
4
5
6
7
8

unsigned long LFUDecrAndReturn(robj *o) {
    unsigned long ldt = o->lru >> 8;
    unsigned long counter = o->lru & 255;
    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;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

struct redisServer{
    ...
    // 当内存达到上限后是否执行 lazyfree
    int lazyfree_lazy_eviction;
    // 是否对过期键执行 lazyfree
    int lazyfree_lazy_expire;
    // 是否对 del 命令执行 lazyfree
    int lazyfree_lazy_server_del;
    ...
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

#define LAZYFREE_THRESHOLD 64
int dbAsyncDelete(redisDb *db, robj *key) {
    // 从 expires 字典中删除过期键
    if (dictSize(db->expires) > 0) dictDelete(db->expires,key->ptr);
    // 如果这个值的元素数量很少，同步释放对象
    // 从字典中解除绑定但不释放内存
    dictEntry *de = dictUnlink(db->dict,key->ptr);

    if (de) {
        // 获取值对象指针
        robj *val = dictGetVal(de);
        // 获取值对象所包含的元素个数
        size_t free_effort = lazyfreeGetFreeEffort(val);
        // 如果值对象较大
        if (free_effort > LAZYFREE_THRESHOLD && val->refcount == 1) {
            // 原子操作给 lazyfree_objects 加1，以备 info 命令查看后台进程有多少对象待删除
            atomicIncr(lazyfree_objects,1);
            // 把对象 val 丢到后台进程的任务队列中
            bioCreateBackgroundJob(BIO_LAZY_FREE,val,NULL,NULL);
            // 把字典节点的指针设置为 NULL
            dictSetVal(db->dict,de,NULL);
        }
    }
    if (de) {
        // 释放数据库字典中的键值对或只释放键
        dictFreeUnlinkedEntry(db->dict,de);
    } else {
        return 0;
    }
}

1
2
3
4
5
6
7

void emptyDbAsync(redisDb *db) {
    dict *oldht1 = db->dict, *oldht2 = db->expires;
    db->dict = dictCreate(&dbDictType,NULL);
    db->expires = dictCreate(&keyptrDictType,NULL);
    atomicIncr(lazyfree_objects,dictSize(oldht1));
    bioCreateBackgroundJob(BIO_LAZY_FREE,NULL,oldht1,oldht2);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

void *bioProcessBackgroundJobs(void *arg) {
    ...
    if (type == BIO_LAZY_FREE) {
            /* What we free changes depending on what arguments are set:
             * arg1 -> free the object at pointer.
             * arg2 & arg3 -> free two dictionaries (a Redis DB).
             * only arg3 -> free the skiplist. */
            if (job->arg1)
                lazyfreeFreeObjectFromBioThread(job->arg1);
            else if (job->arg2 && job->arg3)
                lazyfreeFreeDatabaseFromBioThread(job->arg2,job->arg3);
            else if (job->arg3)
                lazyfreeFreeSlotsMapFromBioThread(job->arg3);
    }
    ...
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

void lazyfreeFreeObjectFromBioThread(robj *o) {
    decrRefCount(o);
    atomicDecr(lazyfree_objects,1);
}
void lazyfreeFreeDatabaseFromBioThread(dict *ht1, dict *ht2) {
    size_t numkeys = dictSize(ht1);
    dictRelease(ht1);
    dictRelease(ht2);
    atomicDecr(lazyfree_objects,numkeys);
}
void lazyfreeFreeSlotsMapFromBioThread(rax *rt) {
    size_t len = rt->numele;
    raxFree(rt);
    atomicDecr(lazyfree_objects,len);
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

typedef struct blockingState {
    /* Generic fields. */
    // 阻塞时限
    mstime_t timeout;       /* Blocking operation timeout. If UNIX current time
                             * is > timeout then the operation timed out. */

    /* BLOCKED_LIST, BLOCKED_ZSET and BLOCKED_STREAM */
    // 造成阻塞的键
    dict *keys;             /* The keys we are waiting to terminate a blocking
                             * operation such as BLPOP or XREAD. Or NULL. */
    // 在被阻塞的键有新元素进入时，需要将这些新元素添加到哪里的目标键
    // 用于 BRPOPLPUSH 命令
    robj *target;           /* The key that should receive the element,
                             * for BRPOPLPUSH. */

    /* BLOCK_STREAM */
    size_t xread_count;     /* XREAD COUNT option. */
    robj *xread_group;      /* XREADGROUP group name. */
    robj *xread_consumer;   /* XREADGROUP consumer name. */
    mstime_t xread_retry_time, xread_retry_ttl;
    int xread_group_noack;

    /* BLOCKED_WAIT */
    int numreplicas;        /* Number of replicas we are waiting for ACK. */
    // 复制偏移量
    long long reploffset;   /* Replication offset to reach. */

    /* BLOCKED_MODULE */
    void *module_blocked_handle; /* RedisModuleBlockedClient structure.
                                    which is opaque for the Redis core, only
                                    handled in module.c. */
} blockingState;

1
2
3
4

typedef struct readyList {
    redisDb *db;
    robj *key;
} readyList;