BeansDB源码剖析——htree.c

typedef struct t_item Item;
struct t_item {
	//int bucket = item->pos & 0xff; //表示是第几个文件
	//uint32_t pos = item->pos & 0xffffff00; //表示在文件中的位置
	uint32_t pos; 

	//大于0该数据有效，小于0表明无效。
	//ver不会等于0，因此如果set的参数为0时，表示是更新
	//ver不会等于-1，因此set的参数为-1时，表示是删除。
	//ver的更新方法见bitcast.c中的bc_set函数
	//
	int32_t  ver; 

	uint16_t hash; //在bitcask.c的bc_set函数中被赋值
	uint8_t  length; //这个item的长度。通过这个长度找到下一个item
	char     name[1]; 
};

 

/*
*  Beansdb - A high available distributed key-value storage system:
*
*      http://beansdb.googlecode.com
*
*  Copyright 2009 Douban Inc.  All rights reserved.
*
*  Use and distribution licensed under the BSD license.  See
*  the LICENSE file for full text.
*
*  Authors:
*      Davies Liu <[e-mail]>
*/

#include <stdio.h>
#include <errno.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <pthread.h>

#include "fnv1a.h"
#include "htree.h"
#include "codec.h"
#include "quicklz.h"

//key的最大长度
const int MAX_KEY_LENGTH = 200;
//Bucket里存放节点。
//一个非数据节点分成16个bucket（就是子树），每个bucket是另一个节点
//这个跟bitcask的bucket是不同的
const int BUCKET_SIZE = 16;
//非数据节点中的count超过此限制要分裂
const int SPLIT_LIMIT = 32; 
//树的最大深度
const int MAX_DEPTH = 8; 
//g_index[i] = (g_index[i-1] << 4) + g_index[i-1];
//g_index[i]表示的是第i层前共有多少个节点
//HTree最多有410338673个节点
static const int g_index[] = {0, 1, 17, 289, 4913, 83521, 1419857, 24137569, 410338673};

#define max(a,b) ((a)>(b)?(a):(b))
//it是第几个孩子节点,0x0f说明最多有16个孩子节点
#define INDEX(it) (0x0f & (keyhash >> ((7 - node->depth - tree->depth) * 4)))
//length包含了item的size，由于结构的最后一个是变长数组，所以多减了一个字符，需要加上。
#define KEYLENGTH(it) ((it)->length-sizeof(Item)+ITEM_PADDING)
//如果it不是一个有效的节点，该宏返回0
#define HASH(it) ((it)->hash * ((it)->ver>0))


//每个节点有16个孩子，可以通过INDEX宏得到该孩子的位置
//如果这个node里并没有数据，那么这个node只占64比特，参见clear函数
//否则，这个node存储数据Data，Data里有多个item，通过item的长度找到下一个item的地址
//树的数据节点不一定是在同一层的
typedef struct t_data Data;
struct t_data {
	int size; //总大小
	int used; //已用大小
	int count;//item的个数
	Item head[0];
};

typedef struct t_node Node;
struct t_node {
	uint16_t is_node:1; //=1，说明这个node里没有存放数据，否则就是存放了数据的
	uint16_t valid:1; //说明这个节点是有效的，即它与它的所有子节点都没有被改动过
	uint16_t depth:4; //该节点的深度
	uint16_t compressed:1; //是否被压缩了
	uint16_t flag:9; 
	uint16_t hash; //哈希值，如果这个节点是非数据节点，这个值是所有子节点的哈希值之和
	//如果是数据节点，这个值是所有ver大于0的item的哈希值的和
	uint32_t count; //所有子节点里ver>0的item的个数
	Data *data;
};

//node的count和data的count是不一样的。
//data的count表示有多少个item,node的count表示有多少个item的ver是>0的。
//htree是一块连续的内存，相当于使用数组存放一个二叉树。
struct t_hash_tree {
	int depth; //
	int height; //depth = hight-1
	Node *root;
	int pool_size; //节点的数目
	pthread_mutex_t lock;
	char buf[512];

	bool compress; //是否压缩
	char wbuf[QLZ_SCRATCH_COMPRESS];
	char cbuf[1024 * 10];
};


// forward dec
static void add_item(HTree *tree, Node *node, Item *it, uint32_t keyhash, bool enlarge);
static void remove_item(HTree *tree, Node *node, Item *it, uint32_t keyhash);
static void split_node(HTree *tree, Node *node);
static void merge_node(HTree *tree, Node *node);
static void update_node(HTree *tree, Node *node);

//node在这一层的位置
inline uint32_t get_pos(HTree *tree, Node *node)
{
	return (node - tree->root) //node在整个pool中的位置
		- g_index[(int)node->depth]; //node的上一层总共有多少个节点
}

//找到node的第b个孩子节点，b从0开始
//一个节点最多有BUCKET_SIXZE=16（1<<4）个孩子节点
inline Node *get_child(HTree *tree, Node *node, int b)
{
	int i = g_index[node->depth + 1] //node的孩子节点这一层之前总共有多少个节点
		+ (get_pos(tree, node) << 4) //node的第1个孩子节点的位置 
		+ b;
	return tree->root + i;
}

//有一个get_data，就要有一个free_data或者set_data与之对应
inline Data* get_data(Node *node)
{
	if (node->compressed) {
		char wbuf[QLZ_SCRATCH_DECOMPRESS];
		Data *d = malloc(qlz_size_decompressed((char*)node->data));
		qlz_decompress((char*)node->data, (char*)d, wbuf);
		return d;
	} else {
		return node->data;
	}
}

//saved，节约的意思,通过压缩节省了多少空间
static int saved = 0;

//设置node的data，其中包括了压缩
inline void set_data(HTree *tree, Node *node, Data *data)
{
	if (data != node->data) {
		if (node->data) free(node->data);
		if (tree->compress) {
			int need = data->used + 400;
			char *cbuf = need <= sizeof(tree->cbuf) ? tree->cbuf : malloc(need);
			size_t size = qlz_compress((char*)data, cbuf, data->used, tree->wbuf);
			saved += data->size - size;
			free(data);
			data = malloc(size);
			memcpy(data, cbuf, size);
			node->compressed = 1;
			if (cbuf != tree->cbuf) free(cbuf);
		}
		node->data = data;
	}
}

//这是与get_data相对应的，由于get_data时可能会有压缩，所以get_data并不一定是返回的原data
//可能是原data的一个copy，这样通过判断data != node->data，来删除这个copy
inline void free_data(Node *node, Data *data)
{
	if (data != node->data) {
		free(data);
	}
}

//注意key_hash产生的hash跟Item中的hash是不一样的
//这里的hash是为了便于在htree中查找。
inline uint32_t key_hash(Item* it)
{
	char buf[255];
	//由于有对key的编码，所以要先解码，才能取哈希值
	int n = dc_decode(buf, it->name, KEYLENGTH(it));
	//哈希函数
	return fnv1a(buf, n);
}

static Item* create_item(HTree *tree, const char* name, uint32_t pos, uint16_t hash, int32_t ver)
{
	Item *it = (Item*)tree->buf;
	it->pos = pos;
	it->ver = ver;
	it->hash = hash;
	//对key进行编码
	int n = dc_encode(it->name, name, strlen(name));
	it->length = sizeof(Item) + n - ITEM_PADDING;
	return it;
}

//树再高一层，树是一段连续的空间。
static void enlarge_pool(HTree *tree)
{
	int i;
	int old_size = tree->pool_size;
	int new_size = g_index[tree->height + 1];

	tree->root = (Node*)realloc(tree->root, sizeof(Node) * new_size);
	memset(tree->root + old_size, 0, sizeof(Node) * (new_size - old_size));
	//更新所有新增的树节点的高度
	for (i=old_size; i<new_size; i++){
		tree->root[i].depth = tree->height;
	}

	tree->height ++;
	tree->pool_size = new_size;
}

//将node置为初始状态
static void clear(HTree *tree, Node *node)
{
	if (node->data) free(node->data);
	//这个node占64位
	node->data = (Data*) malloc(64);
	node->data->size = 64;
	node->data->used = sizeof(Data);
	node->data->count = 0;

	node->is_node = 0;
	node->valid = 1;
	node->compressed = 0;
	node->count = 0;
	node->hash = 0;
}

//将it插入到树中node节点开始的位置
//1.找到这个node下面的数据节点
//2.数据节点中存放的是Item组成的数组，根据Item的length域遍历这个数据节点的信息
//3.接下来就相当于数组的插入了，更新数据节点的count域和hash域
//	3.1.如果找到了相同的key，那么更新这个Item
//	3.2.否则将it放入到数组的末尾，更新Data的used域
//4.如果是插入，则有可能造成count的扩大，需要对数据节点进行分裂。
//	4.1.数据节点在树的最底层，那么允许一个数据节点存储的Item的个数为LIMIT*4，
//			这是为了防止enlarge_pool造成过多内存的使用
//	4.2.数据节点在树的中间部分，也就是说数据节点下面还有节点
//			那么为了使查找更有效率，需要尽量减少数据节点中Item的个数，超过LIMIT就要分裂
static void add_item(HTree *tree, Node *node, Item *it, uint32_t keyhash, bool enlarge)
{
	//1
	//由于对数据节点进行了变更，所以要把所走过的路径中的所有节点的valid设为0
	//这样update_node时就可以根据valid值决定是否要更新此节点及它的子节点
	while (node->is_node) {
		node->valid = 0;
		node = get_child(tree, node, INDEX(it));
	}

	//2
	//得到这个数据节点的数据
	Data *data = get_data(node);
	//第一个item
	Item *p = data->head;
	int i;
	//node的count和data的count是不一样的。
	for (i=0; i<data->count; i++){
		//3.1
		if (it->length == p->length && 
			memcmp(it->name, p->name, KEYLENGTH(it)) == 0)
		{//key相同
				//减去原来item的哈希值，加上新的item的哈希值
				node->hash += (HASH(it) - HASH(p)) * keyhash;
				//更新该节点的count
				node->count += it->ver > 0;
				node->count -= p->ver > 0;
				//只要name的长度一样，两个Item的length就是一样的
				//不会对旁边Item的内容产生影响，直接cpy
				memcpy(p, it, sizeof(Item));
				//data会在set_data中被free
				set_data(tree, node, data);
				return;
		}
		p = (Item*)((char*)p + p->length);
	}

	//3.2
	//如果data的size不够用了，就要malloc，新建一个data空间,至少要比64大
	if (data->size < data->used + it->length){
		int size = max(data->used + it->length, data->size + 64);
		//item的位置
		int pos = (char*)p-(char*)data;
		Data *new_data = (Data*) malloc(size);
		memcpy(new_data, data, data->used);
		//与get_data对应
		free_data(node, data);
		data = new_data;
		data->size = size;
		//保存item要插入的位置
		p = (Item *)((char*)data + pos);
	}

	//p所指就是it要插入的地方
	memcpy(p, it, it->length);
	//data的item的数量加1
	data->count ++;
	data->used += it->length;
	//如果item的ver大于0，node的count加1
	node->count += it->ver > 0;
	//node的哈希值是所有数据子节点的哈希值之和，更新
	node->hash += keyhash * HASH(it);
	set_data(tree, node, data);

	//4
	//node是数据节点
	if (node->count > SPLIT_LIMIT){
		//这个node是树的最底层
		if (node->depth == tree->height - 1){//4.1
			//如果这个数的在树的最底层，就要*4，防止频繁地enlarge造成空间太大
			if (enlarge && node->count > SPLIT_LIMIT * 4){
				int pos = node - tree->root;
				//树的高度加深
				enlarge_pool(tree);
				node = tree->root + pos; // reload
				//分裂节点
				split_node(tree, node);
			}
			
		}else{//4.2
			//分裂节点
			split_node(tree, node);
		}
	}
}

//将node中的数据分发到它的下一层孩子节点中
//完成后，这个节点就变成了一个普通的节点，里面没有数据；它的16个孩子成为新的数据节点
//1.得到node的孩子节点，并reset
//2.根据哈希值将数据放入对应的孩子节点
//3.更新node对应的域
static void split_node(HTree *tree, Node *node)
{
	//1
	//得到这个节点的第一个孩子
	Node *child = get_child(tree, node, 0);
	int i;
	//把所有的孩子节点都清空
	for (i=0; i<BUCKET_SIZE; i++){
		clear(tree, child+i);
	}

	//2
	Data *data = get_data(node);
	Item *it = data->head;
	//把这个数据节点的所有item放入它的孩子节点中
	for (i=0; i<data->count; i++) {
		int32_t keyhash = key_hash(it);
		//添加item
		add_item(tree, child + INDEX(it), it, keyhash, false);
		it = (Item*)((char*)it + it->length);
	}

	//3
	free_data(node, data);
	free(node->data);
	node->data = NULL;

	//这个节点变为普通节点
	node->is_node = 1;
	//这个节点更改了，update_node的时候就要更新这个节点的哈希值
	node->valid = 0;
}

//移除一个Item
//1.找到数据节点
//2.在Data中查找对应的Item
//3.删除之，并更新数据节点对应的域
static void remove_item(HTree *tree, Node *node, Item *it, uint32_t keyhash)
{
	//1
	//由于对数据节点进行了变更，所以要把所走过的路径中的所有节点的valid设为0
	//这样update_node时就可以根据valid值决定是否要更新此节点及它的子节点
	while (node->is_node) {
		node->valid = 0;
		node = get_child(tree, node, INDEX(it));
	}

	//2
	Data *data = get_data(node);
	if (data->count == 0) return ;
	Item *p = data->head;
	int i;
	for (i=0; i<data->count; i++){
		if (it->length == p->length && 
			memcmp(it->name, p->name, KEYLENGTH(it)) == 0){
			//3
			data->count --;
			data->used -= p->length;
			node->count -= p->ver > 0;
			node->hash -= keyhash * HASH(p);
			//将it删除，后面的移动过来                               
			memcpy(p, (char*)p + p->length, 
				data->size - ((char*)p - (char*)data) - p->length);
			set_data(tree, node, data);
			return;
		}
		//否则检查下一个item
		p = (Item*)((char*)p + p->length);
	}
	//与get_data对应
	free_data(node, data);
}

//将数据节点node的孩子节点中的数据放入到node中，ver<0的节点则被抛弃
//这样node成为数据节点，它的孩子节点成为普通节点,减少数据的分散性
//1.reset node节点
//2.遍历每个孩子节点的Data，将其中ver>0的Item放入到node中
//3.reset 这个孩子节点
static void merge_node(HTree *tree, Node *node)
{
	//1
	clear(tree, node);

	//2
	Node* child = get_child(tree, node, 0);
	int i, j;
	//将node所有孩子节点的item都集中到自己身上
	//同时删除了ver小于0的item
	for (i=0; i<BUCKET_SIZE; i++){
		Data *data = get_data(child+i); 
		Item *it = data->head;
		//count计算不对，(child+i)->count是有效Item的大小
		//这里应该是int count = data->count因为要遍历的是所有的item
		int count = (child+i)->count;
		for (j=0; j < count; j++){
			if (it->ver > 0) {//只merge那些ver>0的
				add_item(tree, node, it, key_hash(it), false);
			} // drop deleted items, ver < 0
			it = (Item*)((char*)it + it->length);
		}
		free_data(child+i, data);
		//3
		clear(tree, child + i);
	}
}

//递归更新HTree中每个Node的hash和count域，将更新完成的Node的valid域设置为1
static void update_node(HTree *tree, Node *node)
{
	//这个节点及它的所有子节点都没有被改动过
	//就没有必要更新这个节点的哈希值
	if (node->valid) return ;

	int i;
	node->hash = 0;
	//只更新普通节点的哈希，数据节点的哈希在add_item的时候已经计算过了，它永远是valid的
	if (node->is_node){
		Node *child = get_child(tree, node, 0);
		node->count = 0;
		//递归遍历所有的子节点，得到它们的有效item的数目
		for (i=0; i<BUCKET_SIZE; i++){
			update_node(tree, child+i);
			node->count += child[i].count;
		}
		//遍历孩子节点，更新node的哈希值
		for (i=0; i<BUCKET_SIZE; i++){
			if (node->count > SPLIT_LIMIT * 4){ 
				node->hash *= 97;               
			}
			node->hash += child[i].hash;
		}
	}
	node->valid = 1;

	// merge nodes
	//如果有必要，merge
	if (node->count <= SPLIT_LIMIT) {
		merge_node(tree, node);
	}
}

//通过哈希得到item
static Item* get_item_hash(HTree* tree, Node* node, Item* it, uint32_t keyhash)
{
	//找到存储该item的实际节点
	while (node->is_node) {
		node = get_child(tree, node, INDEX(it));
	}

	Data *data = get_data(node);
	Item *p = data->head, *r = NULL;
	int i;
	//在数据节点中找到该item
	for (i=0; i<data->count; i++){
		if (it->length == p->length && 
			memcmp(it->name, p->name, KEYLENGTH(it)) == 0){
				r = p;
				break;
		}
		p = (Item*)((char*)p + p->length);
	}
	free_data(node, data);
	return r;
}

//十六进制到十进制的转换
inline int hex2int(char b)
{
	if (('0'<=b && b<='9') || ('a'<=b && b<='f')) {
		return (b>='a') ?  (b-'a'+10) : (b-'0');
	}else{
		return -1;
	}
}

//dir所表示的节点的哈希值之和
//dir[i]表示的是从node开始的第i层的节点的子节点的位置
//dir的len最多只有tree->hight
//将node的ver>0的item个数存储在count中
static uint16_t get_node_hash(HTree* tree, Node* node, const char* dir, 
															int *count)
{
	if (node->is_node && strlen(dir) > 0){
		char i = hex2int(dir[0]);
		if (i >= 0) {
			return get_node_hash(tree, get_child(tree, node, i), dir+1, count);
		}else{
			if(count) *count = 0;
			return 0;
		}
	}
	//更新dir对应的node的哈希，并返回
	update_node(tree, node);
	if (count) *count = node->count;
	return node->hash;
}

//找到前缀是prefix的item
static char* list_dir(HTree *tree, Node* node, const char* dir, const char* prefix)
{
	int dlen = strlen(dir); 
	//直到它的最后一个孩子节点（dlen == 0），或者这个孩子节点已经是叶子节点了（!is_node）
	while (node->is_node && dlen > 0){
		int b = hex2int(dir[0]);
		if (b >=0 && b < 16) {
			node = get_child(tree, node, b);
			dir ++;
			dlen --;
		}else{
			return NULL;
		}
	}

	int bsize = 4096;
	char *buf = (char*) malloc(bsize);
	memset(buf, 0, bsize);
	int n = 0, i, j;
	//如果dir的最后一个节点不是数据节点
	if (node->is_node) {
		//更新这个节点，得到它最新的count值和哈希值
		update_node(tree, node);

		//找到它的子节点
		Node *child = get_child(tree, node, 0);
		//if (node->count > 100000 || (prefix==NULL && node->count > 128)) {
		//&&的优先级高？
		if (node->count > 100000 || prefix==NULL && node->count > 128) {
			//把它的所有子节点都打印出来
			for (i=0; i<BUCKET_SIZE; i++) {
				Node *t = child + i;
				n += snprintf(buf + n, bsize - n, "%x/ %u %u\n", 
					i, t->hash, t->count);
			}
		}else{
			for (i=0; i<BUCKET_SIZE; i++) {
				//找到这个孩子节点的孩子节点.dir = ""说明直接找到它的孩子节点
				char *r = list_dir(tree, child + i, "", prefix);
				if (bsize - n < strlen(r)) {
					buf = (char*)realloc(buf, bsize * 2);
					bsize *= 2;
				}
				n += sprintf(buf + n, "%s", r);
				free(r);
			}
		}
	}else{//如果dir的最后一个节点是数据节点
		Data *data = get_data(node); 
		Item *it = data->head;
		char pbuf[20], name[255];
		int prefix_len = 0;
		if (prefix != NULL) prefix_len = strlen(prefix);
		for (i=0; i<data->count; i++, it = (Item*)((char*)it + it->length)){
			if (dlen > 0){
				sprintf(pbuf, "%08x", key_hash(it));
				if (memcmp(pbuf + tree->depth + node->depth, dir, dlen) != 0){
					continue;
				}
			}
			int l = dc_decode(name, it->name, KEYLENGTH(it));
			if (prefix == NULL || l >= prefix_len 
				&& strncmp(name, prefix, prefix_len) == 0) {
				n += snprintf(buf+n, bsize-n-1, "%s %u %d\n", name, it->hash, it->ver);
				if (bsize - n < 200) {
					buf = (char*)realloc(buf, bsize * 2);
					bsize *= 2;
				}
			}
		}
		free_data(node, data);
	}
	return buf;
}

//遍历HTree
static void visit_node(HTree *tree, Node* node, fun_visitor visitor, void* param)
{
	int i;
	//如果不是叶子节点，则向下寻找叶子节点
	if (node->is_node){
		Node *child = get_child(tree, node, 0);
		for (i=0; i<BUCKET_SIZE; i++){
			visit_node(tree, child+i, visitor, param);
		}
	}else{//是叶子节点，可以遍历里面的数据
		Data *data = get_data(node);
		Item *p = data->head;
		Item *it = (Item*)tree->buf;
		for (i=0; i<data->count; i++){
			memcpy(it, p, sizeof(Item));
			dc_decode(it->name, p->name, KEYLENGTH(p));
			it->length = sizeof(Item) + strlen(it->name) - ITEM_PADDING;
			//对it进行操作
			visitor(it, param);
			p = (Item*)((char*)p + p->length);
		}
		free_data(node, data);
	}    
}

static void optimize_node(HTree *tree, Node* node)
{
	int i;
	if (node->is_node){
		Node *child = get_child(tree, node, 0);
		for (i=0; i<BUCKET_SIZE; i++){
			optimize_node(tree, child+i);
		}
	}else{
		if (!node->compressed && node->data->size > 64) {
			Data *data = get_data(node);
			node->data = NULL;
			set_data(tree, node, data);
		}
	}    
}

/*
* API
*/

HTree* ht_new(int depth)
{
	HTree *tree = (HTree*)malloc(sizeof(HTree));
	if (!tree) return NULL;
	memset(tree, 0, sizeof(HTree));
	tree->depth = depth;   
	tree->height = 1;
	tree->compress = false;

	int pool_size = g_index[tree->height];
	Node *root = (Node*)malloc(sizeof(Node) * pool_size);
	if (!root){
		free(tree);
		return NULL;
	}
	memset(root, 0, sizeof(Node) * pool_size);

	// init depth
	int i,j;
	for (i=0; i<tree->height; i++){
		for (j=g_index[i]; j<g_index[i+1]; j++){
			root[j].depth = i;
		}
	}

	tree->root = root;
	tree->pool_size = pool_size;
	clear(tree, tree->root);

	pthread_mutex_init(&tree->lock, NULL);
	dc_init(NULL);

	return tree;
}

void ht_destroy(HTree *tree)
{
	if (!tree) return;

	pthread_mutex_lock(&tree->lock);

	int i;
	for(i=0; i<tree->pool_size; i++){
		if (tree->root[i].data) free(tree->root[i].data);
	}
	free(tree->root);
	free(tree);
}

inline uint32_t keyhash(const char *s)
{
	return fnv1a(s, strlen(s));
}

void ht_add(HTree *tree, const char* name, uint32_t pos, uint16_t hash, int32_t ver)
{
	if (!tree || !name) return;
	if (name[0] <= 32 || strlen(name) > MAX_KEY_LENGTH){
		fprintf(stderr, "bad key\n");
		return;
	}

	pthread_mutex_lock(&tree->lock);
	//注意item中的hash跟add_item用的hash是不同的变量
	Item *it = create_item(tree, name, pos, hash, ver);
	add_item(tree, tree->root, it, keyhash(name), true);
	pthread_mutex_unlock(&tree->lock);
}

void ht_remove(HTree* tree, const char *name)
{
	if (!tree || !name) return;
	if (name[0] <= 32 || strlen(name) > MAX_KEY_LENGTH){
		fprintf(stderr, "bad key\n");
		return;
	}

	pthread_mutex_lock(&tree->lock);
	Item *it = create_item(tree, name, 0, 0, 0);
	remove_item(tree, tree->root, it, keyhash(name));
	pthread_mutex_unlock(&tree->lock);
}

//返回一个新的item
Item* ht_get(HTree* tree, const char* key)
{
	if (!tree || !key || key[0] <= 0 || strlen(key) > MAX_KEY_LENGTH) {
		return NULL;
	}

	pthread_mutex_lock(&tree->lock);
	Item *it = create_item(tree, key, 0, 0, 0);
	Item *r = get_item_hash(tree, tree->root, it, keyhash(key));
	if (r != NULL){
		Item *rr = (Item*)malloc(sizeof(Item) + strlen(key) + 1);
		memcpy(rr, r, sizeof(Item));
		memcpy(rr->name, key, strlen(key) + 1);
		r = rr; // r is in node->Data block 
	}
	pthread_mutex_unlock(&tree->lock);
	return r;   
}

uint32_t ht_get_hash(HTree* tree, const char* key, int* count)
{
	if (!tree || !key || strlen(key) > 8) {
		if(count) *count = 0;
		return 0;
	}

	uint32_t hash = 0;
	pthread_mutex_lock(&tree->lock);

	update_node(tree, tree->root);

	if (key[0] == '@'){
		hash = get_node_hash(tree, tree->root, key+1, count);
	}
	pthread_mutex_unlock(&tree->lock);
	return hash;
}

char* ht_list(HTree* tree, const char* dir, const char* prefix)
{
	if (!tree || !dir || strlen(dir) > 8) return NULL;
	if (prefix != NULL && strlen(prefix) == 0) prefix = NULL;

	pthread_mutex_lock(&tree->lock);
	char* r = list_dir(tree, tree->root, dir, prefix);
	pthread_mutex_unlock(&tree->lock);

	return r;
}

void ht_optimize(HTree *tree)
{
	pthread_mutex_lock(&tree->lock);
	tree->compress = true;
	optimize_node(tree, tree->root);
	pthread_mutex_unlock(&tree->lock);

	fprintf(stderr, "%dM bytes saved\n", saved >> 20);
}

void ht_visit(HTree *tree, fun_visitor visitor, void *param)
{
	pthread_mutex_lock(&tree->lock);
	visit_node(tree, tree->root, visitor, param);
	pthread_mutex_unlock(&tree->lock);  
}

HTree是一个16叉树，需要注意对子节点下标的确定，以及对状态节点，数据节点的结构理解。