From 6a6d9db54daa51eb118ccdfdf65259500f629c60 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Tue, 6 May 2014 13:35:36 +0800 Subject: [PATCH] remove blockcall --- examples/main.lua | 2 - lualib/skynet.lua | 21 ++------ skynet-src/skynet_mq.c | 102 +++++++------------------------------ skynet-src/skynet_mq.h | 7 +-- skynet-src/skynet_server.c | 25 +++------ 5 files changed, 35 insertions(+), 122 deletions(-) diff --git a/examples/main.lua b/examples/main.lua index a9a2143b..7731bdab 100644 --- a/examples/main.lua +++ b/examples/main.lua @@ -14,7 +14,5 @@ skynet.start(function() maxclient = max_client, }) - skynet.newservice("testmulticast") - skynet.exit() end) diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 6fcbb66d..c3b38df9 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -305,17 +305,6 @@ function skynet.call(addr, typename, ...) return p.unpack(yield_call(addr, session)) 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 " .. skynet.address(addr)) - end - return p.unpack(yield_call(addr, session)) -end - function skynet.rawcall(addr, typename, msg, sz) local p = proto[typename] local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address") @@ -328,7 +317,7 @@ function skynet.ret(msg, sz) end function skynet.retpack(...) - return skynet.ret(skynet.pack(...)) + return skynet.ret(skynet.pack(...)) end function skynet.wakeup(co) @@ -345,14 +334,14 @@ function skynet.dispatch(typename, func) end local function unknown_request(session, address, msg, sz) - print("Unknown request :" , c.tostring(msg,sz)) + print("Unknown request :" , c.tostring(msg,sz)) error(string.format("Unknown session : %d from %x", session, address)) end function skynet.dispatch_unknown_request(unknown) - local prev = unknown_request - unknown_request = unknown - return prev + local prev = unknown_request + unknown_request = unknown + return prev end local function unknown_response(session, address, msg, sz) diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index f0e48440..46503b6b 100644 --- a/skynet-src/skynet_mq.c +++ b/skynet-src/skynet_mq.c @@ -13,12 +13,8 @@ // 0 means mq is not in global mq. // 1 means mq is in global mq , or the message is dispatching. -// 2 means message is dispatching with locked session set. -// 3 means mq is not in global mq, and locked session has been set. #define MQ_IN_GLOBAL 1 -#define MQ_DISPATCHING 2 -#define MQ_LOCKED 3 struct message_queue { uint32_t handle; @@ -27,7 +23,6 @@ struct message_queue { int tail; int lock; int release; - int lock_session; int in_global; struct skynet_message *queue; }; @@ -89,9 +84,11 @@ skynet_mq_create(uint32_t handle) { q->head = 0; q->tail = 0; q->lock = 0; + // When the queue is create (always between service create and service init) , + // set in_global flag to avoid push it to global queue . + // If the service init success, skynet_context_new will call skynet_mq_force_push to push it to global queue. q->in_global = MQ_IN_GLOBAL; q->release = 0; - q->lock_session = 0; q->queue = skynet_malloc(sizeof(struct skynet_message) * q->cap); return q; @@ -161,82 +158,28 @@ 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; - if (head < 0) { - head = q->cap - 1; - } - if (head == q->tail) { - expand_queue(q); - --q->tail; - head = q->cap - 1; - } - - q->queue[head] = *message; - q->head = head; - - _unlock(q); -} - void skynet_mq_push(struct message_queue *q, struct skynet_message *message) { assert(message); LOCK(q) - if (q->lock_session !=0 && message->session == q->lock_session) { - _pushhead(q,message); - } else { - q->queue[q->tail] = *message; - if (++ q->tail >= q->cap) { - q->tail = 0; - } + q->queue[q->tail] = *message; + if (++ q->tail >= q->cap) { + q->tail = 0; + } - if (q->head == q->tail) { - expand_queue(q); - } + if (q->head == q->tail) { + expand_queue(q); + } - if (q->lock_session == 0) { - if (q->in_global == 0) { - q->in_global = MQ_IN_GLOBAL; - skynet_globalmq_push(q); - } - } + if (q->in_global == 0) { + q->in_global = MQ_IN_GLOBAL; + skynet_globalmq_push(q); } UNLOCK(q) } -void -skynet_mq_lock(struct message_queue *q, int session) { - LOCK(q) - assert(q->lock_session == 0); - assert(q->in_global == MQ_IN_GLOBAL); - q->in_global = MQ_DISPATCHING; - q->lock_session = 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 = skynet_malloc(sizeof(*q)); @@ -257,14 +200,8 @@ void skynet_mq_pushglobal(struct message_queue *queue) { LOCK(queue) assert(queue->in_global); - if (queue->in_global == MQ_DISPATCHING) { - // lock message queue just now. - queue->in_global = MQ_LOCKED; - } - if (queue->lock_session == 0) { - skynet_globalmq_push(queue); - queue->in_global = MQ_IN_GLOBAL; - } + skynet_globalmq_push(queue); + queue->in_global = MQ_IN_GLOBAL; UNLOCK(queue) } @@ -280,26 +217,25 @@ skynet_mq_mark_release(struct message_queue *q) { } static int -_drop_queue(struct message_queue *q) { - // todo: send message back to message source +_drop_queue(struct message_queue *q, message_drop drop_func, void *ud) { struct skynet_message msg; int s = 0; while(!skynet_mq_pop(q, &msg)) { ++s; - skynet_free(msg.data); + drop_func(ud, &msg); } _release(q); return s; } int -skynet_mq_release(struct message_queue *q) { +skynet_mq_release(struct message_queue *q, message_drop drop_func, void *ud) { int ret = 0; LOCK(q) if (q->release) { UNLOCK(q) - ret = _drop_queue(q); + ret = _drop_queue(q, drop_func, ud); } else { skynet_mq_force_push(q); UNLOCK(q) diff --git a/skynet-src/skynet_mq.h b/skynet-src/skynet_mq.h index 11945ce9..be30018c 100644 --- a/skynet-src/skynet_mq.h +++ b/skynet-src/skynet_mq.h @@ -17,14 +17,15 @@ struct message_queue * skynet_globalmq_pop(void); struct message_queue * skynet_mq_create(uint32_t handle); void skynet_mq_mark_release(struct message_queue *q); -int skynet_mq_release(struct message_queue *q); + +typedef void (*message_drop)(struct skynet_message *, void *); + +int skynet_mq_release(struct message_queue *q, message_drop drop_func, void *ud); uint32_t skynet_mq_handle(struct message_queue *); // 0 for success 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); // return the length of message queue, for debug int skynet_mq_length(struct message_queue *q); diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index f8f12f4c..08fdc016 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -90,6 +90,11 @@ _id_to_hex(char * str, uint32_t id) { str[9] = '\0'; } +static void +drop_message(struct skynet_message *msg, void *ud) { + skynet_free(msg->data); +} + struct skynet_context * skynet_context_new(const char * name, const char *param) { struct skynet_module * mod = skynet_module_query(name); @@ -134,7 +139,7 @@ skynet_context_new(const char * name, const char *param) { skynet_error(ctx, "FAILED launch %s", name); skynet_context_release(ctx); skynet_handle_retire(ctx->handle); - skynet_mq_release(queue); + skynet_mq_release(queue, drop_message, NULL); return NULL; } } @@ -222,7 +227,7 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) { struct skynet_context * ctx = skynet_handle_grab(handle); if (ctx == NULL) { - int s = skynet_mq_release(q); + int s = skynet_mq_release(q, drop_message, NULL); if (s>0) { skynet_error(NULL, "Drop message queue %x (%d messages)", handle,s); } @@ -301,22 +306,6 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * return context->result; } - if (strcmp(cmd,"LOCK") == 0) { - if (context->init == false) { - return NULL; - } - skynet_mq_lock(context->queue, context->session_id+1); - 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);