点击下载
本文文档

当前位置：首页 - 科技 - 知识百科 - 正文

memcached源码阅读之hashtable

来源：动视网责编：小采时间：2020-11-09 12:59:13

memcached源码阅读之hashtable

memcached源码阅读之hashtable:前言昨晚用一个小时把 memcached 的服务端程序看了，发现踩到一个坑，大部分程序都在实现服务器的网络编程的部分。而我是不懂网络编程的，于是又花了半个小时去找 memcached 的储存代码，发现时使用 hash table 储存的。于是这里研究一下 memcach

推荐度：

点击下载本文 文档为doc格式

导读memcached源码阅读之hashtable:前言昨晚用一个小时把 memcached 的服务端程序看了，发现踩到一个坑，大部分程序都在实现服务器的网络编程的部分。而我是不懂网络编程的，于是又花了半个小时去找 memcached 的储存代码，发现时使用 hash table 储存的。于是这里研究一下 memcach

前言昨晚用一个小时把 memcached 的服务端程序看了，发现踩到一个坑，大部分程序都在实现服务器的网络编程的部分。而我是不懂网络编程的，于是又花了半个小时去找 memcached 的储存代码，发现时使用 hash table 储存的。于是这里研究一下 memcached 的 ha

前言

昨晚用一个小时把 memcached 的服务端程序看了，发现踩到一个坑，大部分程序都在实现服务器的网络编程的部分。
而我是不懂网络编程的，于是又花了半个小时去找 memcached 的储存代码，发现时使用 hash table 储存的。
于是这里研究一下 memcached 的 hash table .
昨晚记录的 memcached 源码阅读之原理篇最后说了，服务器端做两部分：一部分是网络编程方面。另一部分就是 hash table 的实现。
刚好前几天我自己实现过一个 hash table 研究与实现, 于是 memcached 的 hash table 可以很快的看完并理解。
这里就简单的讲解一下 memcached 的 hash table .

我的 hash table

大家可以先阅读一下我之前的 hash table 研究与实现。
大家可以看出来，我的这个 hash table 实现的太简单了，所以没有实用价值的，只能用来学习 hash table 。
这里我先说一下我的这个 hash table 需要加强之处吧。

Node(节点) 与 bucket(桶)

先熟悉一下我的 Node 节点和桶。

class Node {
public:
 Node():next(-1) {
 }
 int next;
 int pos;
};
Node * hash_bucket ; //指向桶内存的指针

val 的改造

首先我的 val 是一个对象，且所有的 val 必须有相同的内存大小。
这往往是不现实的，所以一个很简单的改造就是 val 修改为动态申请内存，然后用指针来储存在 Node 里面即可。

class Node {
public:
 Node():next(-1) {
 }
 int next;
 int pos;
 void* val;
 int valLen;
};

key 改造

我这里是没有 key 的，所以 key 是由 val hash 得到，但是实际中往往是 key-value 的储存方式，所以实际中是有个 key 的。
不过我的那个程序很容易就加入 key.
由于之后还有 key 的比较，且 key 的长度也是未知，所以这里我们就需要也把 key 储存在 Node 里面了。

class Node {
public:
 Node():next(-1) {
 }
 int next;
 int pos;
 void* val;
 int valLen;
 void* key;
 int keyLen;
};

内存管理

大家还可以明显看到，我的内存管理其实更不现实。
使用前是根本不知道我们会插入多少个加点，所以这个内存需要时动态增加的。
关于这个实现，一会我们看 memcached 里面的实现方式吧。

其他改造

改造之后，发现我的桶没有必要是 Node 节点，也就是链表不需要一个头，所以这个桶需要改造。
当然链表又没有头的学问也很大，大家可以自己研究一番。

Node** hash_bucket;

memcached 的 hash table

头文件认识

memcached 的 hash table 有两个文件，一个 .c, 一个 .h .
这是他们的位置 assoc.h 和 assoc.c .

在 .h 中，代码就不足 10 行，可以看一下。后两个是我自己加上的。

/* associative array */
void assoc_init(const int hashpower_init); //初始化 hash table
item *assoc_find(const char *key, const size_t nkey, const uint32_t hv); //查找
int assoc_insert(item *item, const uint32_t hv);//插入
void assoc_delete(const char *key, const size_t nkey, const uint32_t hv);//删除
void do_assoc_move_next_bucket(void);//
int start_assoc_maintenance_thread(void); //开启一个线程来合并两个hash table
void stop_assoc_maintenance_thread(void); //结束线程
extern unsigned int hashpower; //hash table
static void *assoc_maintenance_thread(void *arg);//合并代码
static void assoc_expand(void);//自动扩大hash table 的大小，有两个hash table.

核心变量介绍

变量的介绍详见我的注释

//hash table 的桶大小永远是2的倍数，且按2的倍数扩增
unsigned int hashpower = HASHPOWER_DEFAULT; 
// 桶的大小
#define hashsize(n) ((ub4)1<<(n))
// 主要用于取模 a % (2^n) = a & (2^n - 1)
#define hashmask(n) (hashsize(n)-1)
// 当前桶的指针
static item** primary_hashtable = 0;
//桶扩增后，旧的桶的指针
static item** old_hashtable = 0;
//目前桶内的节点个数
static unsigned int hash_items = 0;
//旧桶数据是否清理
static bool expanding = false;
//是否开始扩增桶
static bool started_expanding = false;
//旧桶清理到那个位置，每次清理 2^expand_bucket 个，expand_bucket 递增
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);
 }
}

查找

由于 key-value 的储存需要保证相同的key 对应唯一的 value.
这样也就代表一个 key 是唯一的，就和我的 hash table 的 val 是一个性质的。
nkey 代表这个 key 的字符串长度。
hv 代表这个字符串 key 的 hash 值。

expanding 代表是否有旧桶，有的话我们需要先判断当前 key 是在新桶还是旧桶里面。
怎么判断呢？
新桶范围是 0~ 2^hashpower，插入到新桶的值的范围是 0 ~ expand_bucket
旧桶范围是 0 ~ 2^(hashpower - 1),旧桶的值范围是 expand_bucket ~ 2^(hashpower - 1)

这时可能就会有人说不对呀，那对于 2^(hashpower - 1) ~ 2^hashpower 的数据在哪呢？
其实，那些数据超过了 2^(hashpower - 1)，所以会进行取模，这样就还在那个范围了。
什么意思呢？
对于新来的数据，只看范围，如果在 expand_bucket ~ 2^(hashpower - 1)，即使有新桶还会存在旧桶里。

it 指针指向当前 key 对应的桶的位置。
然后就可以循环判断了。
由于是内存比较，所以需要先比较长度，再比较内存，完全相同了就找到了。

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)
 {
 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;
 }
 return ret;
}

添加

添加比较简单，实现方式和我的差不多，插在链表头部。
这里多了一步桶大小的检测，节点个数超过当前桶大小的 1.5 倍时就增大桶(调用启动增大桶线程)。这里要注意的一点是对于插入的key,已经在其他地方检察过是否存在了。
意思就是这里保证一定不存在。

int assoc_insert(item *it, const uint32_t hv) {
 unsigned int oldbucket;
 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) {
 assoc_start_expand();
 }
 return 1;
}

删除

删除也比较简单，先找到需要删除的那个节点的父节点，然后删除即可。
需要注意的是这里也是保证要删除的节点已经存在。
另外大家不能理解的是为什么要用指向指针的指针。
这个问题曾经在 segmentfault 上有人问过这个问题，不过他那个问题就没有办法使用指向指针的指针了。

问题就是，如果不使用指向指针的指针，查找的节点不在第一个位置的话，可以正常操作。
但是在第一个位置的话，我们的操作不会生效的。
如果你不能明白的话，先去 segmentfault 研究一下那个问题，那个明白了，这个就明白了。

void assoc_delete(const char *key, const size_t nkey, const uint32_t hv) {
 item **before = _hashitem_before(key, nkey, hv);
 if (*before) {
 item *nxt;
 hash_items--;
 nxt = (*before)->h_next;
 (*before)->h_next = 0; 
 *before = nxt;
 return;
 }
}

父节点的指针

实现方式和普通的查找类似，只是使用指向指针的指针。

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_start_expand(void) {
 if (started_expanding)
 return;
 started_expanding = true;
 pthread_cond_signal(&maintenance_cond);
}

线程控制

线程控制做两件事：启动线程和停止线程。
线程主要执行 assoc_maintenance_thread 函数。

static pthread_t maintenance_tid;
int start_assoc_maintenance_thread() {
 int ret;
 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() {
 mutex_lock(&cache_lock);
 do_run_maintenance_thread = 0;
 pthread_cond_signal(&maintenance_cond);
 mutex_unlock(&cache_lock);
 pthread_join(maintenance_tid, NULL);
}

线程做的事 - 合并桶

这个合并桶的线程只做两件事。
第一件事是桶内的节点一个一个的合并，合并完了回收旧桶。
第二事是旧桶释放后，开始监听是否增大桶，接收到信号，调用增大桶 assoc_expand 函数。

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;
 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);
 }
 }
 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 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;
 } else {
 primary_hashtable = old_hashtable;
 }
}

memcached 的 hash table 的评价

看到这里，大家的心理可能会想几件事
1. 节点的管理怎么没有呢？
2. 简单的几个函数竟然实现了这个复杂功能？

我猜想这就是软件设计的强大之处。
节点管理从 hash table 里面抽象出去，你只需要给我传一个指向那个节点的指针，且那个节点有key的信息和下一个节点的指针即可。
另外对于面向过程的编程方式，想编出好的代码，就要学会函数封装。
每个函数都实现一个简单的功能，合起来就是很强大的功能了。
就像在 acm 比赛的时候，对于模拟题，我就感觉很简单 -- 不就是若干简单函数的实现嘛！

写到这文章也要结束了，打个广告，大家留意到了吗？
没上面没有提起对 key 的 hash 方法，那是因为对字符串的hash 又是一个很大的学问，我下篇文章慢慢介绍。

这篇文章之所以介绍的这个详细，是为了补偿昨晚写的那两篇没有价值的记录《memcached 源码阅读之原理篇》和《memcached 源码阅读之库函数介绍》。
《完》

本文出自：http://tiankonguse.github.io, 原文地址：http://github.tiankonguse.com//blog/2014/11/07/memcached-hash-table/, 感谢原作者分享。