Compare commits

..

9 Commits

Author SHA1 Message Date
Cloud Wu
8b7ba7bdbe add testresponse 2014-09-01 11:17:16 +08:00
Cloud Wu
165bed46a8 only skynet.call response error 2014-08-27 18:54:46 +08:00
云风
9bb8099008 Merge pull request #160 from cloudwu/dev
release v0.6.1
2014-08-25 10:15:34 +08:00
Cloud Wu
98630badd5 update history for release v0.6.1 2014-08-25 10:13:04 +08:00
Cloud Wu
d0c4a4012a clear socket when connect failed 2014-08-22 15:41:05 +08:00
Cloud Wu
04688e98a7 improve seri lib 2014-08-20 23:34:23 +08:00
Cloud Wu
e0feea9995 fix issue #156 2014-08-19 23:44:07 +08:00
Cloud Wu
55ea556060 avoid issue #154 2014-08-19 23:26:42 +08:00
Cloud Wu
4803e50977 fix error log 2014-08-19 20:52:43 +08:00
8 changed files with 158 additions and 195 deletions

View File

@@ -1,3 +1,13 @@
v0.6.2 (2014-9-1)
-----------
* bugfix: only skynet.call response PTYPE_ERROR
v0.6.1 (2014-8-25)
-----------
* bugfix: datacenter.wakeup
* change struct msg name to avoid conflict in mac
* improve seri library
v0.6.0 (2014-8-18)
-----------
* add sharedata

View File

@@ -589,11 +589,11 @@ read64(lua_State *L, uint32_t xx[2], uint32_t yy[2]) {
size_t sz = 0;
const uint8_t *x = (const uint8_t *)luaL_checklstring(L, 1, &sz);
if (sz != 8) {
luaL_error(L, "Invalid hmac x");
luaL_error(L, "Invalid uint64 x");
}
const uint8_t *y = (const uint8_t *)luaL_checklstring(L, 2, &sz);
if (sz != 8) {
luaL_error(L, "Invalid hmac y");
luaL_error(L, "Invalid uint64 y");
}
xx[0] = x[0] | x[1]<<8 | x[2]<<16 | x[3]<<24;
xx[1] = x[4] | x[5]<<8 | x[6]<<16 | x[7]<<24;
@@ -703,7 +703,7 @@ ldhexchange(lua_State *L) {
size_t sz = 0;
const uint8_t *x = (const uint8_t *)luaL_checklstring(L, 1, &sz);
if (sz != 8) {
luaL_error(L, "Invalid hmac x");
luaL_error(L, "Invalid dh uint64 key");
}
uint32_t xx[2];
xx[0] = x[0] | x[1]<<8 | x[2]<<16 | x[3]<<24;

View File

@@ -1,5 +1,5 @@
/*
https://github.com/cloudwu/lua-serialize
modify from https://github.com/cloudwu/lua-serialize
*/
#include "skynet_malloc.h"
@@ -35,14 +35,13 @@ struct block {
struct write_block {
struct block * head;
int len;
struct block * current;
int len;
int ptr;
};
struct read_block {
char * buffer;
struct block * current;
int len;
int ptr;
};
@@ -78,38 +77,17 @@ _again:
static void
wb_init(struct write_block *wb , struct block *b) {
if (b==NULL) {
wb->head = blk_alloc();
wb->len = 0;
wb->current = wb->head;
wb->ptr = 0;
wb_push(wb, &wb->len, sizeof(wb->len));
} else {
wb->head = b;
int * plen = (int *)b->buffer;
int sz = *plen;
wb->len = sz;
while (b->next) {
sz -= BLOCK_SIZE;
b = b->next;
}
wb->current = b;
wb->ptr = sz;
}
}
static struct block *
wb_close(struct write_block *b) {
b->current = b->head;
b->ptr = 0;
wb_push(b, &b->len, sizeof(b->len));
b->current = NULL;
return b->head;
wb->head = b;
assert(b->next == NULL);
wb->len = 0;
wb->current = wb->head;
wb->ptr = 0;
}
static void
wb_free(struct write_block *wb) {
struct block *blk = wb->head;
blk = blk->next; // the first block is on stack
while (blk) {
struct block * next = blk->next;
skynet_free(blk);
@@ -124,7 +102,6 @@ wb_free(struct write_block *wb) {
static void
rball_init(struct read_block * rb, char * buffer, int size) {
rb->buffer = buffer;
rb->current = NULL;
rb->len = size;
rb->ptr = 0;
}
@@ -135,63 +112,10 @@ rb_read(struct read_block *rb, void *buffer, int sz) {
return NULL;
}
if (rb->buffer) {
int ptr = rb->ptr;
rb->ptr += sz;
rb->len -= sz;
return rb->buffer + ptr;
}
if (rb->ptr == BLOCK_SIZE) {
struct block * next = rb->current->next;
skynet_free(rb->current);
rb->current = next;
rb->ptr = 0;
}
int copy = BLOCK_SIZE - rb->ptr;
if (sz <= copy) {
void * ret = rb->current->buffer + rb->ptr;
rb->ptr += sz;
rb->len -= sz;
return ret;
}
char * tmp = buffer;
memcpy(tmp, rb->current->buffer + rb->ptr, copy);
sz -= copy;
tmp += copy;
rb->len -= copy;
for (;;) {
struct block * next = rb->current->next;
skynet_free(rb->current);
rb->current = next;
if (sz < BLOCK_SIZE) {
memcpy(tmp, rb->current->buffer, sz);
rb->ptr = sz;
rb->len -= sz;
return buffer;
}
memcpy(tmp, rb->current->buffer, BLOCK_SIZE);
sz -= BLOCK_SIZE;
tmp += BLOCK_SIZE;
rb->len -= BLOCK_SIZE;
}
}
static void
rb_close(struct read_block *rb) {
while (rb->current) {
struct block * next = rb->current->next;
skynet_free(rb->current);
rb->current = next;
}
rb->len = 0;
rb->ptr = 0;
int ptr = rb->ptr;
rb->ptr += sz;
rb->len -= sz;
return rb->buffer + ptr;
}
static inline void
@@ -271,7 +195,7 @@ wb_string(struct write_block *wb, const char *str, int len) {
}
}
static void _pack_one(lua_State *L, struct write_block *b, int index, int depth);
static void pack_one(lua_State *L, struct write_block *b, int index, int depth);
static int
wb_table_array(lua_State *L, struct write_block * wb, int index, int depth) {
@@ -288,7 +212,7 @@ wb_table_array(lua_State *L, struct write_block * wb, int index, int depth) {
int i;
for (i=1;i<=array_size;i++) {
lua_rawgeti(L,index,i);
_pack_one(L, wb, -1, depth);
pack_one(L, wb, -1, depth);
lua_pop(L,1);
}
@@ -307,8 +231,8 @@ wb_table_hash(lua_State *L, struct write_block * wb, int index, int depth, int a
continue;
}
}
_pack_one(L,wb,-2,depth);
_pack_one(L,wb,-1,depth);
pack_one(L,wb,-2,depth);
pack_one(L,wb,-1,depth);
lua_pop(L, 1);
}
wb_nil(wb);
@@ -324,7 +248,7 @@ wb_table(lua_State *L, struct write_block *wb, int index, int depth) {
}
static void
_pack_one(lua_State *L, struct write_block *b, int index, int depth) {
pack_one(lua_State *L, struct write_block *b, int index, int depth) {
if (depth > MAX_DEPTH) {
wb_free(b);
luaL_error(L, "serialize can't pack too depth table");
@@ -370,27 +294,24 @@ _pack_one(lua_State *L, struct write_block *b, int index, int depth) {
}
static void
_pack_from(lua_State *L, struct write_block *b, int from) {
pack_from(lua_State *L, struct write_block *b, int from) {
int n = lua_gettop(L) - from;
int i;
for (i=1;i<=n;i++) {
_pack_one(L, b , from + i, 0);
pack_one(L, b , from + i, 0);
}
}
static inline void
__invalid_stream(lua_State *L, struct read_block *rb, int line) {
invalid_stream_line(lua_State *L, struct read_block *rb, int line) {
int len = rb->len;
if (rb->buffer == NULL) {
rb_close(rb);
}
luaL_error(L, "Invalid serialize stream %d (line:%d)", len, line);
}
#define _invalid_stream(L,rb) __invalid_stream(L,rb,__LINE__)
#define invalid_stream(L,rb) invalid_stream_line(L,rb,__LINE__)
static int
_get_integer(lua_State *L, struct read_block *rb, int cookie) {
get_integer(lua_State *L, struct read_block *rb, int cookie) {
switch (cookie) {
case 0:
return 0;
@@ -398,93 +319,93 @@ _get_integer(lua_State *L, struct read_block *rb, int cookie) {
uint8_t n = 0;
uint8_t * pn = rb_read(rb,&n,1);
if (pn == NULL)
_invalid_stream(L,rb);
invalid_stream(L,rb);
return *pn;
}
case 2: {
uint16_t n = 0;
uint16_t * pn = rb_read(rb,&n,2);
if (pn == NULL)
_invalid_stream(L,rb);
invalid_stream(L,rb);
return *pn;
}
case 4: {
int n = 0;
int * pn = rb_read(rb,&n,4);
if (pn == NULL)
_invalid_stream(L,rb);
invalid_stream(L,rb);
return *pn;
}
default:
_invalid_stream(L,rb);
invalid_stream(L,rb);
return 0;
}
}
static double
_get_number(lua_State *L, struct read_block *rb, int cookie) {
get_number(lua_State *L, struct read_block *rb, int cookie) {
if (cookie == 8) {
double n = 0;
double * pn = rb_read(rb,&n,8);
if (pn == NULL)
_invalid_stream(L,rb);
invalid_stream(L,rb);
return *pn;
} else {
return (double)_get_integer(L,rb,cookie);
return (double)get_integer(L,rb,cookie);
}
}
static void *
_get_pointer(lua_State *L, struct read_block *rb) {
get_pointer(lua_State *L, struct read_block *rb) {
void * userdata = 0;
void ** v = (void **)rb_read(rb,&userdata,sizeof(userdata));
if (v == NULL) {
_invalid_stream(L,rb);
invalid_stream(L,rb);
}
return *v;
}
static void
_get_buffer(lua_State *L, struct read_block *rb, int len) {
get_buffer(lua_State *L, struct read_block *rb, int len) {
char tmp[len];
char * p = rb_read(rb,tmp,len);
if (p == NULL) {
_invalid_stream(L,rb);
invalid_stream(L,rb);
}
lua_pushlstring(L,p,len);
}
static void _unpack_one(lua_State *L, struct read_block *rb);
static void unpack_one(lua_State *L, struct read_block *rb);
static void
_unpack_table(lua_State *L, struct read_block *rb, int array_size) {
unpack_table(lua_State *L, struct read_block *rb, int array_size) {
if (array_size == MAX_COOKIE-1) {
uint8_t type = 0;
uint8_t *t = rb_read(rb, &type, 1);
if (t==NULL || (*t & 7) != TYPE_NUMBER) {
_invalid_stream(L,rb);
invalid_stream(L,rb);
}
array_size = (int)_get_number(L,rb,*t >> 3);
array_size = (int)get_number(L,rb,*t >> 3);
}
lua_createtable(L,array_size,0);
int i;
for (i=1;i<=array_size;i++) {
_unpack_one(L,rb);
unpack_one(L,rb);
lua_rawseti(L,-2,i);
}
for (;;) {
_unpack_one(L,rb);
unpack_one(L,rb);
if (lua_isnil(L,-1)) {
lua_pop(L,1);
return;
}
_unpack_one(L,rb);
unpack_one(L,rb);
lua_rawset(L,-3);
}
}
static void
_push_value(lua_State *L, struct read_block *rb, int type, int cookie) {
push_value(lua_State *L, struct read_block *rb, int type, int cookie) {
switch(type) {
case TYPE_NIL:
lua_pushnil(L);
@@ -493,82 +414,69 @@ _push_value(lua_State *L, struct read_block *rb, int type, int cookie) {
lua_pushboolean(L,cookie);
break;
case TYPE_NUMBER:
lua_pushnumber(L,_get_number(L,rb,cookie));
lua_pushnumber(L,get_number(L,rb,cookie));
break;
case TYPE_USERDATA:
lua_pushlightuserdata(L,_get_pointer(L,rb));
lua_pushlightuserdata(L,get_pointer(L,rb));
break;
case TYPE_SHORT_STRING:
_get_buffer(L,rb,cookie);
get_buffer(L,rb,cookie);
break;
case TYPE_LONG_STRING: {
uint32_t len;
if (cookie == 2) {
uint16_t *plen = rb_read(rb, &len, 2);
if (plen == NULL) {
_invalid_stream(L,rb);
invalid_stream(L,rb);
}
_get_buffer(L,rb,(int)*plen);
get_buffer(L,rb,(int)*plen);
} else {
if (cookie != 4) {
_invalid_stream(L,rb);
invalid_stream(L,rb);
}
uint32_t *plen = rb_read(rb, &len, 4);
if (plen == NULL) {
_invalid_stream(L,rb);
invalid_stream(L,rb);
}
_get_buffer(L,rb,(int)*plen);
get_buffer(L,rb,(int)*plen);
}
break;
}
case TYPE_TABLE: {
_unpack_table(L,rb,cookie);
unpack_table(L,rb,cookie);
break;
}
default: {
_invalid_stream(L,rb);
invalid_stream(L,rb);
break;
}
}
}
static void
_unpack_one(lua_State *L, struct read_block *rb) {
unpack_one(lua_State *L, struct read_block *rb) {
uint8_t type = 0;
uint8_t *t = rb_read(rb, &type, 1);
if (t==NULL) {
_invalid_stream(L, rb);
invalid_stream(L, rb);
}
_push_value(L, rb, *t & 0x7, *t>>3);
push_value(L, rb, *t & 0x7, *t>>3);
}
static void
_seri(lua_State *L, struct block *b) {
uint32_t len = 0;
memcpy(&len, b->buffer ,sizeof(len));
len -= 4;
seri(lua_State *L, struct block *b, int len) {
uint8_t * buffer = skynet_malloc(len);
uint8_t * ptr = buffer;
int sz = len;
if (len < BLOCK_SIZE - 4) {
memcpy(ptr, b->buffer+4, len);
} else {
memcpy(ptr, b->buffer+4, BLOCK_SIZE-4);
ptr += BLOCK_SIZE-4;
len -= BLOCK_SIZE-4;
b = b->next;
while(len>0) {
if (len >= BLOCK_SIZE) {
memcpy(ptr, b->buffer, BLOCK_SIZE);
ptr += BLOCK_SIZE;
len -= BLOCK_SIZE;
} else {
memcpy(ptr, b->buffer, len);
break;
}
while(len>0) {
if (len >= BLOCK_SIZE) {
memcpy(ptr, b->buffer, BLOCK_SIZE);
ptr += BLOCK_SIZE;
len -= BLOCK_SIZE;
b = b->next;
} else {
memcpy(ptr, b->buffer, len);
break;
}
}
@@ -611,7 +519,7 @@ _luaseri_unpack(lua_State *L) {
uint8_t *t = rb_read(&rb, &type, 1);
if (t==NULL)
break;
_push_value(L, &rb, *t & 0x7, *t>>3);
push_value(L, &rb, *t & 0x7, *t>>3);
}
// Need not free buffer
@@ -621,17 +529,15 @@ _luaseri_unpack(lua_State *L) {
int
_luaseri_pack(lua_State *L) {
struct block temp;
temp.next = NULL;
struct write_block wb;
wb_init(&wb, NULL);
_pack_from(L,&wb,0);
struct block * b = wb_close(&wb);
_seri(L,b);
wb_init(&wb, &temp);
pack_from(L,&wb,0);
assert(wb.head == &temp);
seri(L, &temp, wb.len);
while (b) {
struct block * next = b->next;
skynet_free(b);
b = next;
}
wb_free(&wb);
return 2;
}

View File

@@ -144,7 +144,8 @@ function suspend(co, result, command, param, size)
if not result then
local session = session_coroutine_id[co]
local addr = session_coroutine_address[co]
if session then
if session ~= 0 then
-- only call response error
c.send(addr, skynet.PTYPE_ERROR, session, "")
end
session_coroutine_id[co] = nil

View File

@@ -145,6 +145,8 @@ local function connect(id, func)
suspend(s)
if s.connected then
return id
else
socket_pool[id] = nil
end
end

View File

@@ -36,17 +36,17 @@ struct remote_message_header {
uint32_t session;
};
struct msg {
struct harbor_msg {
struct remote_message_header header;
void * buffer;
size_t size;
};
struct msg_queue {
struct harbor_msg_queue {
int size;
int head;
int tail;
struct msg * data;
struct harbor_msg * data;
};
struct keyvalue {
@@ -54,7 +54,7 @@ struct keyvalue {
char key[GLOBALNAME_LENGTH];
uint32_t hash;
uint32_t value;
struct msg_queue * queue;
struct harbor_msg_queue * queue;
};
struct hashmap {
@@ -69,7 +69,7 @@ struct hashmap {
struct slave {
int fd;
struct msg_queue *queue;
struct harbor_msg_queue *queue;
int status;
int length;
int read;
@@ -88,11 +88,11 @@ struct harbor {
// hash table
static void
push_queue_msg(struct msg_queue * queue, struct msg * m) {
push_queue_msg(struct harbor_msg_queue * queue, struct harbor_msg * m) {
// If there is only 1 free slot which is reserved to distinguish full/empty
// of circular buffer, expand it.
if (((queue->tail + 1) % queue->size) == queue->head) {
struct msg * new_buffer = skynet_malloc(queue->size * 2 * sizeof(struct msg));
struct harbor_msg * new_buffer = skynet_malloc(queue->size * 2 * sizeof(struct harbor_msg));
int i;
for (i=0;i<queue->size-1;i++) {
new_buffer[i] = queue->data[(i+queue->head) % queue->size];
@@ -103,46 +103,46 @@ push_queue_msg(struct msg_queue * queue, struct msg * m) {
queue->tail = queue->size - 1;
queue->size *= 2;
}
struct msg * slot = &queue->data[queue->tail];
struct harbor_msg * slot = &queue->data[queue->tail];
*slot = *m;
queue->tail = (queue->tail + 1) % queue->size;
}
static void
push_queue(struct msg_queue * queue, void * buffer, size_t sz, struct remote_message_header * header) {
struct msg m;
push_queue(struct harbor_msg_queue * queue, void * buffer, size_t sz, struct remote_message_header * header) {
struct harbor_msg m;
m.header = *header;
m.buffer = buffer;
m.size = sz;
push_queue_msg(queue, &m);
}
static struct msg *
pop_queue(struct msg_queue * queue) {
static struct harbor_msg *
pop_queue(struct harbor_msg_queue * queue) {
if (queue->head == queue->tail) {
return NULL;
}
struct msg * slot = &queue->data[queue->head];
struct harbor_msg * slot = &queue->data[queue->head];
queue->head = (queue->head + 1) % queue->size;
return slot;
}
static struct msg_queue *
static struct harbor_msg_queue *
new_queue() {
struct msg_queue * queue = skynet_malloc(sizeof(*queue));
struct harbor_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));
queue->data = skynet_malloc(DEFAULT_QUEUE_SIZE * sizeof(struct harbor_msg));
return queue;
}
static void
release_queue(struct msg_queue *queue) {
release_queue(struct harbor_msg_queue *queue) {
if (queue == NULL)
return;
struct msg * m;
struct harbor_msg * m;
while ((m=pop_queue(queue)) != NULL) {
skynet_free(m->buffer);
}
@@ -334,7 +334,7 @@ send_remote(struct skynet_context * ctx, int fd, const char * buffer, size_t sz,
static void
dispatch_name_queue(struct harbor *h, struct keyvalue * node) {
struct msg_queue * queue = node->queue;
struct harbor_msg_queue * queue = node->queue;
uint32_t handle = node->value;
int harbor_id = handle >> HANDLE_REMOTE_SHIFT;
assert(harbor_id != 0);
@@ -352,7 +352,7 @@ dispatch_name_queue(struct harbor *h, struct keyvalue * node) {
s->queue = node->queue;
node->queue = NULL;
} else {
struct msg * m;
struct harbor_msg * m;
while ((m = pop_queue(queue))!=NULL) {
push_queue_msg(s->queue, m);
}
@@ -360,7 +360,7 @@ dispatch_name_queue(struct harbor *h, struct keyvalue * node) {
}
return;
}
struct msg * m;
struct harbor_msg * m;
while ((m = pop_queue(queue)) != NULL) {
m->header.destination |= (handle & HANDLE_MASK);
send_remote(context, fd, m->buffer, m->size, &m->header);
@@ -373,11 +373,11 @@ dispatch_queue(struct harbor *h, int id) {
int fd = s->fd;
assert(fd != 0);
struct msg_queue *queue = s->queue;
struct harbor_msg_queue *queue = s->queue;
if (queue == NULL)
return;
struct msg * m;
struct harbor_msg * m;
while ((m = pop_queue(queue)) != NULL) {
send_remote(h->ctx, fd, m->buffer, m->size, &m->header);
}

View File

@@ -33,7 +33,7 @@ local function update(db, key, value, ...)
end
end
local function wakeup(db, key1, key2, value, ...)
local function wakeup(db, key1, key2, ...)
if key1 == nil then
return
end
@@ -43,7 +43,7 @@ local function wakeup(db, key1, key2, value, ...)
end
if q[mode] == "queue" then
db[key1] = nil
if value then
if key2 then
-- throw error because can't wake up a branch
for _,response in ipairs(q) do
response(false)
@@ -53,7 +53,7 @@ local function wakeup(db, key1, key2, value, ...)
end
else
-- it's branch
return wakeup(q , key2, value, ...)
return wakeup(q , key2, ...)
end
end

44
test/testresponse.lua Normal file
View File

@@ -0,0 +1,44 @@
local skynet = require "skynet"
local mode = ...
if mode == "TICK" then
-- this service whould response the request every 1s.
local response_queue = {}
local function response()
while true do
skynet.sleep(100) -- sleep 1s
for k,v in ipairs(response_queue) do
v(true, skynet.now()) -- true means succ, false means error
response_queue[k] = nil
end
end
end
skynet.start(function()
skynet.fork(response)
skynet.dispatch("lua", function()
table.insert(response_queue, skynet.response())
end)
end)
else
local function request(tick, i)
print(i, "call", skynet.now())
print(i, "response", skynet.call(tick, "lua"))
print(i, "end", skynet.now())
end
skynet.start(function()
local tick = skynet.newservice(SERVICE_NAME, "TICK")
for i=1,5 do
skynet.fork(request, tick, i)
skynet.sleep(10)
end
end)
end