memcache源码分析之assoc

ifuleyou

　　memcache对item信息的存储是采用的hash表的形式,而item的内容则是存储在slab中,本篇文章只介绍item在hash表中的存储。关于slab的存储介绍请关注后续文章。
　　item经过hash后存储在一个桶中,这个桶是hash表的一个元素,在同一个桶中,item是通过链表来存储的。
　　这部分的初始化工作在memcached.c的main函数中,初始化时,分配了2^16个元素的hash表,如果hash表比较大了,需要重新申请内存扩充。
　　那什么时候需要扩充呢,当每个item被加入到hash表中时,程序会去计算item的数量,如果item的数量大于2^16的1.5倍,即开始扩充,这部分的代码在assoc_insert函数中。
　　
　　扩充的过程
　　hash表在扩充的时候会去申请一个是原来2被大小的空间,然后启动一个线程去同步原hash表中的item数据,如果在扩充的时候,有个请求item的请求,则会判断这个item是在新的hash表中还是原来的hash表中,然后再去各自的hash表的桶中查找。
　　
　　下面是源码分析,感兴趣的同学可以查阅
　　
　　assoc.h
　　


/* associative array */
void assoc_init(void);
item *assoc_find(const char *key, const size_t nkey);
int assoc_insert(item *item);
void assoc_delete(const char *key, const size_t nkey);
void do_assoc_move_next_bucket(void);
int start_assoc_maintenance_thread(void);
void stop_assoc_maintenance_thread(void);


　　
　　assoc.c
　　
　　


/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
* Hash table
*
* The hash function used here is by Bob Jenkins, 1996:
*    <http://burtleburtle.net/bob/hash/doobs.html>
*       "By Bob Jenkins, 1996.  bob_jenkins@burtleburtle.net.
*       You may use this code any way you wish, private, educational,
*       or commercial.  It's free."
*
* The rest of the file is licensed under the BSD license.  See LICENSE.
*/
#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;
typedef  unsigned       char ub1;
//2的幂大小,桶的个数
static unsigned int hashpower = 16;
#define hashsize(n) ((ub4)1<<(n))
#define hashmask(n) (hashsize(n)-1)
/* Main hash table. This is where we look except during expansion. */
//hash表
static item** primary_hashtable = 0;
//hash表扩充后的旧hash表
static item** old_hashtable = 0;
//hash表中item个数
static unsigned int hash_items = 0;
//是否在扩充中标识
static bool expanding = false;

//扩充过程中,桶迁移到新hash表的进度,值的范围是0 .. hashsize(hashpower - 1) - 1
static unsigned int expand_bucket = 0;
//初始化
void assoc_init(void) {
primary_hashtable = calloc(hashsize(hashpower), sizeof(void *));//为hash表分配内存空间
if (! primary_hashtable) {
fprintf(stderr, "Failed to init hashtable.\n");
exit(EXIT_FAILURE);
}
}

//查找item
item *assoc_find(const char *key, const size_t nkey) {
uint32_t hv = hash(key, nkey, 0);
item *it;
unsigned int oldbucket;
//寻找在哪个桶中
if (expanding && (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket)//扩充过程中,并且要寻找的item还在旧hash 表中
{
it = old_hashtable[oldbucket];
} else {
it = primary_hashtable[hv & hashmask(hashpower)];
}
item *ret = NULL;
int depth = 0;
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;
}

//获取item的指针的指针
static item** _hashitem_before (const char *key, const size_t nkey) {
uint32_t hv = hash(key, nkey, 0);
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;
}

//扩充hash表为原先的两倍
static void assoc_expand(void) {
old_hashtable = primary_hashtable;
primary_hashtable = calloc(hashsize(hashpower + 1), sizeof(void *));
if (primary_hashtable) {
if (settings.verbose > 1)//debug
fprintf(stderr, "Hash table expansion starting\n");
hashpower++;
expanding = true;
expand_bucket = 0;
pthread_cond_signal(&maintenance_cond);//唤醒扩充线程
} else {
primary_hashtable = old_hashtable;
/* Bad news, but we can keep running. */
}
}

//item插入hash表,并作为对应桶内的head元素
int assoc_insert(item *it) {
uint32_t hv;
unsigned int oldbucket;
assert(assoc_find(ITEM_key(it), it->nkey) == 0);  /* shouldn't have duplicately named things defined */
hv = hash(ITEM_key(it), it->nkey, 0);
if (expanding && (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket)
{
it->h_next = old_hashtable[oldbucket];
old_hashtable[oldbucket] = it;
} else {
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倍的时候,扩充hash表
assoc_expand();
}
MEMCACHED_ASSOC_INSERT(ITEM_key(it), it->nkey, hash_items);
return 1;
}

//从hash表中删除item
void assoc_delete(const char *key, const size_t nkey) {
item **before = _hashitem_before(key, nkey);
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);
}

//是否可以执行扩充操作flag
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. */
pthread_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, 0) & hashmask(hashpower);//hash属于哪个桶
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);
if (settings.verbose > 1)
fprintf(stderr, "Hash table expansion done\n");
}
}
if (!expanding) {
/* We are done expanding.. just wait for next invocation */
pthread_cond_wait(&maintenance_cond, &cache_lock);
}
pthread_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) {
fprintf(stderr, "Can't create thread: %s\n", strerror(ret));
return -1;
}
return 0;
}
//结束线程
void stop_assoc_maintenance_thread() {
pthread_mutex_lock(&cache_lock);
do_run_maintenance_thread = 0;
pthread_cond_signal(&maintenance_cond);
pthread_mutex_unlock(&cache_lock);
/* Wait for the maintenance thread to stop */
pthread_join(maintenance_tid, NULL);
}