mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-22 02:53:09 +00:00
Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8b7ba7bdbe | ||
|
|
165bed46a8 | ||
|
|
9bb8099008 | ||
|
|
98630badd5 | ||
|
|
d0c4a4012a | ||
|
|
04688e98a7 | ||
|
|
e0feea9995 | ||
|
|
55ea556060 | ||
|
|
4803e50977 |
10
HISTORY.md
10
HISTORY.md
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -145,6 +145,8 @@ local function connect(id, func)
|
||||
suspend(s)
|
||||
if s.connected then
|
||||
return id
|
||||
else
|
||||
socket_pool[id] = nil
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
@@ -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
44
test/testresponse.lua
Normal 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
|
||||
Reference in New Issue
Block a user