skynet源码分析（3）--消息名字和ID之handle
2017-08-19 本文已影响188人
天一阁图书管理员
作者:shihuaping0918@163.com，转载请注明作者
消息机制是skynet中一个比较难理解的东西，在开始分析代码前，建议先读一下云风的博客和skynet的WIKI。SKYNET设计综述：http://blog.codingnow.com/2012/09/the_design_of_skynet.html，CLUSTER：https://github.com/cloudwu/skynet/wiki/Cluster。
SKYNET设计综述讲到模块被称为服务。“服务间可以自由发送消息。每个模块可以向 Skynet 框架注册一个 callback 函数，用来接收发给它的消息。”还提到“把一个符合规范的 C 模块，从动态库（so 文件）中启动起来，绑定一个永不重复（即使模块退出）的数字 id 做为其 handle 。Skynet 提供了名字服务，还可以给特定的服务起一个易读的名字，而不是用 id 来指代它。id 和运行时态相关，无法保证每次启动服务，都有一致的 id ，但名字可以。”今天要分析的两个文件skynet_handle.c和skynet_handle.h就是实现名字服务的。
而WIKI中的CLUSTER讲到的是harbor相关的内容，“每个 skynet 服务都有一个全网唯一的地址，这个地址是一个 32bit 数字，其高 8bit 标识着它所属 slave 的号码。即 harbor id 。在 master/slave 网络中，id 为 0 是保留的。所以最多可以有 255 个 slave 节点。”
前面写这么一大段东西都是分析代码所需要的，不是为了凑字数，也不是为了别的原因。下面开始分析代码。
#include "skynet.h"
#define DEFAULT_SLOT_SIZE 4
#define MAX_SLOT_SIZE 0x40000000
//名字服务结构，一个名字对应一个handle
struct handle_name {
char * name; //服务名字
uint32_t handle; //服务ID，下面以handle来称呼
};
//保存handle/name列表的数据结构，skynet_handle_init会初始化它
struct handle_storage {
struct rwlock lock;
uint32_t harbor;//这就是wiki里提到的harbor!!
uint32_t handle_index; //必须从1开始
int slot_size; //数组长度
struct skynet_context ** slot; //数组，实际上里面存的是服务的上下文
int name_cap; //容量
int name_count; //长度或者说个数
struct handle_name *name; //数组
};
static struct handle_storage *H = NULL;
//注册服务，返回给它一个handle
uint32_t
skynet_handle_register(struct skynet_context *ctx) {
struct handle_storage *s = H;
rwlock_wlock(&s->lock);
//死循环，所以服务数量如果太大了你懂得
for (;;) {
int i;
for (i=0;i<s->slot_size;i++) { //遍历服务列表，找个空位
uint32_t handle = (i+s->handle_index) & HANDLE_MASK; //只取后24位
int hash = handle & (s->slot_size-1); //&操作符，就是取到一个小于s->slot_size-1的值，slot_size-1源码下方有详细解释。
if (s->slot[hash] == NULL) { //没有hash碰撞，好巧
s->slot[hash] = ctx;
s->handle_index = handle + 1; //handle_index增加了
rwlock_wunlock(&s->lock);
handle |= s->harbor; // 位操作或，把harbor id附上去，实在不理解你就把它当加法吧
return handle;
}
}
//不幸的事情发生了，一直都在HASH碰撞，也就是说坑已经填满了
assert((s->slot_size*2 - 1) <= HANDLE_MASK); //slot_size是不是达到0xffffff+1的一半了？
//既然坑填满了我们就扩容一下吧，扩大一倍
struct skynet_context ** new_slot = skynet_malloc(s->slot_size * 2 * sizeof(struct skynet_context *));
//老套路，数据清零
memset(new_slot, 0, s->slot_size * 2 * sizeof(struct skynet_context *));
//把老数据拷过来，要重新hash，但是handle_index没增加
for (i=0;i<s->slot_size;i++) {
int hash = skynet_context_handle(s->slot[i]) & (s->slot_size * 2 - 1);
assert(new_slot[hash] == NULL);
new_slot[hash] = s->slot[i];
}
skynet_free(s->slot);
s->slot = new_slot; //直接替换指针
s->slot_size *= 2; //容量扩大一倍
}
//容量都扩一倍了，再试试会不会hash碰撞吧，再碰我们就再扩大一倍
}
//退休某个服务，反注册
int
skynet_handle_retire(uint32_t handle) {
int ret = 0;
struct handle_storage *s = H;
rwlock_wlock(&s->lock);
//取有效位hash
uint32_t hash = handle & (s->slot_size-1);
//取服务上下文，一会儿要释放的
struct skynet_context * ctx = s->slot[hash];
//较验这个服务是存在的，而且确实对应的就是这个handle
if (ctx != NULL && skynet_context_handle(ctx) == handle) {
s->slot[hash] = NULL; //把空位让出来
ret = 1;
int i;
int j=0, n=s->name_count;
for (i=0; i<n; ++i) {
if (s->name[i].handle == handle) {
skynet_free(s->name[i].name); //释放内存
continue;
} else if (i!=j) { //这里在做数组元素删除操作，把后面的都往前移一下
s->name[j] = s->name[i];
}
++j;//元素删除辅助
}
s->name_count = j;
} else {
ctx = NULL;
}
rwlock_wunlock(&s->lock);
//这里就释放服务了
if (ctx) {
// release ctx may call skynet_handle_* , so wunlock first.
skynet_context_release(ctx);
}
return ret;
}
//全部退休了
void
skynet_handle_retireall() {
struct handle_storage *s = H;
for (;;) {
int n=0;
int i;
for (i=0;i<s->slot_size;i++) {
rwlock_rlock(&s->lock);
struct skynet_context * ctx = s->slot[i]; //取服务上下文
uint32_t handle = 0;
if (ctx)
handle = skynet_context_handle(ctx);
rwlock_runlock(&s->lock);
if (handle != 0) { //对服务上下文的handle进行“退休”
if (skynet_handle_retire(handle)) {
++n;
}
}
}
if (n==0)
return;
}
}
struct skynet_context *
skynet_handle_grab(uint32_t handle) {
struct handle_storage *s = H;
struct skynet_context * result = NULL;
rwlock_rlock(&s->lock);
uint32_t hash = handle & (s->slot_size-1);
struct skynet_context * ctx = s->slot[hash];
if (ctx && skynet_context_handle(ctx) == handle) {
result = ctx;
skynet_context_grab(result); //引用计数+1
}
rwlock_runlock(&s->lock);
return result;
}
//通过name找handle
//算法是二分查找法
//二分查找法请自行googe/bing/baidu
uint32_t
skynet_handle_findname(const char * name) {
struct handle_storage *s = H;
rwlock_rlock(&s->lock);
uint32_t handle = 0;
int begin = 0;
int end = s->name_count - 1;
while (begin<=end) {
int mid = (begin+end)/2;
struct handle_name *n = &s->name[mid];
int c = strcmp(n->name, name); //strcmp是个c系统函数
if (c==0) { //找到匹配的名字
handle = n->handle;
break;
}
if (c<0) { //当前位置的名字 < 要查找的名字，到后半部分去找
begin = mid + 1;
} else { //当前位置的名字 > 要查找的名字，到前半部分去找
end = mid - 1;
}
}
rwlock_runlock(&s->lock);
return handle;
}
//把name插入到name数组中，再关联handle
static void
_insert_name_before(struct handle_storage *s, char *name, uint32_t handle, int before) {
//扩容
if (s->name_count >= s->name_cap) {
s->name_cap *= 2; //扩容
assert(s->name_cap <= MAX_SLOT_SIZE);
struct handle_name * n = skynet_malloc(s->name_cap * sizeof(struct handle_name)); //开一个新数组，容量是老数据的2倍
int i;
for (i=0;i<before;i++) { //复制before位置前的数据
n[i] = s->name[i];
}
for (i=before;i<s->name_count;i++) { //复制before及后面的数据
n[i+1] = s->name[i];
}
skynet_free(s->name); //把老数据内存回收了
s->name = n; //把新数组设进来
} else { //空间够用
int i;
for (i=s->name_count;i>before;i--) { //从后往前，一次一个移动数组元素，把before位置空出来
s->name[i] = s->name[i-1];
}
}
//赋值了
s->name[before].name = name; //名字
s->name[before].handle = handle; //handle
s->name_count ++; //数量+1
}
//给handle绑定一个name
//name是由小到大顺序排列的
//二分查找法，数据结构很重要啊，少年
static const char *
_insert_name(struct handle_storage *s, const char * name, uint32_t handle) {
int begin = 0;
int end = s->name_count - 1;
while (begin<=end) {
int mid = (begin+end)/2;
struct handle_name *n = &s->name[mid];
int c = strcmp(n->name, name);
if (c==0) { //名字已经存在了，不能再绑或者重复绑
return NULL;
}
if (c<0) { //当前字符串 < 要插入的字符串
begin = mid + 1; //二分查找后半部分
} else { //当前字符串 > 要插入的字符串
end = mid - 1; //二分查找前半部分
}
}
char * result = skynet_strdup(name); //字符串复制
//把name插入到数组中
_insert_name_before(s, result, handle, begin);
return result;
}
//给handle绑定一个name
const char *
skynet_handle_namehandle(uint32_t handle, const char *name) {
rwlock_wlock(&H->lock);
const char * ret = _insert_name(H, name, handle);
rwlock_wunlock(&H->lock);
return ret;
}
//初始化handle_storage
void
skynet_handle_init(int harbor) {
assert(H==NULL);
struct handle_storage * s = skynet_malloc(sizeof(*H));
s->slot_size = DEFAULT_SLOT_SIZE; //初始数组大小，slot_size是会变大的
s->slot = skynet_malloc(s->slot_size * sizeof(struct skynet_context *)); //分配内存
memset(s->slot, 0, s->slot_size * sizeof(struct skynet_context *)); //数据清零
rwlock_init(&s->lock);
// reserve 0 for system
s->harbor = (uint32_t) (harbor & 0xff) << HANDLE_REMOTE_SHIFT; //这就是wiki中提到的，前8位是harbor id，HANDLE_REMOTE_SHIFT为24
s->handle_index = 1; //
s->name_cap = 2; //暂时只让放2个
s->name_count = 0; //空的
s->name = skynet_malloc(s->name_cap * sizeof(struct handle_name));
H = s;
// Don't need to free H
}
代码中有个很巧妙的设计，就是s->slot_size-1，它的低位二进制永远都是1。不信你看，刚开始slot_size是4，4-1就是3，扩了以后是8，8-1就是7，然后16,32....。这样的话，和任何数字与操作，都不会丢失“有效的”低位。
好了，到此为止，代码也看完了。总结一下，skynet_handle.c实际上就做了两个核心的事情，一是给服务分配一个handle，二是把handle和name关联起来。
把handle和name关联起来比较容易懂，实际上使用一个数组，关联的时候使用二分查找到数组里查名字，如果名字不存在，就插入一个元素，然后把名字和handle关联起来。插入元素的时候，如果数组空间不足了，就扩容为原来的2倍。
而给服务分配handle稍复杂一些，实际上也是使用一个slot数组，数组下标使用的是一个hash，数组元素指向服务的上下文。这个hash的算法是比较简单粗暴的，就是看从handle_indx开始累计到slot_size，看中间有没有空闲的下标（也就是下标指向为null的），如果遍历完了还是没有，就把slot扩大一倍，还是没有就再扩大一倍，直到找到空位为止，或者是slot长度超出限制为止。
取到了handle以后呢，还要将harbor id附到handle的高8位。
