Skynet源码之：服务管理(6)

// 用于记录 服务名字 和 服务handle 的映射关系
struct handle_name {
	char * name; // 服务名字
	uint32_t handle; // 服务handle
};

// H的定义：H是一个指针，指向的地址存储着handle_storage这个结构体
static struct handle_storage *H = NULL; // 全局单例

void 
skynet_handle_init(int harbor) {
	assert(H==NULL);
	struct handle_storage * s = skynet_malloc(sizeof(*H));
	s->slot_size = DEFAULT_SLOT_SIZE; // DEFAULT_SLOT_SIZE == 4
	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; // HANDLE_REMOTE_SHIFT == 24
    // harbor == 16777216，二进制为 1000000000000000000000000
	s->handle_index = 1;
	s->name_cap = 2;
	s->name_count = 0;
	s->name = skynet_malloc(s->name_cap * sizeof(struct handle_name));
	
	H = s; // 赋值给全局单例H
	
	// Don't need to free H
}

struct handle_storage {
	struct rwlock lock; // 读写锁

	uint32_t harbor; // skynet的节点id
	uint32_t handle_index; // 记录下一个能用的handle
	int slot_size; // 记录最大能容纳多少数量的服务
	struct skynet_context ** slot;
    // slot 属于数组指针，代表这个指针指向的是一个数组
    // 这个数组里每个元素存储的数据，也是一个指针
    // 数组元素里存放的指针，最后指向 真正存放skynet_context结构体的地方
    // 属于二级指针数组
	
	int name_cap; // 名字最大容量
	int name_count; // 名字现有数量
	struct handle_name *name;
};

void 
skynet_handle_init(int harbor) {
	assert(H==NULL);
	struct handle_storage * s = skynet_malloc(sizeof(*H)); // 分配handle_storage内存
    
    // ----- 开始初始化 -----
    
	s->slot_size = DEFAULT_SLOT_SIZE; // DEFAULT_SLOT_SIZE的值为4，即管理的服务数量
    // 重点：分配一个容量为4的指针数组，s->slot是指向这个数组的指针
    // 数组元素存放的是一个指针，该指针 真正指向struct skynet_context的地址
    // 即为二级指针数组
	s->slot = skynet_malloc(s->slot_size * sizeof(struct skynet_context *));
	memset(s->slot, 0, s->slot_size * sizeof(struct skynet_context *)); // 全部置为0
	
	rwlock_init(&s->lock);
	// reserve 0 for system 0这个index是为系统保留的，不使用
    // harbor == 16777216，二进制为 1000000000000000000000000
	s->harbor = (uint32_t) (harbor & 0xff) << HANDLE_REMOTE_SHIFT; // 设置harbor_id
	s->handle_index = 1; // 0是Skynet保留的，跳过0，直接使用1作为第一个能用的handle
	s->name_cap = 2; // 设置名字容量为2
	s->name_count = 0; // 已设置的名字数量为0
    // 分配 名字容量 * sizeof(struct handle_name) 大小的内存
    // 后面用于存放 服务名字 和 服务handle 的映射关系
	s->name = skynet_malloc(s->name_cap * sizeof(struct handle_name)); 
	
    // ----- 初始化结束 -----
    
	H = s; // 赋值给全局单例H
	
	// Don't need to free H
}

// 将需要绑定的服务ctx，记录在 handle_storage.slot 中，并返回一个handle
uint32_t
skynet_handle_register(struct skynet_context *ctx) {
	struct handle_storage *s = H; // 拿到handle_storage结构体

	rwlock_wlock(&s->lock); // 先上锁，免得被其他人干扰
	
	for (;;) { // 如果服务数量非常庞大，这个循环执行的时间将会非常久
		int i;
		uint32_t handle = s->handle_index; //找到能用的handle，看来handle_index就是记录下一个可以使用的handle而已
		for (i=0;i<s->slot_size;i++,handle++) {
			if (handle > HANDLE_MASK) { // HANDLE_MASK == 0xffffff == 16777215
				// 0 is reserved，跳过0，直接使用1作为第一个能用的handle
				handle = 1;
                // 这里handle发生了回溯, 0xffffff的值为16777215，一般一个进程不可能会有这么多服务，内存不够用
                // 但是服务不断退出，又开启新服务，可能会导致handle用完，产生回溯的问题
                // 产生回溯就直接使用 handle == 1
			}
            // 注意这部分代码：
            // 并没有使用handle作为index来访问slot，而是进行了一次hash
            // 而实际却是：即使hash了，得到的hash值也是和handle一模一样的
            // 那么我们为什么还要再执行一次 int hash = handle & (s->slot_size-1)呢？
            // 云风的实现：
            // 		云风的blog：https://blog.codingnow.com/2015/04/handlemap.html
            // 		github仓库：https://github.com/cloudwu/handlemap
            // 网友的分析：
            // 		https://wmf.im/p/skynet%E6%BA%90%E7%A0%81%E5%88%86%E6%9E%90%E5%85%ADhandle%E7%9A%84%E7%AE%A1%E7%90%86/
            // 		https://www.jianshu.com/p/ed04d9c700ed
            
			int hash = handle & (s->slot_size-1); // hash一次就是确保取到一个小于slot_size-1的值
			if (s->slot[hash] == NULL) { // 找到空位，可以存放ctx
				s->slot[hash] = ctx; // 把 ctx 放进数组里面
				s->handle_index = handle + 1; // 把handle+1，留给下一个服务使用
	
				rwlock_wunlock(&s->lock);
				
                // harbor == 1000000000000000000000000
				handle |= s->harbor; // 或操作，就是给每个handle加上harbor_id, 即1000000000000000000000001
				return handle; // 返回handle值
			}
		}
		assert((s->slot_size*2 - 1) <= HANDLE_MASK); // 判断一下slot_size的容量是否小于HANDLE_MASK == 16777215
		struct skynet_context ** new_slot = skynet_malloc(s->slot_size * 2 * sizeof(struct skynet_context *));// 发生扩容，扩容2倍
		memset(new_slot, 0, s->slot_size * 2 * sizeof(struct skynet_context *)); // 把新申请的内存都置为0
        
		for (i=0;i<s->slot_size;i++) { // 把旧的handle移动到新的内存
            // 注意：在这里只有旧的handle存在，才会移动
            // 如果旧的handle已经被置为NULL，那么新的内存就会放着不动
			if (s->slot[i]) {
				int hash = skynet_context_handle(s->slot[i]) & (s->slot_size * 2 - 1); // 重新计算hash值
                // 这里有个特别注意的地方：
                // 旧的hash值是怎么样，新的hash值就是怎么样
                // handle 为 20，hash = handle & (32-1)，值为20&31 = 20 （旧容量）
                // handle 为 20，hash = handle & (64-1)，值为20&63 = 20 （新容量）
                // 因此旧hash和新hash是一样的
                // 
                // 除此之外，假设 20 21 22这3个位置被使用了，后来某个21的服务被退出
                // 那么在新容量中 21 的位置也是一直存在的，会一直空着
                // 即 新的内存放着不会动
				assert(new_slot[hash] == NULL);
				new_slot[hash] = s->slot[i]; // 把ctx从 旧的位置 挪到 新的位置
			}
		}
		skynet_free(s->slot); // 要释放掉旧的内存
		s->slot = new_slot; // 赋值新的内存
		s->slot_size *= 2; // 设置容量为旧的2倍
	}
}

// 将需要绑定的handle传入，删除handle在 handle_storage.slot 中的记录
int
skynet_handle_retire(uint32_t handle) {
	int ret = 0;
	struct handle_storage *s = H; // 拿到handle_storage结构体
	
	rwlock_wlock(&s->lock); // 先上锁，免得被其他人干扰
	
	uint32_t hash = handle & (s->slot_size-1); // 计算出此handle的hash值
	struct skynet_context * ctx = s->slot[hash]; // 获得数组的值，即读取出 指向服务ctx的指针
	
	if (ctx != NULL && skynet_context_handle(ctx) == handle) { // 最后要再确认一次
		s->slot[hash] = NULL; // 直接清空数组的值，即清空指向 ctx地址 的指针
		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); // 调用服务ctx的释放接口
	}
	
	return ret;
}

void 
skynet_handle_retireall() {
	struct handle_storage *s = H; // 拿到handle_storage结构体
    // 为什么这里不需要上锁呢？而是在for循环里上锁？
	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); // 根据服务ctx，拿到handle
				++n;
			}
			rwlock_runlock(&s->lock);
			if (handle != 0) {
				skynet_handle_retire(handle); // 调用函数，一个一个轮流解绑 服务ctx和handle 的关系
			}
		}
		if (n==0)
			return;
	}
}

// 在 skynet_handle_retireall() 函数中，做了遍历 handle_storage 结构体中的 slot 数组的操作
// 需要在遍历的过程中获取每个槽位对应的 skynet_context 上下文，并进行相应的处理
// 为了保证数据的一致性和线程安全，需要使用读写锁 rwlock 对 handle_storage 结构体进行保护

// 在函数的开始处，获取了 handle_storage 结构体的指针 s ，此时还没有开始遍历 slot 数组
// 接下来进入一个无限循环，每次循环中都会遍历 slot 数组
// 在每次遍历之前，通过调用 rwlock_rlock(&s->lock) 上读锁，对 handle_storage 结构体进行保护
// 这是因为在遍历 slot 数组的过程中，需要访问其中的元素，而读锁可以允许多个线程同时读取数据，不会互斥
// 在遍历 slot 数组的过程中，先获取当前槽位的 skynet_context 上下文，并根据上下文获取对应的 handle
// 这里的目的是判断该槽位是否存在有效的上下文，如果有，则需要进行相应的处理，否则继续遍历下一个槽位
// 
// 在获取完 handle 后，通过调用 rwlock_runlock(&s->lock) 解读锁，释放对 handle_storage 结构体的保护
// 这样其他线程就可以在获取读锁后继续访问 handle_storage
// 
// 判断获取的 handle 是否为 0
// 如果不为 0，则调用 skynet_handle_retire(handle) 函数进行处理，解绑服务上下文和 handle 的关系
// 但 skynet_handle_retire(handle) 函数处理的时间可能非常长
// 所以前一步，一旦获取到 handle 后，就先立马解开读锁
// 这就解释了为什么不能在一开始就加锁
// 
// 循环结束后，如果遍历过程中没有发现有效的上下文，即 `n==0`，则函数返回
//
// 总结起来，就是为了减少锁的粒度，避免过多的锁竞争，这样可以在保证数据一致性的同时，提高并发性能

struct skynet_context * 
skynet_handle_grab(uint32_t handle) {
	struct handle_storage *s = H; // 拿到handle_storage结构体
	struct skynet_context * result = NULL;
	
	rwlock_rlock(&s->lock); // 先上锁，免得被其他人干扰
	
	uint32_t hash = handle & (s->slot_size-1); // 计算出hash值
	struct skynet_context * ctx = s->slot[hash]; // 获得指向 服务ctx地址 的指针
	if (ctx && skynet_context_handle(ctx) == handle) { // 再确认一下服务ctx内的handle与查找的handle对不对得上
		result = ctx; // 赋值给result
		skynet_context_grab(result); // 既然你来寻找服务ctx的地址，你肯定是要引用它，那我需要把此服务ctx的引用+1
	}
	
	rwlock_runlock(&s->lock);
	
	return result; // 返回结果
}

const char * 
skynet_handle_namehandle(uint32_t handle, const char *name) {
	rwlock_wlock(&H->lock);

	const char * ret = _insert_name(H, name, handle); // 把 服务名字 和 服务handle 插入记录中的具体实现
	
	rwlock_wunlock(&H->lock);
	
	return ret;
}

// 实际就是采用二分法，把名字插入进去
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;
}

// 把 字符串 复制到 一块新的内存
char *
skynet_strdup(const char *str) {
	size_t sz = strlen(str);
	char * ret = skynet_malloc(sz+1);
	memcpy(ret, str, sz+1);
	return ret;
}

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;
        
        // HANDLE_MASK   ==   16777216
        // MAX_SLOT_SIZE == 1073741824
        // 为什么handle的数量限制是 16777216，名字容量限制却是 1073741824
		assert(s->name_cap <= MAX_SLOT_SIZE); 
		struct handle_name * n = skynet_malloc(s->name_cap * sizeof(struct handle_name));
		int i;
		for (i=0;i<before;i++) {
			n[i] = s->name[i];
		}
		for (i=before;i<s->name_count;i++) {
			n[i+1] = s->name[i];
		}
		skynet_free(s->name);
		s->name = n;
	} else {
		int i;
		for (i=s->name_count;i>before;i--) {
			s->name[i] = s->name[i-1];
		}
	}
	s->name[before].name = name;
	s->name[before].handle = handle;
	s->name_count ++;
}

struct handle_name {
	char * name; // 命名的名字
	uint32_t handle; // 服务的handle
};

struct handle_storage {
	struct rwlock lock;
	
	uint32_t harbor; // 本进程所属的节点
	uint32_t handle_index; // 用于记录下一个能用的handle
	int slot_size; // 用于记录服务的最大容量
	struct skynet_context ** slot; // 属于指针数组，该值指向一个数组，数组元素存放的数据是真正指向skynet_context的指针
	
	int name_cap; // 设置服务命名的容量
	int name_count; // 设置已经命名的数量
	struct handle_name *name; // 名称和handle的结构体
};

uint32_t 
skynet_handle_findname(const char * name) {
	struct handle_storage *s = H; // 拿到handle_storage结构体
	
	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);
		if (c==0) {
			handle = n->handle;
			break;
		}
		if (c<0) {
			begin = mid + 1;
		} else {
			end = mid - 1;
		}
	}
	
	rwlock_runlock(&s->lock);
	
	return handle;
}