diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c index 1e76ab62..4e23233d 100644 --- a/lualib-src/lua-skynet.c +++ b/lualib-src/lua-skynet.c @@ -309,7 +309,7 @@ _send(lua_State *L) { if (session < 0) { // send to invalid address // todo: maybe throw error is better - session = 0; + return 0; } lua_pushinteger(L,session); return 1; diff --git a/lualib/skynet.lua b/lualib/skynet.lua index e1b83650..8a26ac1a 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -233,20 +233,24 @@ skynet.tostring = assert(c.tostring) function skynet.call(addr, typename, ...) local p = proto[typename] local session = c.send(addr, p.id , nil , p.pack(...)) - return p.unpack(coroutine_yield("CALL", session)) + return p.unpack(coroutine_yield("CALL", assert(session, "call to invalid address"))) end function skynet.blockcall(addr, typename , ...) local p = proto[typename] c.command("LOCK") local session = c.send(addr, p.id , nil , p.pack(...)) + if session == nil then + c.command("UNLOCK") + error("call to invalid address") + end return p.unpack(coroutine_yield("CALL", session)) end function skynet.rawcall(addr, typename, msg, sz) local p = proto[typename] local session = c.send(addr, p.id , nil , msg, sz) - return coroutine_yield("CALL", session) + return coroutine_yield("CALL", assert(session, "call to invalid address")) end function skynet.ret(msg, sz) @@ -405,7 +409,7 @@ do f = function(...) local addr = remote_query(t.__remote) -- the proto is 11 (lua is 10) - local session = _send(addr, 11 , nil, _pack(t,method,...)) + local session = assert(_send(addr, 11 , nil, _pack(t,method,...)), "call to invalid address") local msg, sz = _yield("CALL", session) return select(2,assert(_unpack(msg,sz))) end diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index ffd6c45c..abdeb9b0 100644 --- a/skynet-src/skynet_mq.c +++ b/skynet-src/skynet_mq.c @@ -147,6 +147,19 @@ expand_queue(struct message_queue *q) { q->queue = new_queue; } +static void +_unlock(struct message_queue *q) { + // this api use in push a unlock message, so the in_global flags must not be 0 , + // but the q is not exist in global queue. + if (q->in_global == MQ_LOCKED) { + skynet_globalmq_push(q); + q->in_global = MQ_IN_GLOBAL; + } else { + assert(q->in_global == MQ_DISPATCHING); + } + q->lock_session = 0; +} + static void _pushhead(struct message_queue *q, struct skynet_message *message) { int head = q->head - 1; @@ -162,15 +175,7 @@ _pushhead(struct message_queue *q, struct skynet_message *message) { q->queue[head] = *message; q->head = head; - // this api use in push a unlock message, so the in_global flags must not be 0 , - // but the q is not exist in global queue. - if (q->in_global == MQ_LOCKED) { - skynet_globalmq_push(q); - q->in_global = MQ_IN_GLOBAL; - } else { - assert(q->in_global == MQ_DISPATCHING); - } - q->lock_session = 0; + _unlock(q); } void @@ -211,6 +216,13 @@ skynet_mq_lock(struct message_queue *q, int session) { UNLOCK(q) } +void +skynet_mq_unlock(struct message_queue *q) { + LOCK(q) + _unlock(q); + UNLOCK(q) +} + void skynet_mq_init() { struct global_queue *q = malloc(sizeof(*q)); diff --git a/skynet-src/skynet_mq.h b/skynet-src/skynet_mq.h index bd71e91f..ced1beff 100644 --- a/skynet-src/skynet_mq.h +++ b/skynet-src/skynet_mq.h @@ -24,6 +24,7 @@ uint32_t skynet_mq_handle(struct message_queue *); int skynet_mq_pop(struct message_queue *q, struct skynet_message *message); void skynet_mq_push(struct message_queue *q, struct skynet_message *message); void skynet_mq_lock(struct message_queue *q, int session); +void skynet_mq_unlock(struct message_queue *q); void skynet_mq_force_push(struct message_queue *q); void skynet_mq_pushglobal(struct message_queue *q); diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index dda09622..05c83618 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -384,6 +384,14 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * return NULL; } + if (strcmp(cmd,"UNLOCK") == 0) { + if (context->init == false) { + return NULL; + } + skynet_mq_unlock(context->queue); + return NULL; + } + if (strcmp(cmd,"REG") == 0) { if (param == NULL || param[0] == '\0') { sprintf(context->result, ":%x", context->handle);