Memcached源码分析之assoc.c

#include "memcached.h"

#include <sys/stat.h>

#include <sys/socket.h>

#include <sys/signal.h>

#include <sys/resource.h>

#include <fcntl.h>

#include <netinet/in.h>

#include <errno.h>

#include <stdlib.h>

#include <stdio.h>

#include <string.h>

#include <assert.h>

#include <pthread.h>

 

static pthread_cond_t maintenance_cond = PTHREAD_COND_INITIALIZER;

 

 

typedef  unsigned long  int  ub4;   /* unsigned 4-byte quantities */

typedef  unsigned       char ub1;   /* unsigned 1-byte quantities */

 

unsigned int hashpower = HASHPOWER_DEFAULT;

 

#define hashsize(n) ((ub4)1<<(n)) //2^n 次方  默认n是16（上面hashpower）

#define hashmask(n) (hashsize(n)-1) //0x1111 1111 1111 1111

 

static item** primary_hashtable = 0; //主hash表，注意理解，这个表只是存指针，没有存item

 

 

static item** old_hashtable = 0; //旧的hash表，在扩展hash表的时候用到

 

static unsigned int hash_items = 0; //hash表总数

 

static bool expanding = false;    //是不是正在扩展hash表中（通过线程assoc_maintenance_thread）

static bool started_expanding = false;

 

/*

 在扩展时，是以桶为粒度进行的，这是告诉我们扩张到哪个桶了。3

 从0 到 hashsize(hashpower - 1) - 1 

 */

static unsigned int expand_bucket = 0;

 

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. ");

        exit(EXIT_FAILURE);

    }

    STATS_LOCK();

    stats.hash_power_level = hashpower;

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

    STATS_UNLOCK();

}

 

/**

通过key查找item

*/

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

    item *it;

    unsigned int oldbucket;

 

 

    //hv & hashmask(hashpower)得到的是桶在hash表中的下标

    if (expanding &&

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

    {

        it = old_hashtable[oldbucket];

    } else {

        it = primary_hashtable[hv & hashmask(hashpower)]; //找出item所在的桶链表的首item

    }

 

    item *ret = NULL;

    int depth = 0;

    //遍历相同桶的链表，直到指定的key名为止。

    while (it) {

        //这里为什么要先判断长度？&&的执行过程是先判断左边，如果不为true，那右边的条件也不用判断了，

        //所以个人认为是为了判断memcmp(key, ITEM_key(it), nkey)的调用

        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;

}

 

/**

这里的查找过程和上面的assoc_find 基本一致，不同的地方在于：

这里返回的是指向 “指向当前item的指针”，并且引用当前item的上一个item的h_next，

所以这里返回的就是当前item在桶链表中的前一个item的h_next，这也是为什么命名叫

_hashitem_before的原因~

*/

 

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 - 1))) >= 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;

}

 

/**

扩展哈希表

*/

static void assoc_expand(void) {

    old_hashtable = primary_hashtable;

 

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

    if (primary_hashtable) {

        if (settings.verbose > 1)

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

        hashpower++;

        expanding = true;

        expand_bucket = 0;

        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. */

    }

}

 

/**

主要是唤醒哈希表维护线程，执行哈希表扩展工作。

*/

static void assoc_start_expand(void) {

    if (started_expanding)

        return;

    started_expanding = true;

    /**

     发送一个信号给正在处于阻塞等待状态的哈希表维护线程。见assoc_maintenance_thread

    */

    pthread_cond_signal(&maintenance_cond);

}

 

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

/**

把item插入到hash表

*/

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 */

 

    //hv & hashmask(hashpower)得到的是桶在hash表中的下标

 

    if (expanding &&

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

    {

        it->h_next = old_hashtable[oldbucket];

        old_hashtable[oldbucket] = it;

    } else {

        //哈希表难免会冲突，这里用链表保存相同桶下标的item

        //这里是把新的item放到桶的链表头

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

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

    }

 

    hash_items++;

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

        //当哈希表中的item数大于哈希表桶数的1.5倍时，开始扩展哈希表

        assoc_start_expand();

    }

 

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

    return 1;

}

 

/**

从哈希表中删除某个item

*/

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

    /**

    调用_hashitem_before取到指向 指向当前item的上一个item的h_next指针

    */

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

 

    //下面利用before指针，把当前item的h_next指向0，把上一个item的h_next指向原来before的h_next达到删除作用

    if (*before) {

        item *nxt;

        hash_items--;

        /* 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 = 0;   /* 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 != 0);

}

 

 

static volatile int do_run_maintenance_thread = 1;

 

#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 = 0;

 

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

         * hash table. */

        item_lock_global();

        mutex_lock(&cache_lock);

 

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

            item *it, *next;

            int bucket;

 

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

                next = it->h_next;

 

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

                it->h_next = primary_hashtable[bucket];

                primary_hashtable[bucket] = it;

            }

 

            old_hashtable[expand_bucket] = NULL;

 

            expand_bucket++;

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

                expanding = false;

                free(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 ");

            }

        }

 

        mutex_unlock(&cache_lock);

        item_unlock_global();

 

        if (!expanding) {

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

            switch_item_lock_type(ITEM_LOCK_GRANULAR);

            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_rebalancer_pause();

            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 == 0) {

            hash_bulk_move = DEFAULT_HASH_BULK_MOVE;

        }

    }

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

                              assoc_maintenance_thread, NULL)) != 0) { //assoc_maintenance_thread为线程执行入口

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

        return -1;

    }

    return 0;

}

 

/**

停止哈希表维护线程，在memcached服务退出时执行，见memcached.c中main函数，event_base_loop之后

*/

void stop_assoc_maintenance_thread() {

    mutex_lock(&cache_lock);

 

    /**

    发送信号assoc_maintenance_thread进入while循环相应的上下文，

    而设置do_run_maintenance_thread = 0让线程在下次while(do_run_maintenance_thread)语句

    中退出循环，线程退出。

    */

    do_run_maintenance_thread = 0;

    pthread_cond_signal(&maintenance_cond); 

    mutex_unlock(&cache_lock);

 

    pthread_join(maintenance_tid, NULL); //等待线程退出

}