首页 > 编程学习 > Redis2.2.2源码学习——dict中的hashtable扩容和rehash

[置顶]

分类: Redis 377人阅读 评论(0) 收藏 举报
redishash扩容

Redis2.2.2 

        dict是Redis的hash数据结构,所有类型的元素都可以依据key值计算hashkey,然后将元素插入到dict的某个hash链上(采用拉链法解决hash冲突)。其中,dict的中的hashtable(dictht)的扩容是dict很重要的部分。Redis的“管家”函数serverCron会依据一定的算法(dict中的used与size的比值)判定是否开始进行hashtable的扩容。dict中的ht[1]是作为扩容的临时数据,扩容之后,hashtalbe的长度将变长,那么hashtalbe的masksize与原来的makssize就不同了,那么计算出的hashkey也将不同。所以就需要Rehash对ht[0]中的元素重新计算hashkey。

        在Rehash阶段,首先将原来的ht[0]中的元素重新rehash到ht[1]中,故而要耗费大量的力气从新计算原来的ht[0]表中元素的在ht[1]表总的hashkey,并将元素转移到ht[1]的表中。由于这样Rehash会耗费大量的系统资源,如果一次性完成一个dict的Rehash工作,那么将会对系统中的其他任务造成延迟? 作者的处理方式是:同样在serverCron中调用rehash相关函数,1ms的时间限定内,调用rehash()函数,每次仅处理少量的转移任务(100个元素)。这样有点类似于操作系统的时间片轮转的调度算法。


下图是Dict相关数据结构(引用: Redis源码剖析(经典版).docx )



代码分析如下:

[cpp] view plaincopy
  1. -----------------------Data Sturcter----------------------------------------  
  2.   
  3. typedef struct dictEntry {  
  4.     void *key;  
  5.     void *val;  //空类型,redis的类型主要有sds,list,set等,val指针指向其中的结构  
  6.     struct dictEntry *next;  
  7. } dictEntry;  
  8.   
  9. typedef struct dictType {  
  10.     unsigned int (*hashFunction)(const void *key);  
  11.     void *(*keyDup)(void *privdata, const void *key);  
  12.     void *(*valDup)(void *privdata, const void *obj);  
  13.     int (*keyCompare)(void *privdata, const void *key1, const void *key2);  
  14.     void (*keyDestructor)(void *privdata, void *key);  
  15.     void (*valDestructor)(void *privdata, void *obj);  
  16. } dictType;  
  17.   
  18. /* This is our hash table structure. Every dictionary has two of this as we 
  19.  * implement incremental rehashing, for the old to the new table. */  
  20. typedef struct dictht {  
  21.     dictEntry **table;//hash表,每个table[i]链表存储着一个hashkey相等的dictEntry指针  
  22.     unsigned long size;//table[]的大小  
  23.     unsigned long sizemask;// = size-1  
  24.     unsigned long used;//当前table中存储的dictEntry指针的个数  
  25. } dictht;  
  26.   
  27. typedef struct dict {  
  28.     dictType *type;//hash相关的操作hander  
  29.     void *privdata;  
  30.     dictht ht[2];//ht[0]作为dict的实际hash结构,ht[1]做为扩容阶段的转储结构  
  31.     int rehashidx; //标志dict是否处于rehash阶段,如果值为-1,表示不处于rehash。否则,在rehash中所谓hashtalbe的索引下标  
  32.     int iterators; /* number of iterators currently running */  
  33. } dict;  
  34.   
  35. ------------------------Redis's main----------------------------------------  
  36.   
  37. main() >   
  38.     initServerConfig()  
  39.         server.activerehashing = 1;  
  40.     initServer()  
  41.         //设置定时事件,1ms调用一次serverCron()  
  42.         aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);  
  43.     //进入事件轮询,详见:http://blog.csdn.net/ordeder/article/details/12791359  
  44.     aeMain(server.el)  
  45.   
  46. ------------------------serverCron-----------------------------------------  
  47.   
  48. /*serverCron是Redis的协调员,其中就包括:  
[cpp] view plaincopy
  1. 1.检查是否有hashtalbe需要扩容,并执行必要的扩容  
[cpp] view plaincopy
  1. 2.执行incrementRehash执行固定时间的Rehash任务  
[cpp] view plaincopy
  1. */  
  2. int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData)  
  3.     /*记录sever调用serverCron的总次数 
  4.     *对于需要协调的不同的任务,可以依据loops%n的方式设置不同的频率 */  
  5.     int loops = server.cronloops   
  6.     。。。  
  7.     /* We don't want to resize the hash tables while a bacground saving 
  8.      * is in progress: the saving child is created using fork() that is 
  9.      * implemented with a copy-on-write semantic in most modern systems, so 
  10.      * if we resize the HT while there is the saving child at work actually 
  11.      * a lot of memory movements in the parent will cause a lot of pages 
  12.      * copied. 后台没有对数据进行操作的程序...*/  
  13.     if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1) {  
  14.         /*loops % 10 :  
  15.         serverCron没循环10次,进行一次tryResizeHashTables检查*/  
  16.         if (!(loops % 10)) tryResizeHashTables();  
  17.             //下面的for是tryResizeHashTables源码  
  18.             >for (j = 0; j < server.dbnum; j++) {  
  19.                 //htNeedsResize:检查used*100/size < REDIS_HT_MINFILL(设定的阈值)  
  20.                 if (htNeedsResize(server.db[j].dict))   
  21.                     dictResize(server.db[j].dict);//见下文分析 扩容Resize  
  22.                 if (htNeedsResize(server.db[j].expires))  
  23.                     dictResize(server.db[j].expires);  
  24.             >}  
  25.         if (server.activerehashing)   
  26.             incrementallyRehash();//见下文分析 Rehash  
  27.     }  
  28.   
  29. ------------------------扩容Resize------------------------------------------  
  30.   
  31. /* Resize the table to the minimal size that contains all the elements, 
  32.  * but with the invariant of a USER/BUCKETS ratio near to <= 1 */  
  33. int dictResize(dict *d)  
  34. {  
  35.     int minimal;  
  36.   
  37.     if (!dict_can_resize || dictIsRehashing(d)/*rehashidx ?= -1*/return DICT_ERR;  
  38.     minimal = d->ht[0].used;  
  39.     if (minimal < DICT_HT_INITIAL_SIZE)  
  40.         minimal = DICT_HT_INITIAL_SIZE;  
  41.     //minimal = Max(d->ht[0].used,DICT_HT_INITIAL_SIZE)  
  42.     //原来的容量为size,现要扩充到used或DICT_HT_INITIAL_SIZE  
  43.     return dictExpand(d, minimal);  
  44. }  
  45.   
  46. /* Expand or create the hashtable */  
  47. int dictExpand(dict *d, unsigned long size)  
  48. {  
  49.     dictht n; /* the new hashtable */  
  50.     //将size扩展到2^n : while(1) { if (i >= size) return i; i *= 2; }  
  51.     unsigned long realsize = _dictNextPower(size);  
  52.   
  53.     /* the size is invalid if it is smaller than the number of 
  54.      * elements already inside the hashtable */  
  55.     if (dictIsRehashing(d) || d->ht[0].used > size)  
  56.         return DICT_ERR;  
  57.   
  58.     /* Allocate the new hashtable and initialize all pointers to NULL */  
  59.     n.size = realsize;  
  60.     n.sizemask = realsize-1;  
  61.     n.table = zcalloc(realsize*sizeof(dictEntry*));  
  62.     n.used = 0;  
  63.   
  64.     /* 如果是在dict的第一次申请空间?那么就直接将n赋给d->ht[0],而且不需要rehash */  
  65.     if (d->ht[0].table == NULL) {  
  66.         d->ht[0] = n;  
  67.         return DICT_OK;  
  68.     }  
  69.   
  70.     /* Prepare a second hash table for incremental rehashing */  
  71.     d->ht[1] = n;  
  72.     d->rehashidx = 0;    // rehashidx! = -1; 表示d进入了rehash阶段  
  73.     return DICT_OK;  
  74. }  
  75.   
  76.   
  77. -------------------------Rehash-----------------------------------------  
  78. /*记得前文:serverCron作为一个定时器事件的处理函数,定时的时间为1ms 
  79. 在serverCron会调用incrementallyRehash()函数去不断的完成rehash任务。 
  80. 这里说“不断的"的意思是,rehash的任务不是一次性的函数调用完成的,可能需要 
  81. serverCron调用多次incrementallyRehash()来完成。 
  82. 下文就是incrementallyRehash()函数和子函数的额调用关系,incrementallyRehash() 
  83. 的执行限时为1ms,在这时间内,dictRehash()会以一定量任务(100)为单元进行d->ht的 
  84. 转移。 
  85. */  
  86.   
  87. incrementallyRehash(void)  
  88.     遍历server.db[],对db中的dict进行rehash,每个dict的限定时间为1ms  
  89.     dictRehashMilliseconds(server.db[j].dict,1)  
  90.         while(dictRehash的执行时间<1ms)  
  91.             dictRehash(d,100)//详见下文源码  
  92.                 将dict->ht[1]取出100个元素(dictEntry) Rehash到dict->ht[0]  
  93.       
  94.   
  95. /* Performs N steps of incremental rehashing. Returns 1 if there are still 
  96.  * keys to move from the old to the new hash table, otherwise 0 is returned. 
  97.  * Note that a rehashing step consists in moving a bucket (that may have more 
  98.  * thank one key as we use chaining) from the old to the new hash table. */  
  99.   
  100. /*将依次将d->ht[1].table[]中的元素搬到d->ht[0].table[],修改相关的used。 
  101. d->rehashidx:记录着当前的hashtable中搬了那一条链表的索引下标 
  102. d->ht[0].table[d->rehashidx] => d->ht[0].table[d->rehashidx]              
  103. 完成一个dict的Rehash后,重置d->rehashidx = -1 */  
  104. int dictRehash(dict *d, int n) {  
  105.     if (!dictIsRehashing(d)) return 0;  
  106.   
  107.     while(n--) {  
  108.         dictEntry *de, *nextde;  
  109.   
  110.         /* Check if we already rehashed the whole table... */  
  111.         //d->ht[0].used == 0 : 说明d->ht[0] 已经全部搬到 d->ht[1]  
  112.         if (d->ht[0].used == 0) {  
  113.             zfree(d->ht[0].table);  
  114.             d->ht[0] = d->ht[1];  
  115.             _dictReset(&d->ht[1]);  
  116.             d->rehashidx = -1;   //该dict的Rehash结束 设置为 -1  
  117.             return 0;  
  118.         }  
  119.   
  120.         /* Note that rehashidx can't overflow as we are sure there are more 
  121.          * elements because ht[0].used != 0 */  
  122.         while(d->ht[0].table[d->rehashidx] == NULL) d->rehashidx++;  
  123.         de = d->ht[0].table[d->rehashidx];  
  124.         /* Move all the keys in this bucket from the old to the new hash HT */  
  125.         while(de) {  
  126.             unsigned int h;  
  127.   
  128.             nextde = de->next;  
  129.             /* Get the index in the new hash table */  
  130.             h = dictHashKey(d, de->key) & d->ht[1].sizemask;  
  131.             de->next = d->ht[1].table[h];  
  132.             d->ht[1].table[h] = de;  
  133.             d->ht[0].used--;   
  134.             d->ht[1].used++;   
  135.             de = nextde;  
  136.         }  
  137.         d->ht[0].table[d->rehashidx] = NULL;  
  138.         d->rehashidx++;  
  139.     }  
  140.     return 1;  
  141. }  


本文链接:https://www.ngui.cc/el/1380838.html
Copyright © 2010-2022 ngui.cc 版权所有 |关于我们| 联系方式| 豫B2-20100000