Redis 数据结构

typedef struct redisObject {
     unsigned type:4;        // 对象类型：string, hash, list, set
     unsigned encoding:4;    // 编码方式
     unsigned lru:LRU_BITS;  // lru：24位，最近一次访问时间，单位秒，
                             // lfu: 高16位，最近一次访问时间；低8位，逻辑访问次数
     int refcount;           // 引用计数，计数为 0，对象无人引用，可以回收
     void *ptr;              // 数据指针，指向对象内容
 } robj;

#define OBJ_ENCODING_RAW 0     /* Raw representation */
 #define OBJ_ENCODING_INT 1     /* Encoded as integer */
 #define OBJ_ENCODING_HT 2      /* Encoded as hash table */
 #define OBJ_ENCODING_ZIPMAP 3  /* Encoded as zipmap */
 #define OBJ_ENCODING_LINKEDLIST 4 /* No longer used: old list encoding. */
 #define OBJ_ENCODING_ZIPLIST 5 /* Encoded as ziplist */
 #define OBJ_ENCODING_INTSET 6  /* Encoded as intset */
 #define OBJ_ENCODING_SKIPLIST 7  /* Encoded as skiplist */
 #define OBJ_ENCODING_EMBSTR 8  /* Embedded sds string encoding */
 #define OBJ_ENCODING_QUICKLIST 9 /* Encoded as linked list of ziplists */
 #define OBJ_ENCODING_STREAM 10 /* Encoded as a radix tree of listpacks */

#define OBJ_STRING 0    /* String object. */
 #define OBJ_LIST 1      /* List object. */
 #define OBJ_SET 2       /* Set object. */
 #define OBJ_ZSET 3      /* Sorted set object. */
 #define OBJ_HASH 4      /* Hash object. */

// 字典的实现，相当于 c++ 中类的封装
 typedef struct dict {
     dictType *type;     // dict 类型，封装成员函数
     void *privdata;     // 私有数据，连接的上下文
     dictht ht[2];       // 散列表，一个存储当前数据，另一个 rehash 时使用
     long rehashidx;     // rehash 的位置，-1 表示未进行 
     unsigned long iterators; /* number of iterators currently running */
 } dict;
 // 哈希表
 typedef struct dictht {
     dictEntry **table;      // entry 指针数组，保存 entry 的指针
     unsigned long size;     // 哈希表大小，2的n次幂
     unsigned long sizemask; // 哈希表掩码 size-1，hash 取余运算优化成位运算
     unsigned long used;     // 实际存储元素 entry 的个数
 } dictht;
 // 哈希节点
 typedef struct dictEntry {
     void *key; 
     union {
         void *val;
         uint64_t u64;
         int64_t s64;
         double d;
     } v;        
     struct dictEntry *next;
 } dictEntry;

// dict.c
 dictEntry *dictAddRaw(dict *d, void *key, dictEntry **existing)
 {
     ...
     /* Get the index of the new element, or -1 if
      * the element already exists. */
     if ((index = _dictKeyIndex(d, key, dictHashKey(d,key), existing)) == -1)
          return NULL;
     ...
 }
 static long _dictKeyIndex(dict *d, const void *key, uint64_t hash, dictEntry **existing)
 {
     ...
     /* Expand the hash table if needed */
     if (_dictExpandIfNeeded(d) == DICT_ERR)
         return -1;
     ...
 }
 /* Expand the hash table if needed */
 static int _dictExpandIfNeeded(dict *d)
 {
     /* Incremental rehashing already in progress. Return. */
     // 如果正在 rehash，则返回ok
     if (dictIsRehashing(d)) return DICT_OK;
     /* If the hash table is empty expand it to the initial size. */
     // 如果哈希表为空，则初始化哈希表为默认大小为4
     if (d->ht[0].size == 0) return dictExpand(d, DICT_HT_INITIAL_SIZE);
     /* If we reached the 1:1 ratio, and we are allowed to resize the hash
      * table (global setting) or we should avoid it but the ratio between
      * elements/buckets is over the "safe" threshold, we resize doubling
      * the number of buckets. */
     // 当负载因子达到1以上且没有 fork 进程 or 负载因子超过5，开始扩容
     if (d->ht[0].used >= d->ht[0].size &&
         (dict_can_resize ||
          d->ht[0].used/d->ht[0].size > dict_force_resize_ratio))
     {
         // 扩容翻倍
         return dictExpand(d, d->ht[0].used*2);
     }
     return DICT_OK;
 }
 /* Expand or create the hash table */
 // 重置哈希表的大小，扩容或缩容，重置大小为恰好包含 size 的二次幂
 int dictExpand(dict *d, unsigned long size)
 {
     // 若正在 rehash 或如果当前存储元素 entry 数量超过了要申请的 size 大小，报错
     if (dictIsRehashing(d) || d->ht[0].used > size)
         return DICT_ERR;
     // 申请新的哈希表，实际申请大小为第一个大于或等于size的2次幂
     dictht n;  
     unsigned long realsize = _dictNextPower(size); 
     /* Rehashing to the same table size is not useful. */
     if (realsize == d->ht[0].size) return DICT_ERR;
     /* Allocate the new hash table and initialize all pointers to NULL */
     n.size = realsize;       // 设置哈希表的 size 为实际申请大小
     n.sizemask = realsize-1; // 设置哈希表掩码 size-1   
     n.table = zcalloc(realsize*sizeof(dictEntry*));
     n.used = 0;
     /* Is this the first initialization? If so it's not really a rehashing
      * we just set the first hash table so that it can accept keys. */
     // 第一次初始化，不需要rehash，直接设置ht[0]指向新创建的哈希表即可
     if (d->ht[0].table == NULL) {
         d->ht[0] = n;
         return DICT_OK;
     }
     /* Prepare a second hash table for incremental rehashing */
     // 否则，rehash
     d->ht[1] = n;     // ht[1] 指向新创建的哈希表
     d->rehashidx = 0; // 标识 rehash 的开始  
     return DICT_OK;
 }

// t_hash.c
 int hashTypeDelete(robj *o, sds field) {
     ...
     if (dictDelete((dict*)o->ptr, field) == C_OK) {
         deleted = 1;
         // 删除成功后，检查是否需要缩容
         if (htNeedsResize(o->ptr)) dictResize(o->ptr);
     }
     ...
 }
 // server.c
 // 是否需要 rehash
 int htNeedsResize(dict *dict) {
     long long size, used;
     size = dictSlots(dict); // 哈希表的大小
     used = dictSize(dict);  // entry 的数量
     // 若哈希表的 size > 4 (哈希表的初始大小)且负载因子低于0.1
     return (size > DICT_HT_INITIAL_SIZE &&
             (used*100/size < HASHTABLE_MIN_FILL));
 }
 // dict.c
 int dictResize(dict *d)
 {   
     // 哈希表重置后的大小
     unsigned long minimal;
     // 如果正在 rehash or bgsave or bgwriteof ，返回错误
     if (!dict_can_resize || dictIsRehashing(d)) return DICT_ERR;
     // 获取 entry 个数
     minimal = d->ht[0].used;    
     // 若缩容后小于哈希表初始大小(4)，则重置为哈希表初始大小
     if (minimal < DICT_HT_INITIAL_SIZE)
         minimal = DICT_HT_INITIAL_SIZE;
     // 哈希表缩容
     return dictExpand(d, minimal);
 }

int dictRehash(dict *d, int n) { // n: rehash 的次数
     // 访问空槽位的最大次数，否则一直移动空桶，没有 rehash，造成长时间阻塞
     int empty_visits = n * 10; 
     // 正在 rehash，返回
     if (!dictIsRehashing(d)) return 0;  
     // 执行n次 reash 且 ht[0] 有元素
     while(n-- && d->ht[0].used != 0) {  
         dictEntry *de, *nextde;
         assert(d->ht[0].size > (unsigned long)d->rehashidx);
         // 循环寻找非空槽位
         while(d->ht[0].table[d->rehashidx] == NULL) {
             d->rehashidx++;                   // ht[0] 向后移动一个槽位
             if (--empty_visits == 0) return 1;  // 访问空桶次数达到最大，退出
         }
         // 寻找到非空槽位
         de = d->ht[0].table[d->rehashidx];
         // 将旧哈希表 ht[0] 该槽位中所有元素放入新哈希表 ht[1]
         while(de) {
             uint64_t h;
             nextde = de->next;
             // 获取该槽位的 entry 链表在新哈希表中的索引位置
             h = dictHashKey(d, de->key) & d->ht[1].sizemask;
             // 头插到新哈希表
             de->next = d->ht[1].table[h];
             d->ht[1].table[h] = de;
             d->ht[0].used--;
             d->ht[1].used++;
             de = nextde;
         }
         d->ht[0].table[d->rehashidx] = NULL;
         d->rehashidx++;
     }
     // rehash 是否完成
     if (d->ht[0].used == 0) {
         // rehash 完成
         zfree(d->ht[0].table); // 释放 ht[0] 的内存
         d->ht[0] = d->ht[1];   // 将 ht[1] 交给 ht[0]
         _dictReset(&d->ht[1]); // ht[1] 初始化为空  
         d->rehashidx = -1;     // 标识 rehash 结束
         return 0;
     }
     return 1;
 }

// rehash 1次
 static void _dictRehashStep(dict *d) {
     if (d->iterators == 0) dictRehash(d,1);
 }
 // rehash 1ms
 int dictRehashMilliseconds(dict *d, int ms) {
     long long start = timeInMilliseconds();
     int rehashes = 0;
     // 每次步长： rehash 100次
     while(dictRehash(d,100)) {
         rehashes += 100;
         // 超时，则退出
         if (timeInMilliseconds()-start > ms) break;
     }
     return rehashes;
 }

# 迭代数据库中的键
 scan cursor [MATCH pattern] [COUNT count] [TYPE type]

#define ZSKIPLIST_MAXLEVEL 32 // 跳表的层级
 #define ZSKIPLIST_P 0.25      // 每个节点增加层级的概率
 typedef struct zskiplistNode {
     sds ele;        // 节点存储的数据
     double score;   // 节点分数，排序使用
     struct zskiplistNode *backward; // 前一个节点指针
     struct zskiplistLevel {         // 多级索引数组
         struct zskiplistNode *forward; // 下一个节点指针
         unsigned long span;            // 索引跨度
     } level[];  
 } zskiplistNode;
 typedef struct zskiplist {
     struct zskiplistNode *header, *tail; // 头尾节点指针
     unsigned long length;   // 节点数量
     int level;              // 最大的索引层，默认是1
 } zskiplist;

// uint32_t类型: zlbytes，zltail; uint16_t类型: zllen; uint8_t类型: zlend  
 // zlbytes：压缩列表占用的总字节数
 #define ZIPLIST_BYTES(zl)       (*((uint32_t*)(zl))) 
 // zltail：尾节点 zlentry 的偏移量
 #define ZIPLIST_TAIL_OFFSET(zl) (*((uint32_t*)((zl)+sizeof(uint32_t)))) 
 // zllen: 压缩列表包含的 zlentry 节点数量
 #define ZIPLIST_LENGTH(zl)      (*((uint16_t*)((zl)+sizeof(uint32_t)*2)))
 // 压缩列表的 header 大小: zlbytes + zltail + zlen 
 #define ZIPLIST_HEADER_SIZE     (sizeof(uint32_t)*2+sizeof(uint16_t))
 // zlend：压缩列表结束标志
 #define ZIPLIST_END_SIZE        (sizeof(uint8_t))  
 // 指向头节点 zlentry 的起始位置
 #define ZIPLIST_ENTRY_HEAD(zl)  ((zl)+ZIPLIST_HEADER_SIZE) 
 // 指向尾节点 zlentry 的起始位置
 #define ZIPLIST_ENTRY_TAIL(zl)  ((zl)+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))) 
 // 指向压缩列表的结束位置
 #define ZIPLIST_ENTRY_END(zl)   ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-1)

typedef struct zlentry {
     unsigned int prevrawlensize;  // 编码 prevrawlen 的大小，占1个或5个字节
     unsigned int prevrawlen;      // 前一个 entry 的长度
     unsigned int lensize;   // 编码 len 的大小
     unsigned int len;       // 当前节点长度  
     unsigned int headersize; // 当前节点的 header 大小
     unsigned char encoding;  // 编码方式，记录节点的数据类型（字符串或整数）
     unsigned char *p;        // 指向节点的数据（字符串或整数）
 } zlentry;

// 压缩的阈值：元素长度小于48，不压缩；元素压缩前后长度差不超过8，不压缩
 #define MIN_COMPRESS_BYTES 48   
 // 双向链表中的节点
 typedef struct quicklistNode {
     struct quicklistNode *prev; // 指向前一个节点
     struct quicklistNode *next; // 指向后一个节点
     unsigned char *zl; // 指向ziplist
     unsigned int sz;   // 该 ziplist 占用空间大小
     unsigned int count : 16;    // 该 ziplist 存储的元素个数，16bit，最大65536
     unsigned int encoding : 2;   // 是否采用LZF压缩算法: RAW=1 LZF=2
     unsigned int container : 2;  // 存储方式，ziplist=2，未来可能支持其他数据结构
     unsigned int recompress : 1;  // was this node previous compressed?
     unsigned int attempted_compress : 1; // node can't compress; too small
     unsigned int extra : 10; //预留
 } quicklistNode;
 // 管理双向链表
 typedef struct quicklist {
     quicklistNode *head;    // 指向头节点
     quicklistNode *tail;    // 指向尾节点
     unsigned long count;     // 元素总数，即所有ziplist中的元素总数
     unsigned long len;       // 双向链表中的节点个数
     int fill : QL_FILL_BITS;  // ziplist 大小设置           
     unsigned int compress : QL_COMP_BITS;  // 节点压缩深度
     unsigned int bookmark_count: QL_BM_BITS;
     quicklistBookmark bookmarks[];
 } quicklist;
 // 被压缩过的ziplist
 typedef struct quicklistLZF {
     unsigned int sz;   // 压缩后的ziplist的大小
     char compressed[]; // 柔性数组，存储压缩后的ziplist
 } quicklistLZF;

struct __attribute__ ((__packed__)) sdshdr8 {
     // header
     uint8_t len;    // 已经保存的字符串字节数
     uint8_t alloc;  // buf 申请的总字节数
     unsigned char flags; // 不同SDS的头部类型，用来控制SDS头的大小
     // body
     char buf[]; // 存储字符串，柔性数组，不占用空间
 };

// 申请动态字符串的空间
 s = malloc(sizeof(struct sdshdr8) + alloc);
 sds = s + sizeof(struct sdshdr8)
 // 释放空间
 free(sds-sizeof(struct sdshdr8))

typedef struct intset {
     uint32_t encoding;  // 编码方式：支持16位，32位，64位整数
     uint32_t length;    // 元素个数
     int8_t contents[];  // 整数数组，保存集合数据
 } intset;

/* Insert an integer in the intset */
 // 参数：待插入的 *is, 待插入的值 value, 是否插入成功 *success
 intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {
     // 获取当前值对应的编码方式
     uint8_t valenc = _intsetValueEncoding(value);
     // 要插入的位置
     uint32_t pos;
     if (success) *success = 1;
     // 判断编码是否超过了当前 intset 的编码
     if (valenc > intrev32ifbe(is->encoding)) {
         // 超出编码，需要升级，intsetUpgradeAndAdd 函数
         return intsetUpgradeAndAdd(is,value);
     } else {
         // 在当前 intset 中查找值与 value 一样的元素的下标 pos
         if (intsetSearch(is,value,&pos)) {
             // 如果找到了，则无需插入，直接结束并返回失败
             if (success) *success = 0;
             return is;
         }
         // 数组扩容
         is = intsetResize(is,intrev32ifbe(is->length)+1);
         // 移动数组中 pos 之后的元素到 pos + 1，给新元素腾出空间
         if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);
     }
     // 插入新元素
     _intsetSet(is,pos,value);
     // 重置元素长度
     is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
     return is;
 }
 /* Upgrades the intset to a larger encoding and inserts the given integer. */
 static intset *intsetUpgradeAndAdd(intset *is, int64_t value) {
     // 获取当前 insert 编码
     uint8_t curenc = intrev32ifbe(is->encoding);
     // 获取新编码
     uint8_t newenc = _intsetValueEncoding(value);
     // 获取元素个数
     int length = intrev32ifbe(is->length);
     // 判断元素值的正负（超出表示范围正负均可），小于0插入队首，大于0插入队尾
     int prepend = value < 0 ? 1 : 0;
     // 重置编码为新编码
     is->encoding = intrev32ifbe(newenc);
     // 重置数组的大小
     is = intsetResize(is,intrev32ifbe(is->length)+1);
     // 倒序，逐个元素搬到新的位置
     while(length--)
         // _intsetGetEncoded 按照旧编码方式查找旧元素
         // _intsetSet按照新编码方式插入新元素
         _intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));
     // 插入新元素，prepend 决定是队首还是队尾
     if (prepend)
         _intsetSet(is,0,value);
     else
         _intsetSet(is,intrev32ifbe(is->length),value);
     // 修改数组长度
     is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
     return is;
 }
 static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {
     // 初始化二分查找需要的 min、max、mid
     int min = 0, max = intrev32ifbe(is->length)-1, mid = -1;
     int64_t cur = -1;
     // 数组为空，则不用查找
     if (intrev32ifbe(is->length) == 0) {
         if (pos) *pos = 0;
         return 0;
     } else {
         // 数组不为空，判断 value 是否大于最大值，小于最小值
         // 大于最大值，不用找了，插入队尾
         if (value > _intsetGet(is,max)) {
             if (pos) *pos = intrev32ifbe(is->length);
             return 0;
         } else if (value < _intsetGet(is,0)) { // 小于最小值，不用找了，插入队首
             if (pos) *pos = 0;
             return 0;
         }
     }
     // 二分查找
     while(max >= min) {
         mid = ((unsigned int)min + (unsigned int)max) >> 1;
         cur = _intsetGet(is,mid);
         if (value > cur) {
             min = mid+1;
         } else if (value < cur) {
             max = mid-1;
         } else {
             break;
         }
     }
     if (value == cur) {
         if (pos) *pos = mid;
         return 1;
     } else {
         if (pos) *pos = min;
         return 0;
     }
 }