void 
skynet_mq_init() {
	struct global_queue *q = (struct global_queue*)skynet_malloc(sizeof(*q));
	memset(q,0,sizeof(*q));
	q->queue = (message_queue**)skynet_malloc(MAX_GLOBAL_MQ * sizeof(struct message_queue *));
	q->flag = (bool*)skynet_malloc(MAX_GLOBAL_MQ * sizeof(bool));
	memset(q->flag, 0, sizeof(bool) * MAX_GLOBAL_MQ);
	Q=q;
}

static struct msg_queue *
new_queue() {
	struct msg_queue * queue = skynet_malloc(sizeof(*queue));
	queue->size = DEFAULT_QUEUE_SIZE;
	queue->head = 0;
	queue->tail = 0;
	queue->data = skynet_malloc(DEFAULT_QUEUE_SIZE * sizeof(struct msg));

	return queue;
}

struct skynet_mq *
skynet_mq_create() {
	struct skynet_mq *mq = skynet_malloc(sizeof(*mq));
	mq->head = 0;
	mq->tail = 0;
	mq->cap = MQ_LENGTH;
	mq->q = skynet_malloc(mq->cap * sizeof(struct skynet_message_package));
	rwlock_init(&mq->lock);
	return mq;
}

static int
pack(lua_State *L, void *data, size_t size) {
	struct mc_package * pack = skynet_malloc(sizeof(struct mc_package));
	pack->reference = 0;
	pack->size = (uint32_t)size;
	pack->data = data;
	struct mc_package ** ret = skynet_malloc(sizeof(*ret));
	*ret = pack;
	// 之所以不是直接push mc_package * pack是因为它们的职责不同
	// mc_package是用于发布到各个监听者的，具有引用计数，计数到0的时候才会释放
	// 而对于mc_package**来说，只是用于普通服务间传递，目前是multicastd.lua会负责自动释放
	lua_pushlightuserdata(L, ret);
	lua_pushinteger(L, sizeof(ret));
	return 2;
}

struct message_queue * 
skynet_mq_create(uint32_t handle) {
	struct message_queue *q = (struct message_queue*)skynet_malloc(sizeof(*q));
	q->handle = handle;
	q->cap = DEFAULT_QUEUE_SIZE;
	q->head = 0;
	q->tail = 0;
	q->lock = 0;
	q->in_global = MQ_IN_GLOBAL;
	q->release = 0;
	q->lock_session = 0;
	q->queue = (struct skynet_message*)skynet_malloc(sizeof(struct skynet_message) * q->cap);

	return q;
}

struct otu *
otu_create(void) {
	struct otu * u = skynet_malloc(sizeof(*u));
	memset(u,0,sizeof(*u));
	u->service_id = -1;
	return u;
}

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;
			int hash = handle & (s->slot_size-1);
			if (s->slot[hash] == NULL) {
				s->slot[hash] = ctx;
				s->handle_index = handle + 1;

				rwlock_wunlock(&s->lock);

				handle |= s->harbor;
				return handle;
			}
		}
		assert((s->slot_size*2 - 1) <= HANDLE_MASK);
		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 *));
		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;
	}
}

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));
		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 ++;
}

static struct hashmap *
_hash_new()
{
    struct hashmap * h = skynet_malloc(sizeof(struct hashmap));
    memset(h, 0, sizeof(*h));
    return h;
}

static void
queue_push(struct queue *q, const void *value) {
	void * slot = q->buffer + q->tail * q->sz;
	++q->tail;
	if (q->tail >= q->cap)
		q->tail = 0;
	if (q->head == q->tail) {
		// full
		assert(q->sz > 0);
		int cap = q->cap * 2;
		char * tmp = skynet_malloc(cap * q->sz);
		int i;
		int head = q->head;
		for (i=0;i<q->cap;i++) {
			memcpy(tmp + i * q->sz, q->buffer + head * q->sz, q->sz);
			++head;
			if (head >= q->cap) {
				head = 0;
			}
		}
		skynet_free(q->buffer);
		q->head = 0;
		slot = tmp + (q->cap-1) * q->sz;
		q->tail = q->cap;
		q->cap = cap;
		q->buffer = tmp;
	}
	memcpy(slot, value, q->sz);
}

//初始化1个monitor
struct skynet_monitor *
skynet_monitor_new()
{
	struct skynet_monitor * ret = skynet_malloc(sizeof(*ret));
	memset(ret, 0, sizeof(*ret));
	return ret;
}

struct logger *
logger_create(void) {
	struct logger * inst = skynet_malloc(sizeof(*inst));
	inst->handle = NULL;
	inst->close = 0;
	return inst;
}

// return 1 when mq is full
static int
perpare_space(struct skynet_mq *mq, struct skynet_mq *expand) {
	int tail = mq->tail + 1;
	if (tail >= mq->cap) {
		tail -= mq->cap;
	}
	if (tail != mq->head) {
		return 0;
	}
	expand->cap = mq->cap * 2;
	expand->q = skynet_malloc(expand->cap * sizeof(struct skynet_message_package));
	int head = mq->head;
	tail = mq->tail;
	if (tail < head) {
		tail += mq->cap;
	}
	expand->tail = tail;
	expand->head = head;
	for (;head<tail;head++) {
		int ptr = head;
		if (ptr >= mq->cap) {
			ptr -= mq->cap;
		}
		expand->q[head] = mq->q[ptr];
	}

	return 1;
}

// mainloop thread
static void
forward_message(int type, bool padding, struct socket_message * result) {
	struct skynet_socket_message *sm;
	int sz = sizeof(*sm);
	if (padding) {
		if (result->data) {
			sz += strlen(result->data);
		} else {
			result->data = "";
		}
	}
	sm = (struct skynet_socket_message *)skynet_malloc(sz);
	sm->type = type;
	sm->id = result->id;
	sm->ud = result->ud;
	if (padding) {
		sm->buffer = NULL;
		memcpy(sm+1, result->data, sz - sizeof(*sm));
	} else {
		sm->buffer = result->data;
	}

	struct skynet_message message;
	message.source = 0;
	message.session = 0;
	message.data = sm;
	message.sz = sz | PTYPE_SOCKET << HANDLE_REMOTE_SHIFT;
	
	if (skynet_context_push((uint32_t)result->opaque, &message)) {
		// todo: report somewhere to close socket
		// don't call skynet_socket_close here (It will block mainloop)
		skynet_free(sm->buffer);
		skynet_free(sm);
	}
}

/* 构建定时器对象, 包括分配内存、初始化触发列表集、初始化锁并将当前时间 time 置为 0.
 * 此函数返回初始化好的定时器对象. */
static struct timer *
timer_create_timer() {
	struct timer *r=(struct timer *)skynet_malloc(sizeof(struct timer));
	/* 将 time 初始化为 0 */
	memset(r,0,sizeof(*r));

	int i,j;

	/* 初始化最近的触发列表集 */
	for (i=0;i<TIME_NEAR;i++) {
		link_clear(&r->near[i]);
	}

	/* 初始化 4 级触发时间较远的触发列表集 */
	for (i=0;i<4;i++) {
		for (j=0;j<TIME_LEVEL;j++) {
			link_clear(&r->t[i][j]);
		}
	}

	SPIN_INIT(r)

	/* current 以及除了 time 以外的其它时间字段还会进一步初始化 */
	r->current = 0;

	return r;
}

struct gate *
gate_create(void) {
	struct gate * g = skynet_malloc(sizeof(*g));
	memset(g,0,sizeof(*g));
	g->listen_id = -1;
	return g;
}

/* 向数据包队列中插入数据, 这些数据会按照数据包的结构, 即两个字节的长度紧跟着内容, 分析这些数据有多少个数据包.
 * 对于其中的完整数据包将插入到完整包队列中去, 对于其中不完整的包将插入到不完整数据包的哈希表中去. 特别是对于
 * 长度只有 1 个字节的不完整包, 此时已经无法计算出数据包的长度, 因而不完整数据包的 read 为 -1 , 而 header 为此一个字节值.
 * 当下一次套接字中接收到数据时会判断 read 的长度, 从而正确拼接数据包.
 *
 * 参数: L 是虚拟机栈; fd 是数据所属的套接字 id; buffer 是数据内容; size 是数据大小;
 * 函数无返回值 */
static void
push_more(lua_State *L, int fd, uint8_t *buffer, int size) {
	if (size == 1) {
		struct uncomplete * uc = save_uncomplete(L, fd);
		uc->read = -1;
		uc->header = *buffer;
		return;
	}
	int pack_size = read_size(buffer);
	buffer += 2;
	size -= 2;

	/* 虽然数据包的内容是不完整的, 但是给 pack.buffer 分配的内存是完整的 */
	if (size < pack_size) {
		struct uncomplete * uc = save_uncomplete(L, fd);
		uc->read = size;
		uc->pack.size = pack_size;
		uc->pack.buffer = skynet_malloc(pack_size);
		memcpy(uc->pack.buffer, buffer, size);
		return;
	}
	push_data(L, fd, buffer, pack_size, 1);

	buffer += pack_size;
	size -= pack_size;
	if (size > 0) {
		push_more(L, fd, buffer, size);
	}
}

struct harbor *
harbor_create(void) {
	struct harbor * h = skynet_malloc(sizeof(*h));
	memset(h,0,sizeof(*h));
	h->map = hash_new();
	return h;
}

struct snlua *
snlua_create(void) {
	struct snlua * l = skynet_malloc(sizeof(*l));
	memset(l,0,sizeof(*l));
	l->L = lua_newstate(skynet_lalloc, NULL);
	return l;
}

static void
queue_init(struct queue *q, int sz) {
	q->head = 0;
	q->tail = 0;
	q->sz = sz;
	q->cap = 4;
	q->buffer = skynet_malloc(q->cap * q->sz);
}