首页 > 代码库 > Memcached源码分析之assoc.c

Memcached源码分析之assoc.c

2024-08-12 11:32:00   224人阅读

#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.\n");
        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\n");
        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\n");
            }
        }
 
        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\n", 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); //等待线程退出
}

Memcached源码分析之assoc.c


联系
我们