memcached哈希表操作主要逻辑笔记

 /* how many powers of 2's worth of buckets we use */

 unsigned int hashpower = HASHPOWER_DEFAULT; /* 哈希表bucket的级别，(1<<hashpower) == bucket的个数 */

 /* Main hash table. This is where we look except during expansion. */

 /**主要的哈希表, 用来存储memcached的key-value数据

 * 扩容时会将数据暂存在另一个指针，之后重新分配空间，再以bucket为单位

 * 将已有的数据迁移到这张表,所以这张表始终代表最新的数据

 */

 static item** primary_hashtable = ;

 /*

 * Previous hash table. During expansion, we look here for keys that haven't

 * been moved over to the primary yet.

 */

 static item** old_hashtable = ; /**原有的哈希表，只有在扩容时才会使用，保存原有的哈希表的数据 */

 /* Number of items in the hash table. */

 static unsigned int hash_items = ; /** 整个哈希表中item的个数*/

 /* Flag: Are we in the middle of expanding now? */

 static bool expanding = false; /** 标识是否正在扩容哈希表*/

 static bool started_expanding = false; /* 是否已经开始扩容*/

 /*

 * During expansion we migrate values with bucket granularity; this is how

 * far we've gotten so far. Ranges from 0 .. hashsize(hashpower - 1) - 1.

 */

 /**

 * 在扩容期间，数据的迁移是以bucket为单位进行迁移, expand_bucket表示迁移进行到第几个bucket

 */

 static unsigned int expand_bucket = ;

 /**

 * 哈希表的初始化

 * 整个哈希表类似于一个二维数组，初始化分配bucket的空间，具体的item直接申请空间链接在bucket子链上(拉链法解决key冲突)

 */

 void assoc_init(const int hashtable_init) {

 if (hashtable_init) {

 hashpower = hashtable_init;

 }

 /**初始化哈希表的存储空间*/

 primary_hashtable = calloc(hashsize(hashpower), sizeof(void *));

 if (! primary_hashtable) {

 fprintf(stderr, "Failed to init hashtable.\n");

 exit(EXIT_FAILURE);

 }

 STATS_LOCK();

 stats.hash_power_level = hashpower;

 stats.hash_bytes = hashsize(hashpower) * sizeof(void *);

 STATS_UNLOCK();

 }

 /**

 * 根据key查找item

 * hv 表示key的hash值

 */

 item *assoc_find(const char *key, const size_t nkey, const uint32_t hv) {

 item *it; //指向所属的bucket地址

 unsigned int oldbucket;

 //先判断是否正在扩容

 if (expanding && //正在扩容时继续判断key所属的bucket是否已经迁移到old_hashtable

 (oldbucket = (hv & hashmask(hashpower - ))) >= expand_bucket)

 {

 it = old_hashtable[oldbucket]; //已经迁移到old_hashtable则在这里查找bucket

 } else {

 it = primary_hashtable[hv & hashmask(hashpower)];//没有迁移或者尚未迁移所属的bucket

 }

 item *ret = NULL;

 int depth = ;

 /** 循环比较拉链上的每个item的key值,相等则返回item的引用*/

 while (it) {

 if ((nkey == it->nkey) && (memcmp(key, ITEM_key(it), nkey) == )) {

 ret = it;

 break;

 }

 it = it->h_next;

 ++depth;

 }

 MEMCACHED_ASSOC_FIND(key, nkey, depth);

 return ret;

 }

 /* returns the address of the item pointer before the key. if *item == 0,

 the item wasn't found */

 /**

 * 查找item的地址

 */

 static item** _hashitem_before (const char *key, const size_t nkey, const uint32_t hv) {

 item **pos;

 unsigned int oldbucket;

 /** 同样是先确定在哪一张表里找*/

 if (expanding &&

 (oldbucket = (hv & hashmask(hashpower - ))) >= expand_bucket)

 {

 pos = &old_hashtable[oldbucket];

 } else {

 pos = &primary_hashtable[hv & hashmask(hashpower)];

 }

 /** */

 while (*pos && ((nkey != (*pos)->nkey) || memcmp(key, ITEM_key(*pos), nkey))) {

 pos = &(*pos)->h_next;

 }

 return pos;

 }

 /* grows the hashtable to the next power of 2. */

 /** 将已有的哈希表扩容为原来的2倍buckets数量*/

 static void assoc_expand(void) {

 /** old_hashtable指向已有的primary_hashtable*/

 old_hashtable = primary_hashtable;

 /**重新为 primary_hashtable分配空间*/

 primary_hashtable = calloc(hashsize(hashpower + ), sizeof(void *));

 if (primary_hashtable) { /** 分配成功*/

 if (settings.verbose > )

 fprintf(stderr, "Hash table expansion starting\n");

 hashpower++;

 expanding = true; /** 设置开始扩容标识*/

 expand_bucket = ; /** 已迁移的bucket序号*/

 STATS_LOCK();

 stats.hash_power_level = hashpower;

 stats.hash_bytes += hashsize(hashpower) * sizeof(void *);

 stats.hash_is_expanding = ;

 STATS_UNLOCK();

 } else {

 primary_hashtable = old_hashtable; /** 分配失败*/

 /* Bad news, but we can keep running. */

 }

 }

 static void assoc_start_expand(void) {

 if (started_expanding)

 return;

 started_expanding = true;

 pthread_cond_signal(&maintenance_cond);

 }

 /* Note: this isn't an assoc_update. The key must not already exist to call this */

 /** 插入一个item到哈希表，这里必须保证item->key尚未存在已有的哈希表*/

 int assoc_insert(item *it, const uint32_t hv) {

 unsigned int oldbucket;

 // assert(assoc_find(ITEM_key(it), it->nkey) == 0); /* shouldn't have duplicately named things defined */

 if (expanding &&

 (oldbucket = (hv & hashmask(hashpower - ))) >= expand_bucket)

 { /** 正在扩容且对应的bucket尚未被迁移到primary_hashtable*/

 it->h_next = old_hashtable[oldbucket];

 old_hashtable[oldbucket] = it;

 } else { /** 没有在扩容或者对应的bucket已经被迁移到primary_hashtable*/

 it->h_next = primary_hashtable[hv & hashmask(hashpower)];

 primary_hashtable[hv & hashmask(hashpower)] = it;

 }

 // 更新hash_item的数量

 hash_items++;

 /** 哈希表item的数量超过bucket数的3分之2, 这里表示只关心由于存储item数量增长必须引起的扩容*/

 if (! expanding && hash_items > (hashsize(hashpower) * ) / ) {

 assoc_start_expand(); //发送条件变量满足的信号

 }

 MEMCACHED_ASSOC_INSERT(ITEM_key(it), it->nkey, hash_items);

 return ;

 }

 /** 删除一个item*/

 void assoc_delete(const char *key, const size_t nkey, const uint32_t hv) {

 /** 找到指向item地址的指针*/

 item **before = _hashitem_before(key, nkey, hv);

 if (*before) {

 item *nxt;

 hash_items--; /** 减少1个 */

 /* The DTrace probe cannot be triggered as the last instruction

 * due to possible tail-optimization by the compiler

 */

 MEMCACHED_ASSOC_DELETE(key, nkey, hash_items);

 /** 链表操作删除一个元素*/

 nxt = (*before)->h_next;

 (*before)->h_next = ; /* probably pointless, but whatever. */

 *before = nxt;

 return;

 }

 /* Note: we never actually get here. the callers don't delete things

 they can't find. */

 assert(*before != );

 }

 /** 标识是否需要执行维护线程主要逻辑*/

 static volatile int do_run_maintenance_thread = ;

 #define DEFAULT_HASH_BULK_MOVE 1

 int hash_bulk_move = DEFAULT_HASH_BULK_MOVE;

 /** 哈希表维护线程的主要逻辑*/

 static void *assoc_maintenance_thread(void *arg) {

 /** 主线程未退出时，这里基本是进入一个无限循环*/

 while (do_run_maintenance_thread) {

 int ii = ;

 /* Lock the cache, and bulk move multiple buckets to the new

 * hash table. */

 /** 获取worker线程共享的item_global_lock锁,批量迁移buckets到新的哈希表*/

 item_lock_global();

 mutex_lock(&cache_lock);

 /** 这个循环默认只走一次，主要目的是不想过久的占用item全局锁，影响worker线程工作效率*/

 for (ii = ; ii < hash_bulk_move && expanding; ++ii) {

 item *it, *next;

 int bucket;

 /** 对bucket上拉链的每一个item进行重新hash到primary_hashtable*/

 for (it = old_hashtable[expand_bucket]; NULL != it; it = next) {

 next = it->h_next;

 /** 重新计算所属的bucket*/

 bucket = hash(ITEM_key(it), it->nkey) & hashmask(hashpower);

 /** 加入到primary_hashtable对应bucket的头部*/

 it->h_next = primary_hashtable[bucket];

 primary_hashtable[bucket] = it;

 }

 /** 将old_hashtable上已经被迁移的bucket置为NULL*/

 old_hashtable[expand_bucket] = NULL;

 /** 递增迁移的bucket序号*/

 expand_bucket++;

 /** 判断是否已经迁移完*/

 if (expand_bucket == hashsize(hashpower - )) {

 expanding = false;

 free(old_hashtable); //释放old_hashtable

 STATS_LOCK();

 stats.hash_bytes -= hashsize(hashpower - ) * sizeof(void *);

 stats.hash_is_expanding = ;

 STATS_UNLOCK();

 if (settings.verbose > )

 fprintf(stderr, "Hash table expansion done\n");

 }

 }

 /** 释放锁*/

 mutex_unlock(&cache_lock);

 /** 释放全局锁，这样worker线程才有机会获得锁进而服务用户请求，减少等待时间*/

 item_unlock_global();

 /** 未进行扩容或者扩容结束*/

 if (!expanding) {

 /** 通知其他线程使用细粒度的锁,通过线程pipe进行通信*/

 /* finished expanding. tell all threads to use fine-grained locks */

 switch_item_lock_type(ITEM_LOCK_GRANULAR);

 /**恢复slabs的自平衡锁，确保哈希表扩容不会与slabs重新分配同时进行*/

 slabs_rebalancer_resume();

 /**本次扩容完成，等待下一次调用*/

 /* We are done expanding.. just wait for next invocation */

 mutex_lock(&cache_lock);

 started_expanding = false;

 /** 刚启动系统时尚未需要进行扩容,线程会阻塞到这里等待线程条件信号*/

 /** 等待条件变量满足信号*/

 pthread_cond_wait(&maintenance_cond, &cache_lock);

 /* Before doing anything, tell threads to use a global lock */

 mutex_unlock(&cache_lock);

 /** 确保slabs没有正在进行重新分配*/

 slabs_rebalancer_pause();

 /**通过pipe的方式通知worker线程改变使用锁的粒度为全局锁*/

 switch_item_lock_type(ITEM_LOCK_GLOBAL);

 mutex_lock(&cache_lock);

 /** 开始扩容*/

 assoc_expand();

 mutex_unlock(&cache_lock);

 }

 }

 return NULL;

 }

 static pthread_t maintenance_tid;

 /** 启动哈希表扩容监听线程*/

 int start_assoc_maintenance_thread() {

 int ret;

 char *env = getenv("MEMCACHED_HASH_BULK_MOVE");

 if (env != NULL) {

 hash_bulk_move = atoi(env);

 if (hash_bulk_move == ) {

 hash_bulk_move = DEFAULT_HASH_BULK_MOVE;

 }

 }

 /** 创建线程*/

 if ((ret = pthread_create(&maintenance_tid, NULL,

 assoc_maintenance_thread, NULL)) != ) {

 fprintf(stderr, "Can't create thread: %s\n", strerror(ret));

 return -;

 }

 return ;

 }

 /** 停止扩容线程，基本是在主线程退出时才会被调用*/

 void stop_assoc_maintenance_thread() {

 mutex_lock(&cache_lock);

 do_run_maintenance_thread = ;

 pthread_cond_signal(&maintenance_cond);

 mutex_unlock(&cache_lock);

 /* Wait for the maintenance thread to stop */

 pthread_join(maintenance_tid, NULL);

 }