memcached-version-1.4.25

介绍

memcache 的 hashtable 就是一块保存 item* 固定大小的内存区域，也就是一个固定大小的指针数组，在程序启动的时候会初始化这个 hashtable 数组的大小，主要核心涉及到的点就是 hashtable动态扩容、hashtable段锁等，对应的 hashtable 处理的文件为 hash.c 、hash.h 、 assoc.c、 assoc.h

Memcache 哈希表

memcache 哈希表

源码实现

初始化设置 hash 算法，hash_init ()

// 目前只提供2种hash算法 jenkins 和 murmur3
enum hashfunc_type {
    JENKINS_HASH=0, MURMUR3_HASH  
};
//启动memcache的时候调用
int hash_init(enum hashfunc_type type) {
    switch(type) {
        case JENKINS_HASH:
            hash = jenkins_hash; // jenkins_hash 函数指针
            settings.hash_algorithm = "jenkins";
            break;
        case MURMUR3_HASH:
            hash = MurmurHash3_x86_32;  // murmur3_hash 函数指针
            settings.hash_algorithm = "murmur3"; 
            break;
        default:
            return -1;
    }
    return 0;
}

初始化 hashtable 表，assoc_init()

settings.hashpower_init //hashtable的基准值，启动memcache的时候设定

void assoc_init(const int hashtable_init) {
    // 如果存在则赋值 hashpower 没有的话就用默认的 hashpower = 16
    if (hashtable_init) {
        hashpower = hashtable_init;
    }
    // hashsize 根据传入的 hashpower 算出最终的 hashtable 长度
    // hashsize(n) -> #define hashsize(n) ((ub4)1<<(n)) 
    // 实际上就是进行右移运算，右移 hashpower 位 ( 1 << 16 = 65536 )
    // 申请内存 item** primary_hashtable -> hashtable头指针 
    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 *); // hashtable 占了多少字节内存
    STATS_UNLOCK();
}

hashtable 段锁

在 memcache 初始化线程那部分，还会初始化一些互斥锁， 其中就包括了 hashtable 的段锁，什么是段锁，就是有可能会多个key对应一把锁，而不是每一个key都对应一把锁，因为不同的key可能hash出来的值一样，那么就都会对应这一把锁。

初始化 hashtable 锁

// memcached_thread_init 函数其中一段代码
  
void memcached_thread_init(int nthreads, struct event_base *main_base) {
    int         i;
    int         power;
    
    //.........
    
    //根据线程数设置 hashtable 锁的启动值 power 跟上面的 hashpower 同理
    if (nthreads < 3) {
        power = 10;
    } else if (nthreads < 4) {
        power = 11;
    } else if (nthreads < 5) {
        power = 12;
    } else {
        /* 8192 buckets, and central locks don't scale much past 5 threads */
        power = 13;
    }
    // power 小于 hashpower 因为 hashtable锁的数量 和 hashtable的数量 并不是一一对应的
    // 也就是说并不是 每一个hashtable的key都对应一把锁，memcache为了省内存，采用多个key
    // 对应一把锁，也就是段锁
    if (power >= hashpower) {
        fprintf(stderr, "Hash table power size (%d) cannot be equal to or less than item lock table (%d)\n", hashpower, power);
        fprintf(stderr, "Item lock table grows with `-t N` (worker threadcount)\n");
        fprintf(stderr, "Hash table grows with `-o hashpower=N` \n");
        exit(1);
    }

    //计算hashtable锁的长度, 默认4个线程 power = 12 、 1 << 12 = 4096/锁
    item_lock_count = hashsize(power); 
    item_lock_hashpower = power;
    
    //申请锁
    //之后这个item_locks锁并不会随着hashtable的扩容而扩容
    //也就是说无论之后hashtable变成多大，都会对应这 hashsize(power) -> 4096/锁 
    //这也会导致越来越多的key对应一把锁
    item_locks = calloc(item_lock_count, sizeof(pthread_mutex_t));
    if (! item_locks) {
        perror("Can't allocate item locks");
        exit(1);
    }
    //循环初始化item_locks锁
    for (i = 0; i < item_lock_count; i++) {
        pthread_mutex_init(&item_locks[i], NULL);
    }
    
    //.........
}

至此通过上面的代码已经将 memcache hashtable 初始化完毕

插入 、 扩容 hashtable

memcache 的 hashtable 扩容处理的方式是，在程序启动阶段的时候开一个线程处于待命状态，当需要扩容的时候触发该线程，动态的进行扩容处理，而且在扩容操作的时候也不是表锁，而是利用上面说的段锁

扩容流程:

在 hashtable 扩容的时候 memcache 会把当前 primary_hashtable （哈希表）复制一份给 old_hashtable（哈希表），然后对 primary_hashtable 进行扩容 (1 << hashpower + 1)，如果当前正处于 hashtable 扩容阶段， 同时有请求要访问 key，则会判断当前的 key - hash 之后的索引位置是否小于当前迁移的索引位置，如果小于则代表已经迁移到新的 primary_hashtable 的索引位置了，如果大于则代表还未迁移到则在新的 primary_hashtable，所以就还在老的 old_hashtable 位置操作，扩容完成之后会释放 old_hashtable，然后就全部都在 primary_hashtable 操作了。

说明:

在迁移 hashtable 的时候，会从小到大一个一个索引位置进行迁移，而这个索引位置就是hash之后的值，所以在迁移每一个索引位置的时候都会对当前的索引位置加锁，这个锁用的就是上面说的段锁，但可能锁的数量有限，会出现很多索引位置共用同一个锁的情况。
例如:
0 & (4096-1) = 0 、4096 & (4096-1) = 0 、8192 & (4096-1) = 0 这几个hash索引位置都会用到 0 这把锁
如果这个时候一个key正好要访问，同时这个key-hash之后的值正好跟我们迁移的这个索引位置对应则会堵塞，因为已经加锁，其他的key如果不命中我们正在迁移的这个位置则正常访问，至于访问 primary_hashtable 还是 old_hashtable 则依据上面（扩容流程）说的条件。

插入 hashtable 函数， assoc_insert()

// hv = hash(key, nkey); 哈希值

int assoc_insert(item *it, const uint32_t hv) {
    unsigned int oldbucket;
    
    //判断是否正在扩容
    if (expanding &&
        (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket)
    {
        //插入到hashtable
        it->h_next = old_hashtable[oldbucket];
        old_hashtable[oldbucket] = it;
    } else {
       //插入到hashtable
       it->h_next = primary_hashtable[hv & hashmask(hashpower)];
       primary_hashtable[hv & hashmask(hashpower)] = it;
    }
    
    // 加锁，防止多个线程同时触发扩容操作
    pthread_mutex_lock(&hash_items_counter_lock);
    hash_items++;
    // 判断是否需要扩容
    if (! expanding && hash_items > (hashsize(hashpower) * 3) / 2) {
        //触发扩容操作
        assoc_start_expand();
    }
    pthread_mutex_unlock(&hash_items_counter_lock);

    MEMCACHED_ASSOC_INSERT(ITEM_key(it), it->nkey, hash_items);
    return 1;
}

触发扩容操作 ， assoc_start_expand()

static void assoc_start_expand(void) {
    if (started_expanding)
        return;

    started_expanding = true;
    // 唤醒 hashtable 扩容线程
    pthread_cond_signal(&maintenance_cond);
}

hashtable 扩容线程函数，assoc_maintenance_thread()

// 默认会堵塞在这个位置 pthread_cond_wait(&maintenance_cond, &maintenance_lock);
// 直到上面 pthread_cond_signal 唤醒操作, 然后往下执行.

static void *assoc_maintenance_thread(void *arg) {

    mutex_lock(&maintenance_lock);
    
    // 死循环
    while (do_run_maintenance_thread) {
        int ii = 0;

        /* 循环迁移 old_hashtable ->  primary_hashtable */
        for (ii = 0; ii < hash_bulk_move && expanding; ++ii) {
            item *it, *next;
            int bucket;
            void *item_lock = NULL;
            
            // 把当前正在迁移的索引位置加锁
            if ((item_lock = item_trylock(expand_bucket))) {
                    // 循环处理当前索引位置的所有item
                    // 因为在当前索引位置可能会存在多个item
                    // 就是hash冲突链式解决
                    for (it = old_hashtable[expand_bucket]; NULL != it; it = next) {
                        // 获取下一个item
                        next = it->h_next;
                        // 按照新的hashtable长度重新定位一个索引位置
                        bucket = hash(ITEM_key(it), it->nkey) & hashmask(hashpower);
                        // 赋值保存
                        it->h_next = primary_hashtable[bucket];
                        primary_hashtable[bucket] = it;
                    }
                    // 当前的索引位置item都迁移完毕之后，将之前的hashtable索引位置至空
                    old_hashtable[expand_bucket] = NULL;
                    // 更新当前索引位置
                    expand_bucket++;
                    // 判断是否全部迁移完毕
                    if (expand_bucket == hashsize(hashpower - 1)) {
                        expanding = false; // 关闭正在扩容hashtable状态
                        free(old_hashtable); // 释放 old_hashtable
                        STATS_LOCK();
                        stats.hash_bytes -= hashsize(hashpower - 1) * sizeof(void *);
                        stats.hash_is_expanding = 0;
                        STATS_UNLOCK();
                        if (settings.verbose > 1)
                            fprintf(stderr, "Hash table expansion done\n");
                    }

            } else {
                // 如果加锁失败，可能存在别的线程正在操作hashtable当前索引位置的item
                // 所以延时等待，直到抢到锁为止
                usleep(10*1000);
            }
            // 释放锁
            if (item_lock) {
                item_trylock_unlock(item_lock);
                item_lock = NULL;
            }
        }

        if (!expanding) {
            /* We are done expanding.. just wait for next invocation */
            started_expanding = false;
            // 等待唤醒
            pthread_cond_wait(&maintenance_cond, &maintenance_lock);
            pause_threads(PAUSE_ALL_THREADS);
            // 扩容 hashtable 操作
            assoc_expand();
            //往下执行while循环
            pause_threads(RESUME_ALL_THREADS);
        }
    }
    return NULL;
}

扩容 hashtable 函数，assoc_expand()

static void assoc_expand(void) {
    // 保存一份当前的 hashtable
    old_hashtable = primary_hashtable;
    
    // 扩容，每次扩容 hashpower + 1 
    // (1 << 16) = 65536、(1 << 17) = 131072
    primary_hashtable = calloc(hashsize(hashpower + 1), sizeof(void *));
    if (primary_hashtable) {
        if (settings.verbose > 1)
            fprintf(stderr, "Hash table expansion starting\n");
        hashpower++; // hashpower++
        expanding = true; // 表示当前正在扩容
        expand_bucket = 0; // 当前迁移到primary_hashtable索引位置
        STATS_LOCK();
        stats.hash_power_level = hashpower;
        stats.hash_bytes += hashsize(hashpower) * sizeof(void *);
        stats.hash_is_expanding = 1;
        STATS_UNLOCK();
    } else {
        primary_hashtable = old_hashtable;
        /* Bad news, but we can keep running. */
    }
}

查找 hashtable 函数，assoc_find()

item *assoc_find(const char *key, const size_t nkey, const uint32_t hv) {
    item *it;
    unsigned int oldbucket;
    
    // 是否正处于扩容操作
    if (expanding &&
        (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket)
    {
        // 获取item
        it = old_hashtable[oldbucket];
    } else {
        // 获取item
        it = primary_hashtable[hv & hashmask(hashpower)];
    }

    item *ret = NULL;
    int depth = 0;
    //循环判断，因为hash冲突采用链式解决法，所以需要遍历这个链，找到key相等的
    while (it) {
        if ((nkey == it->nkey) && (memcmp(key, ITEM_key(it), nkey) == 0)) {
            ret = it;
            break;
        }
        it = it->h_next;
        ++depth;
    }
    MEMCACHED_ASSOC_FIND(key, nkey, depth);
    return ret;
}

结束

以上就是 Memcache 的哈希表以及对应的操作，由于是多线程模型，所以无论是 (插入 、查找、 扩容) 都会对当前操作的key对应的索引位置加锁 item_locks，所以随着 hashtable 不断的扩大，锁的争抢也会越来越大，性能可能也会存在一些影响.