mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-22 11:03:12 +00:00
Compare commits
36 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f393b64124 | ||
|
|
9ea756ce4f | ||
|
|
f9438461eb | ||
|
|
0fb23f9ab8 | ||
|
|
0390818a02 | ||
|
|
e002796d2e | ||
|
|
b8a0e35f21 | ||
|
|
bd52c5fece | ||
|
|
d2cea9b70f | ||
|
|
7e24d46047 | ||
|
|
8078d777ab | ||
|
|
939ef564ce | ||
|
|
6bf8dd9a31 | ||
|
|
809e6c3a89 | ||
|
|
845acd3314 | ||
|
|
2cf69a6ef0 | ||
|
|
978e010e67 | ||
|
|
cde423cb73 | ||
|
|
cfe8b19c8a | ||
|
|
f69f6e8e70 | ||
|
|
d7b228f342 | ||
|
|
0954a906f4 | ||
|
|
023e4abd87 | ||
|
|
cab10edd1d | ||
|
|
87a02ffe72 | ||
|
|
ae5a9bb855 | ||
|
|
c48e0e4c32 | ||
|
|
0747a7d7e1 | ||
|
|
4b0a648b21 | ||
|
|
40519e9ee1 | ||
|
|
ce5adba5b2 | ||
|
|
c78466b486 | ||
|
|
deed51d309 | ||
|
|
68a952ff17 | ||
|
|
434cd1ca79 | ||
|
|
8e09fad233 |
17
HISTORY.md
17
HISTORY.md
@@ -1,3 +1,20 @@
|
||||
v0.6.0 (2014-8-18)
|
||||
-----------
|
||||
* add sharedata
|
||||
* bugfix: service exit before init would not report back
|
||||
* add skynet.response and check multicall skynet.ret
|
||||
* skynet.newservice throw error when lanuch faild
|
||||
* Don't check imported function in snax.hotfix
|
||||
* snax service add change SERVICE_PATH and add it to package.path
|
||||
* skynet.redirect support string address
|
||||
* bugfix: skynet.harbor.link may block
|
||||
* add skynet.harbor.queryname to query globalname
|
||||
* add cluster.proxy
|
||||
* add DEBUG command exit (send a message to lua service by DEBUG)
|
||||
* add DEBUG command run (debug_console command inject)
|
||||
* bugfix : socketchannel connect once
|
||||
* bugfix : mongo driver
|
||||
|
||||
v0.5.2 (2014-8-11)
|
||||
-----------
|
||||
* Bugfix : httpd request
|
||||
|
||||
8
Makefile
8
Makefile
@@ -43,7 +43,7 @@ jemalloc : $(MALLOC_STATICLIB)
|
||||
CSERVICE = snlua logger gate harbor
|
||||
LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack \
|
||||
cjson clientsocket memory profile multicast \
|
||||
cluster crypt
|
||||
cluster crypt sharedata stm
|
||||
|
||||
SKYNET_SRC = skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c \
|
||||
skynet_server.c skynet_start.c skynet_timer.c skynet_error.c \
|
||||
@@ -113,6 +113,12 @@ $(LUA_CLIB_PATH)/cluster.so : lualib-src/lua-cluster.c | $(LUA_CLIB_PATH)
|
||||
$(LUA_CLIB_PATH)/crypt.so : lualib-src/lua-crypt.c | $(LUA_CLIB_PATH)
|
||||
$(CC) $(CFLAGS) $(SHARED) $^ -o $@
|
||||
|
||||
$(LUA_CLIB_PATH)/sharedata.so : lualib-src/lua-sharedata.c | $(LUA_CLIB_PATH)
|
||||
$(CC) $(CFLAGS) $(SHARED) $^ -o $@
|
||||
|
||||
$(LUA_CLIB_PATH)/stm.so : lualib-src/lua-stm.c | $(LUA_CLIB_PATH)
|
||||
$(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@
|
||||
|
||||
clean :
|
||||
rm -f $(SKYNET_BUILD_PATH)/skynet $(CSERVICE_PATH)/*.so $(LUA_CLIB_PATH)/*.so
|
||||
|
||||
|
||||
@@ -2,8 +2,9 @@ local skynet = require "skynet"
|
||||
local cluster = require "cluster"
|
||||
|
||||
skynet.start(function()
|
||||
skynet.newservice("simpledb")
|
||||
print(skynet.call("SIMPLEDB", "lua", "SET", "a", "foobar"))
|
||||
print(skynet.call("SIMPLEDB", "lua", "GET", "a"))
|
||||
local sdb = skynet.newservice("simpledb")
|
||||
skynet.name(".simpledb", sdb)
|
||||
print(skynet.call(".simpledb", "lua", "SET", "a", "foobar"))
|
||||
print(skynet.call(".simpledb", "lua", "GET", "a"))
|
||||
cluster.open "db"
|
||||
end)
|
||||
|
||||
@@ -2,5 +2,7 @@ local skynet = require "skynet"
|
||||
local cluster = require "cluster"
|
||||
|
||||
skynet.start(function()
|
||||
print(cluster.call("db", "SIMPLEDB", "GET", "a"))
|
||||
local proxy = cluster.proxy("db", ".simpledb")
|
||||
print(skynet.call(proxy, "lua", "GET", "a"))
|
||||
print(cluster.call("db", ".simpledb", "GET", "a"))
|
||||
end)
|
||||
|
||||
@@ -4,5 +4,6 @@ skynet.start(function()
|
||||
skynet.dispatch("lua", function(session, address, ...)
|
||||
print("[GLOBALLOG]", skynet.address(address), ...)
|
||||
end)
|
||||
skynet.register ".log"
|
||||
skynet.register "LOG"
|
||||
end)
|
||||
|
||||
72
examples/share.lua
Normal file
72
examples/share.lua
Normal file
@@ -0,0 +1,72 @@
|
||||
local skynet = require "skynet"
|
||||
local sharedata = require "sharedata"
|
||||
|
||||
local mode = ...
|
||||
|
||||
if mode == "host" then
|
||||
|
||||
skynet.start(function()
|
||||
skynet.error("new foobar")
|
||||
sharedata.new("foobar", { a=1, b= { "hello", "world" } })
|
||||
|
||||
skynet.fork(function()
|
||||
skynet.sleep(200) -- sleep 2s
|
||||
skynet.error("update foobar a = 2")
|
||||
sharedata.update("foobar", { a =2 })
|
||||
skynet.sleep(200) -- sleep 2s
|
||||
skynet.error("update foobar a = 3")
|
||||
sharedata.update("foobar", { a = 3, b = { "change" } })
|
||||
skynet.sleep(100)
|
||||
skynet.error("delete foobar")
|
||||
sharedata.delete "foobar"
|
||||
end)
|
||||
end)
|
||||
|
||||
else
|
||||
|
||||
|
||||
skynet.start(function()
|
||||
skynet.newservice(SERVICE_NAME, "host")
|
||||
|
||||
local obj = sharedata.query "foobar"
|
||||
|
||||
local b = obj.b
|
||||
skynet.error(string.format("a=%d", obj.a))
|
||||
|
||||
for k,v in ipairs(b) do
|
||||
skynet.error(string.format("b[%d]=%s", k,v))
|
||||
end
|
||||
|
||||
for i = 1, 5 do
|
||||
skynet.sleep(100)
|
||||
skynet.error("second " ..i)
|
||||
for k,v in pairs(obj) do
|
||||
skynet.error(string.format("%s = %s", k , tostring(v)))
|
||||
end
|
||||
end
|
||||
|
||||
local ok, err = pcall(function()
|
||||
local tmp = { b[1], b[2] } -- b is invalid , so pcall should failed
|
||||
end)
|
||||
|
||||
if not ok then
|
||||
skynet.error(err)
|
||||
end
|
||||
|
||||
-- obj. b is not the same with local b
|
||||
for k,v in ipairs(obj.b) do
|
||||
skynet.error(string.format("b[%d] = %s", k , tostring(v)))
|
||||
end
|
||||
|
||||
collectgarbage()
|
||||
skynet.error("sleep")
|
||||
skynet.sleep(100)
|
||||
b = nil
|
||||
collectgarbage()
|
||||
skynet.error("sleep")
|
||||
skynet.sleep(100)
|
||||
|
||||
skynet.exit()
|
||||
end)
|
||||
|
||||
end
|
||||
@@ -203,7 +203,7 @@ lunpackresponse(lua_State *L) {
|
||||
}
|
||||
|
||||
int
|
||||
luaopen_cluster_c(lua_State *L) {
|
||||
luaopen_cluster_core(lua_State *L) {
|
||||
luaL_Reg l[] = {
|
||||
{ "packrequest", lpackrequest },
|
||||
{ "unpackrequest", lunpackrequest },
|
||||
|
||||
@@ -507,8 +507,8 @@ op_insert(lua_State *L) {
|
||||
for (i=1;i<=s;i++) {
|
||||
lua_rawgeti(L,3,i);
|
||||
document doc = lua_touserdata(L,-1);
|
||||
lua_pop(L,1); // must call lua_pop before luaL_addlstring, because addlstring may change stack top
|
||||
luaL_addlstring(&b, (const char *)doc, get_length(doc));
|
||||
lua_pop(L,1);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -149,7 +149,7 @@ mc_nextid(lua_State *L) {
|
||||
}
|
||||
|
||||
int
|
||||
luaopen_multicast_c(lua_State *L) {
|
||||
luaopen_multicast_core(lua_State *L) {
|
||||
luaL_Reg l[] = {
|
||||
{ "pack", mc_packlocal },
|
||||
{ "unpack", mc_unpacklocal },
|
||||
|
||||
789
lualib-src/lua-sharedata.c
Normal file
789
lualib-src/lua-sharedata.c
Normal file
@@ -0,0 +1,789 @@
|
||||
// build: gcc -O2 -Wall --shared -o conf.so luaconf.c
|
||||
|
||||
#include <lua.h>
|
||||
#include <lauxlib.h>
|
||||
#include <stdint.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <assert.h>
|
||||
|
||||
#define KEYTYPE_INTEGER 0
|
||||
#define KEYTYPE_STRING 1
|
||||
|
||||
#define VALUETYPE_NIL 0
|
||||
#define VALUETYPE_NUMBER 1
|
||||
#define VALUETYPE_STRING 2
|
||||
#define VALUETYPE_BOOLEAN 3
|
||||
#define VALUETYPE_TABLE 4
|
||||
|
||||
struct table;
|
||||
|
||||
union value {
|
||||
lua_Number n;
|
||||
struct table * tbl;
|
||||
int string;
|
||||
int boolean;
|
||||
};
|
||||
|
||||
struct node {
|
||||
union value v;
|
||||
int key; // integer key or index of string table
|
||||
int next; // next slot index
|
||||
uint32_t keyhash;
|
||||
uint8_t keytype; // key type must be integer or string
|
||||
uint8_t valuetype; // value type can be number/string/boolean/table
|
||||
uint8_t nocolliding; // 0 means colliding slot
|
||||
};
|
||||
|
||||
struct table;
|
||||
|
||||
struct state {
|
||||
int dirty;
|
||||
int ref;
|
||||
struct table * root;
|
||||
};
|
||||
|
||||
struct table {
|
||||
int sizearray;
|
||||
int sizehash;
|
||||
uint8_t *arraytype;
|
||||
union value * array;
|
||||
struct node * hash;
|
||||
lua_State * L;
|
||||
};
|
||||
|
||||
struct context {
|
||||
lua_State * L;
|
||||
struct table * tbl;
|
||||
int string_index;
|
||||
};
|
||||
|
||||
struct ctrl {
|
||||
struct table * root;
|
||||
struct table * update;
|
||||
};
|
||||
|
||||
static int
|
||||
countsize(lua_State *L, int sizearray) {
|
||||
int n = 0;
|
||||
lua_pushnil(L);
|
||||
while (lua_next(L, 1) != 0) {
|
||||
int type = lua_type(L, -2);
|
||||
++n;
|
||||
if (type == LUA_TNUMBER) {
|
||||
lua_Number key = lua_tonumber(L, -2);
|
||||
int nkey = (int)key;
|
||||
if ((lua_Number)nkey != key) {
|
||||
luaL_error(L, "Invalid key %f", key);
|
||||
}
|
||||
if (nkey > 0 && nkey <= sizearray) {
|
||||
--n;
|
||||
}
|
||||
} else if (type != LUA_TSTRING && type != LUA_TTABLE) {
|
||||
luaL_error(L, "Invalid key type %s", lua_typename(L, type));
|
||||
}
|
||||
lua_pop(L, 1);
|
||||
}
|
||||
|
||||
return n;
|
||||
}
|
||||
|
||||
static uint32_t
|
||||
calchash(const char * str, size_t l) {
|
||||
uint32_t h = (uint32_t)l;
|
||||
size_t l1;
|
||||
size_t step = (l >> 5) + 1;
|
||||
for (l1 = l; l1 >= step; l1 -= step) {
|
||||
h = h ^ ((h<<5) + (h>>2) + (uint8_t)(str[l1 - 1]));
|
||||
}
|
||||
return h;
|
||||
}
|
||||
|
||||
static int
|
||||
stringindex(struct context *ctx, const char * str, size_t sz) {
|
||||
lua_State *L = ctx->L;
|
||||
lua_pushlstring(L, str, sz);
|
||||
lua_pushvalue(L, -1);
|
||||
lua_rawget(L, 1);
|
||||
int index;
|
||||
// stringmap(1) str index
|
||||
if (lua_isnil(L, -1)) {
|
||||
index = ++ctx->string_index;
|
||||
lua_pop(L, 1);
|
||||
lua_pushinteger(L, index);
|
||||
lua_rawset(L, 1);
|
||||
} else {
|
||||
index = lua_tointeger(L, -1);
|
||||
lua_pop(L, 2);
|
||||
}
|
||||
return index;
|
||||
}
|
||||
|
||||
static int convtable(lua_State *L);
|
||||
|
||||
static void
|
||||
setvalue(struct context * ctx, lua_State *L, int index, struct node *n) {
|
||||
int vt = lua_type(L, index);
|
||||
switch(vt) {
|
||||
case LUA_TNIL:
|
||||
n->valuetype = VALUETYPE_NIL;
|
||||
break;
|
||||
case LUA_TNUMBER:
|
||||
n->v.n = lua_tonumber(L, index);
|
||||
n->valuetype = VALUETYPE_NUMBER;
|
||||
break;
|
||||
case LUA_TSTRING: {
|
||||
size_t sz = 0;
|
||||
const char * str = lua_tolstring(L, index, &sz);
|
||||
n->v.string = stringindex(ctx, str, sz);
|
||||
n->valuetype = VALUETYPE_STRING;
|
||||
break;
|
||||
}
|
||||
case LUA_TBOOLEAN:
|
||||
n->v.boolean = lua_toboolean(L, index);
|
||||
n->valuetype = VALUETYPE_BOOLEAN;
|
||||
break;
|
||||
case LUA_TTABLE: {
|
||||
struct table *tbl = ctx->tbl;
|
||||
ctx->tbl = (struct table *)malloc(sizeof(struct table));
|
||||
if (ctx->tbl == NULL) {
|
||||
ctx->tbl = tbl;
|
||||
luaL_error(L, "memory error");
|
||||
// never get here
|
||||
}
|
||||
memset(ctx->tbl, 0, sizeof(struct table));
|
||||
int absidx = lua_absindex(L, index);
|
||||
|
||||
lua_pushcfunction(L, convtable);
|
||||
lua_pushvalue(L, absidx);
|
||||
lua_pushlightuserdata(L, ctx);
|
||||
|
||||
lua_call(L, 2, 0);
|
||||
|
||||
n->v.tbl = ctx->tbl;
|
||||
n->valuetype = VALUETYPE_TABLE;
|
||||
|
||||
ctx->tbl = tbl;
|
||||
|
||||
break;
|
||||
}
|
||||
default:
|
||||
luaL_error(L, "Unsupport value type %s", lua_typename(L, vt));
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
setarray(struct context *ctx, lua_State *L, int index, int key) {
|
||||
struct node n;
|
||||
setvalue(ctx, L, index, &n);
|
||||
struct table *tbl = ctx->tbl;
|
||||
--key; // base 0
|
||||
tbl->arraytype[key] = n.valuetype;
|
||||
tbl->array[key] = n.v;
|
||||
}
|
||||
|
||||
static int
|
||||
ishashkey(struct context * ctx, lua_State *L, int index, int *key, uint32_t *keyhash, int *keytype) {
|
||||
int sizearray = ctx->tbl->sizearray;
|
||||
int kt = lua_type(L, index);
|
||||
if (kt == LUA_TNUMBER) {
|
||||
*key = lua_tointeger(L, index);
|
||||
if (*key > 0 && *key <= sizearray) {
|
||||
return 0;
|
||||
}
|
||||
*keyhash = (uint32_t)*key;
|
||||
*keytype = KEYTYPE_INTEGER;
|
||||
} else {
|
||||
size_t sz = 0;
|
||||
const char * s = lua_tolstring(L, index, &sz);
|
||||
*keyhash = calchash(s, sz);
|
||||
*key = stringindex(ctx, s, sz);
|
||||
*keytype = KEYTYPE_STRING;
|
||||
}
|
||||
return 1;
|
||||
}
|
||||
|
||||
static void
|
||||
fillnocolliding(lua_State *L, struct context *ctx) {
|
||||
struct table * tbl = ctx->tbl;
|
||||
lua_pushnil(L);
|
||||
while (lua_next(L, 1) != 0) {
|
||||
int key;
|
||||
int keytype;
|
||||
uint32_t keyhash;
|
||||
if (!ishashkey(ctx, L, -2, &key, &keyhash, &keytype)) {
|
||||
setarray(ctx, L, -1, key);
|
||||
} else {
|
||||
struct node * n = &tbl->hash[keyhash % tbl->sizehash];
|
||||
if (n->valuetype == VALUETYPE_NIL) {
|
||||
n->key = key;
|
||||
n->keytype = keytype;
|
||||
n->keyhash = keyhash;
|
||||
n->next = -1;
|
||||
n->nocolliding = 1;
|
||||
setvalue(ctx, L, -1, n); // set n->v , n->valuetype
|
||||
}
|
||||
}
|
||||
lua_pop(L,1);
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
fillcolliding(lua_State *L, struct context *ctx) {
|
||||
struct table * tbl = ctx->tbl;
|
||||
int sizehash = tbl->sizehash;
|
||||
int emptyslot = 0;
|
||||
int i;
|
||||
lua_pushnil(L);
|
||||
while (lua_next(L, 1) != 0) {
|
||||
int key;
|
||||
int keytype;
|
||||
uint32_t keyhash;
|
||||
if (ishashkey(ctx, L, -2, &key, &keyhash, &keytype)) {
|
||||
struct node * mainpos = &tbl->hash[keyhash % tbl->sizehash];
|
||||
if (!(mainpos->keytype == keytype && mainpos->key == key)) {
|
||||
// the key has not insert
|
||||
struct node * n = NULL;
|
||||
for (i=emptyslot;i<sizehash;i++) {
|
||||
if (tbl->hash[i].valuetype == VALUETYPE_NIL) {
|
||||
n = &tbl->hash[i];
|
||||
break;
|
||||
}
|
||||
}
|
||||
assert(n);
|
||||
n->next = mainpos->next;
|
||||
mainpos->next = n - tbl->hash;
|
||||
mainpos->nocolliding = 0;
|
||||
n->key = key;
|
||||
n->keytype = keytype;
|
||||
n->keyhash = keyhash;
|
||||
n->nocolliding = 0;
|
||||
setvalue(ctx, L, -1, n); // set n->v , n->valuetype
|
||||
}
|
||||
}
|
||||
lua_pop(L,1);
|
||||
}
|
||||
}
|
||||
|
||||
// table need convert
|
||||
// struct context * ctx
|
||||
static int
|
||||
convtable(lua_State *L) {
|
||||
int i;
|
||||
struct context *ctx = lua_touserdata(L,2);
|
||||
struct table *tbl = ctx->tbl;
|
||||
|
||||
tbl->L = ctx->L;
|
||||
|
||||
int sizearray = lua_rawlen(L, 1);
|
||||
if (sizearray) {
|
||||
tbl->arraytype = (uint8_t *)malloc(sizearray * sizeof(uint8_t));
|
||||
if (tbl->arraytype == NULL) {
|
||||
goto memerror;
|
||||
}
|
||||
for (i=0;i<sizearray;i++) {
|
||||
tbl->arraytype[i] = VALUETYPE_NIL;
|
||||
}
|
||||
tbl->array = (union value *)malloc(sizearray * sizeof(union value));
|
||||
if (tbl->array == NULL) {
|
||||
goto memerror;
|
||||
}
|
||||
tbl->sizearray = sizearray;
|
||||
}
|
||||
int sizehash = countsize(L, sizearray);
|
||||
if (sizehash) {
|
||||
tbl->hash = (struct node *)malloc(sizehash * sizeof(struct node));
|
||||
if (tbl->hash == NULL) {
|
||||
goto memerror;
|
||||
}
|
||||
for (i=0;i<sizehash;i++) {
|
||||
tbl->hash[i].valuetype = VALUETYPE_NIL;
|
||||
tbl->hash[i].nocolliding = 0;
|
||||
}
|
||||
tbl->sizehash = sizehash;
|
||||
|
||||
fillnocolliding(L, ctx);
|
||||
fillcolliding(L, ctx);
|
||||
} else {
|
||||
int i;
|
||||
for (i=1;i<=sizearray;i++) {
|
||||
lua_rawgeti(L, 1, i);
|
||||
setarray(ctx, L, -1, i);
|
||||
lua_pop(L,1);
|
||||
}
|
||||
}
|
||||
|
||||
return 0;
|
||||
memerror:
|
||||
return luaL_error(L, "memory error");
|
||||
}
|
||||
|
||||
static void
|
||||
delete_tbl(struct table *tbl) {
|
||||
int i;
|
||||
for (i=0;i<tbl->sizearray;i++) {
|
||||
if (tbl->arraytype[i] == VALUETYPE_TABLE) {
|
||||
delete_tbl(tbl->array[i].tbl);
|
||||
}
|
||||
}
|
||||
for (i=0;i<tbl->sizehash;i++) {
|
||||
if (tbl->hash[i].valuetype == VALUETYPE_TABLE) {
|
||||
delete_tbl(tbl->hash[i].v.tbl);
|
||||
}
|
||||
}
|
||||
free(tbl->arraytype);
|
||||
free(tbl->array);
|
||||
free(tbl->hash);
|
||||
free(tbl);
|
||||
}
|
||||
|
||||
static int
|
||||
pconv(lua_State *L) {
|
||||
struct context *ctx = lua_touserdata(L,1);
|
||||
lua_State * pL = lua_touserdata(L, 2);
|
||||
int ret;
|
||||
|
||||
lua_settop(L, 0);
|
||||
|
||||
// init L (may throw memory error)
|
||||
// create a table for string map
|
||||
lua_newtable(L);
|
||||
|
||||
lua_pushcfunction(pL, convtable);
|
||||
lua_pushvalue(pL,1);
|
||||
lua_pushlightuserdata(pL, ctx);
|
||||
|
||||
ret = lua_pcall(pL, 2, 0, 0);
|
||||
if (ret != LUA_OK) {
|
||||
size_t sz = 0;
|
||||
const char * error = lua_tolstring(pL, -1, &sz);
|
||||
lua_pushlstring(L, error, sz);
|
||||
lua_error(L);
|
||||
// never get here
|
||||
}
|
||||
|
||||
luaL_checkstack(L, ctx->string_index + 3, NULL);
|
||||
lua_settop(L,1);
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
static void
|
||||
convert_stringmap(struct context *ctx, struct table *tbl) {
|
||||
lua_State *L = ctx->L;
|
||||
lua_settop(L, ctx->string_index + 1);
|
||||
lua_pushvalue(L, 1);
|
||||
struct state * s = lua_newuserdata(L, sizeof(*s));
|
||||
s->dirty = 0;
|
||||
s->ref = 0;
|
||||
s->root = tbl;
|
||||
lua_replace(L, 1);
|
||||
lua_replace(L, -2);
|
||||
|
||||
lua_pushnil(L);
|
||||
// ... stringmap nil
|
||||
while (lua_next(L, -2) != 0) {
|
||||
int idx = lua_tointeger(L, -1);
|
||||
lua_pop(L, 1);
|
||||
lua_pushvalue(L, -1);
|
||||
lua_replace(L, idx);
|
||||
}
|
||||
|
||||
lua_pop(L, 1);
|
||||
|
||||
lua_gc(L, LUA_GCCOLLECT, 0);
|
||||
}
|
||||
|
||||
static int
|
||||
lnewconf(lua_State *L) {
|
||||
int ret;
|
||||
struct context ctx;
|
||||
struct table * tbl = NULL;
|
||||
luaL_checktype(L,1,LUA_TTABLE);
|
||||
ctx.L = luaL_newstate();
|
||||
ctx.tbl = NULL;
|
||||
ctx.string_index = 1; // 1 reserved for dirty flag
|
||||
if (ctx.L == NULL) {
|
||||
lua_pushliteral(L, "memory error");
|
||||
goto error;
|
||||
}
|
||||
tbl = (struct table *)malloc(sizeof(struct table));
|
||||
if (tbl == NULL) {
|
||||
// lua_pushliteral may fail because of memory error, close first.
|
||||
lua_close(ctx.L);
|
||||
ctx.L = NULL;
|
||||
lua_pushliteral(L, "memory error");
|
||||
goto error;
|
||||
}
|
||||
memset(tbl, 0, sizeof(struct table));
|
||||
ctx.tbl = tbl;
|
||||
|
||||
lua_pushcfunction(ctx.L, pconv);
|
||||
lua_pushlightuserdata(ctx.L , &ctx);
|
||||
lua_pushlightuserdata(ctx.L , L);
|
||||
|
||||
ret = lua_pcall(ctx.L, 2, 1, 0);
|
||||
|
||||
if (ret != LUA_OK) {
|
||||
size_t sz = 0;
|
||||
const char * error = lua_tolstring(ctx.L, -1, &sz);
|
||||
lua_pushlstring(L, error, sz);
|
||||
goto error;
|
||||
}
|
||||
|
||||
convert_stringmap(&ctx, tbl);
|
||||
|
||||
lua_pushlightuserdata(L, tbl);
|
||||
|
||||
return 1;
|
||||
error:
|
||||
if (ctx.L) {
|
||||
lua_close(ctx.L);
|
||||
}
|
||||
if (tbl) {
|
||||
delete_tbl(tbl);
|
||||
}
|
||||
lua_error(L);
|
||||
return -1;
|
||||
}
|
||||
|
||||
static struct table *
|
||||
get_table(lua_State *L, int index) {
|
||||
struct table *tbl = lua_touserdata(L,index);
|
||||
if (tbl == NULL) {
|
||||
luaL_error(L, "Need a conf object");
|
||||
}
|
||||
return tbl;
|
||||
}
|
||||
|
||||
static int
|
||||
ldeleteconf(lua_State *L) {
|
||||
struct table *tbl = get_table(L,1);
|
||||
lua_close(tbl->L);
|
||||
delete_tbl(tbl);
|
||||
return 0;
|
||||
}
|
||||
|
||||
static void
|
||||
pushvalue(lua_State *L, lua_State *sL, uint8_t vt, union value *v) {
|
||||
switch(vt) {
|
||||
case VALUETYPE_NUMBER:
|
||||
lua_pushnumber(L, v->n);
|
||||
break;
|
||||
case VALUETYPE_STRING: {
|
||||
size_t sz = 0;
|
||||
const char *str = lua_tolstring(sL, v->string, &sz);
|
||||
lua_pushlstring(L, str, sz);
|
||||
break;
|
||||
}
|
||||
case VALUETYPE_BOOLEAN:
|
||||
lua_pushboolean(L, v->boolean);
|
||||
break;
|
||||
case VALUETYPE_TABLE:
|
||||
lua_pushlightuserdata(L, v->tbl);
|
||||
break;
|
||||
default:
|
||||
lua_pushnil(L);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
static struct node *
|
||||
lookup_key(struct table *tbl, uint32_t keyhash, int key, int keytype, const char *str, size_t sz) {
|
||||
if (tbl->sizehash == 0)
|
||||
return NULL;
|
||||
struct node *n = &tbl->hash[keyhash % tbl->sizehash];
|
||||
if (keyhash != n->keyhash && n->nocolliding)
|
||||
return NULL;
|
||||
for (;;) {
|
||||
if (keyhash == n->keyhash) {
|
||||
if (n->keytype == KEYTYPE_INTEGER) {
|
||||
if (keytype == KEYTYPE_INTEGER && n->key == key) {
|
||||
return n;
|
||||
}
|
||||
} else {
|
||||
// n->keytype == KEYTYPE_STRING
|
||||
if (keytype == KEYTYPE_STRING) {
|
||||
size_t sz2 = 0;
|
||||
const char * str2 = lua_tolstring(tbl->L, n->key, &sz2);
|
||||
if (sz == sz2 && memcmp(str,str2,sz) == 0) {
|
||||
return n;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if (n->next < 0) {
|
||||
return NULL;
|
||||
}
|
||||
n = &tbl->hash[n->next];
|
||||
}
|
||||
}
|
||||
|
||||
static int
|
||||
lindexconf(lua_State *L) {
|
||||
struct table *tbl = get_table(L,1);
|
||||
int kt = lua_type(L,2);
|
||||
uint32_t keyhash;
|
||||
int key = 0;
|
||||
int keytype;
|
||||
size_t sz = 0;
|
||||
const char * str = NULL;
|
||||
if (kt == LUA_TNUMBER) {
|
||||
lua_Number k = lua_tonumber(L, 2);
|
||||
key = (int)k;
|
||||
if ((lua_Number)key != k) {
|
||||
return luaL_error(L, "Invalid key %f", k);
|
||||
}
|
||||
if (key > 0 && key <= tbl->sizearray) {
|
||||
--key;
|
||||
pushvalue(L, tbl->L, tbl->arraytype[key], &tbl->array[key]);
|
||||
return 1;
|
||||
}
|
||||
keytype = KEYTYPE_INTEGER;
|
||||
keyhash = (uint32_t)key;
|
||||
} else {
|
||||
str = luaL_checklstring(L, 2, &sz);
|
||||
keyhash = calchash(str, sz);
|
||||
keytype = KEYTYPE_STRING;
|
||||
}
|
||||
|
||||
struct node *n = lookup_key(tbl, keyhash, key, keytype, str, sz);
|
||||
if (n) {
|
||||
pushvalue(L, tbl->L, n->valuetype, &n->v);
|
||||
return 1;
|
||||
} else {
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
pushkey(lua_State *L, lua_State *sL, struct node *n) {
|
||||
if (n->keytype == KEYTYPE_INTEGER) {
|
||||
lua_pushinteger(L, n->key);
|
||||
} else {
|
||||
size_t sz = 0;
|
||||
const char * str = lua_tolstring(sL, n->key, &sz);
|
||||
lua_pushlstring(L, str, sz);
|
||||
}
|
||||
}
|
||||
|
||||
static int
|
||||
pushfirsthash(lua_State *L, struct table * tbl) {
|
||||
if (tbl->sizehash) {
|
||||
pushkey(L, tbl->L, &tbl->hash[0]);
|
||||
return 1;
|
||||
} else {
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
static int
|
||||
lnextkey(lua_State *L) {
|
||||
struct table *tbl = get_table(L,1);
|
||||
if (lua_isnoneornil(L,2)) {
|
||||
if (tbl->sizearray > 0) {
|
||||
int i;
|
||||
for (i=0;i<tbl->sizearray;i++) {
|
||||
if (tbl->arraytype[i] != VALUETYPE_NIL) {
|
||||
lua_pushinteger(L, i+1);
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
return pushfirsthash(L, tbl);
|
||||
}
|
||||
int kt = lua_type(L,2);
|
||||
uint32_t keyhash;
|
||||
int key = 0;
|
||||
int keytype;
|
||||
size_t sz=0;
|
||||
const char *str = NULL;
|
||||
int sizearray = tbl->sizearray;
|
||||
if (kt == LUA_TNUMBER) {
|
||||
lua_Number k = lua_tonumber(L,2);
|
||||
key = (int)k;
|
||||
if ((lua_Number)key != k) {
|
||||
return 0;
|
||||
}
|
||||
if (key > 0 && key <= sizearray) {
|
||||
int i;
|
||||
for (i=key;i<sizearray;i++) {
|
||||
if (tbl->arraytype[i] != VALUETYPE_NIL) {
|
||||
lua_pushinteger(L, i+1);
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
return pushfirsthash(L, tbl);
|
||||
}
|
||||
keyhash = (uint32_t)key;
|
||||
keytype = KEYTYPE_INTEGER;
|
||||
} else {
|
||||
str = luaL_checklstring(L, 2, &sz);
|
||||
keyhash = calchash(str, sz);
|
||||
keytype = KEYTYPE_STRING;
|
||||
}
|
||||
|
||||
struct node *n = lookup_key(tbl, keyhash, key, keytype, str, sz);
|
||||
if (n) {
|
||||
++n;
|
||||
int index = n-tbl->hash;
|
||||
if (index == tbl->sizehash) {
|
||||
return 0;
|
||||
}
|
||||
pushkey(L, tbl->L, n);
|
||||
return 1;
|
||||
} else {
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
static int
|
||||
llen(lua_State *L) {
|
||||
struct table *tbl = get_table(L,1);
|
||||
lua_pushinteger(L, tbl->sizearray);
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
lhashlen(lua_State *L) {
|
||||
struct table *tbl = get_table(L,1);
|
||||
lua_pushinteger(L, tbl->sizehash);
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
releaseobj(lua_State *L) {
|
||||
struct ctrl *c = lua_touserdata(L, 1);
|
||||
struct table *tbl = c->root;
|
||||
struct state *s = lua_touserdata(tbl->L, 1);
|
||||
__sync_fetch_and_sub(&s->ref, 1);
|
||||
c->root = NULL;
|
||||
c->update = NULL;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int
|
||||
lboxconf(lua_State *L) {
|
||||
struct table * tbl = get_table(L,1);
|
||||
struct state * s = lua_touserdata(tbl->L, 1);
|
||||
__sync_fetch_and_add(&s->ref, 1);
|
||||
|
||||
struct ctrl * c = lua_newuserdata(L, sizeof(*c));
|
||||
c->root = tbl;
|
||||
c->update = NULL;
|
||||
if (luaL_newmetatable(L, "confctrl")) {
|
||||
lua_pushcfunction(L, releaseobj);
|
||||
lua_setfield(L, -2, "__gc");
|
||||
}
|
||||
lua_setmetatable(L, -2);
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
lmarkdirty(lua_State *L) {
|
||||
struct table *tbl = get_table(L,1);
|
||||
struct state * s = lua_touserdata(tbl->L, 1);
|
||||
s->dirty = 1;
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int
|
||||
lisdirty(lua_State *L) {
|
||||
struct table *tbl = get_table(L,1);
|
||||
struct state * s = lua_touserdata(tbl->L, 1);
|
||||
int d = s->dirty;
|
||||
lua_pushboolean(L, d);
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
lgetref(lua_State *L) {
|
||||
struct table *tbl = get_table(L,1);
|
||||
struct state * s = lua_touserdata(tbl->L, 1);
|
||||
lua_pushinteger(L , s->ref);
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
lincref(lua_State *L) {
|
||||
struct table *tbl = get_table(L,1);
|
||||
struct state * s = lua_touserdata(tbl->L, 1);
|
||||
int ref = __sync_add_and_fetch(&s->ref, 1);
|
||||
lua_pushinteger(L , ref);
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
ldecref(lua_State *L) {
|
||||
struct table *tbl = get_table(L,1);
|
||||
struct state * s = lua_touserdata(tbl->L, 1);
|
||||
int ref = __sync_sub_and_fetch(&s->ref, 1);
|
||||
lua_pushinteger(L , ref);
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
lneedupdate(lua_State *L) {
|
||||
struct ctrl * c = lua_touserdata(L, 1);
|
||||
if (c->update) {
|
||||
lua_pushlightuserdata(L, c->update);
|
||||
lua_getuservalue(L, 1);
|
||||
return 2;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int
|
||||
lupdate(lua_State *L) {
|
||||
luaL_checktype(L, 1, LUA_TUSERDATA);
|
||||
luaL_checktype(L, 2, LUA_TLIGHTUSERDATA);
|
||||
luaL_checktype(L, 3, LUA_TTABLE);
|
||||
struct ctrl * c= lua_touserdata(L, 1);
|
||||
struct table *n = lua_touserdata(L, 2);
|
||||
if (c->update) {
|
||||
return luaL_error(L, "can't update more than once");
|
||||
}
|
||||
if (c->root == n) {
|
||||
return luaL_error(L, "You should update a new object");
|
||||
}
|
||||
lua_settop(L, 3);
|
||||
lua_setuservalue(L, 1);
|
||||
c->update = n;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int
|
||||
luaopen_sharedata_core(lua_State *L) {
|
||||
luaL_Reg l[] = {
|
||||
// used by host
|
||||
{ "new", lnewconf },
|
||||
{ "delete", ldeleteconf },
|
||||
{ "markdirty", lmarkdirty },
|
||||
{ "getref", lgetref },
|
||||
{ "incref", lincref },
|
||||
{ "decref", ldecref },
|
||||
|
||||
// used by client
|
||||
{ "box", lboxconf },
|
||||
{ "index", lindexconf },
|
||||
{ "nextkey", lnextkey },
|
||||
{ "len", llen },
|
||||
{ "hashlen", lhashlen },
|
||||
{ "isdirty", lisdirty },
|
||||
{ "needupdate", lneedupdate },
|
||||
{ "update", lupdate },
|
||||
{ NULL, NULL },
|
||||
};
|
||||
luaL_checkversion(L);
|
||||
luaL_newlib(L, l);
|
||||
|
||||
return 1;
|
||||
}
|
||||
@@ -73,10 +73,17 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int
|
||||
forward_cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
|
||||
_cb(context, ud, type, session, source, msg, sz);
|
||||
// don't delete msg in forward mode.
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
_callback(lua_State *L) {
|
||||
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
|
||||
|
||||
int forward = lua_toboolean(L, 2);
|
||||
luaL_checktype(L,1,LUA_TFUNCTION);
|
||||
lua_settop(L,1);
|
||||
lua_rawsetp(L, LUA_REGISTRYINDEX, _cb);
|
||||
@@ -84,7 +91,11 @@ _callback(lua_State *L) {
|
||||
lua_rawgeti(L, LUA_REGISTRYINDEX, LUA_RIDX_MAINTHREAD);
|
||||
lua_State *gL = lua_tothread(L,-1);
|
||||
|
||||
skynet_callback(context, gL, _cb);
|
||||
if (forward) {
|
||||
skynet_callback(context, gL, forward_cb);
|
||||
} else {
|
||||
skynet_callback(context, gL, _cb);
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
@@ -115,44 +126,13 @@ _genid(lua_State *L) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
// copy from _send
|
||||
|
||||
static int
|
||||
_sendname(lua_State *L, struct skynet_context * context, const char * dest) {
|
||||
int type = luaL_checkinteger(L, 2);
|
||||
int session = 0;
|
||||
if (lua_isnil(L,3)) {
|
||||
type |= PTYPE_TAG_ALLOCSESSION;
|
||||
} else {
|
||||
session = luaL_checkinteger(L,3);
|
||||
static const char *
|
||||
get_dest_string(lua_State *L, int index) {
|
||||
const char * dest_string = lua_tostring(L, index);
|
||||
if (dest_string == NULL) {
|
||||
luaL_error(L, "dest address type (%s) must be a string or number.", lua_typename(L, lua_type(L,index)));
|
||||
}
|
||||
|
||||
int mtype = lua_type(L,4);
|
||||
switch (mtype) {
|
||||
case LUA_TSTRING: {
|
||||
size_t len = 0;
|
||||
void * msg = (void *)lua_tolstring(L,4,&len);
|
||||
session = skynet_sendname(context, dest, type, session , msg, len);
|
||||
break;
|
||||
}
|
||||
case LUA_TNIL :
|
||||
session = skynet_sendname(context, dest, type, session , NULL, 0);
|
||||
break;
|
||||
case LUA_TLIGHTUSERDATA: {
|
||||
luaL_checktype(L, 4, LUA_TLIGHTUSERDATA);
|
||||
void * msg = lua_touserdata(L,4);
|
||||
int size = luaL_checkinteger(L,5);
|
||||
session = skynet_sendname(context, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
|
||||
break;
|
||||
}
|
||||
default:
|
||||
luaL_error(L, "skynet.send invalid param %s", lua_typename(L,lua_type(L,4)));
|
||||
}
|
||||
if (session < 0) {
|
||||
return 0;
|
||||
}
|
||||
lua_pushinteger(L,session);
|
||||
return 1;
|
||||
return dest_string;
|
||||
}
|
||||
|
||||
/*
|
||||
@@ -167,28 +147,10 @@ _sendname(lua_State *L, struct skynet_context * context, const char * dest) {
|
||||
static int
|
||||
_send(lua_State *L) {
|
||||
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
|
||||
int addr_type = lua_type(L,1);
|
||||
uint32_t dest = 0;
|
||||
switch(addr_type) {
|
||||
case LUA_TNUMBER:
|
||||
dest = lua_tounsigned(L,1);
|
||||
break;
|
||||
case LUA_TSTRING: {
|
||||
const char * addrname = lua_tostring(L,1);
|
||||
if (addrname[0] == '.' || addrname[0] == ':') {
|
||||
dest = skynet_queryname(context, addrname);
|
||||
if (dest == 0) {
|
||||
luaL_error(L, "Invalid name %s", addrname);
|
||||
}
|
||||
} else if ('0' <= addrname[0] && addrname[0] <= '9') {
|
||||
luaL_error(L, "Invalid name %s: must not start with a digit", addrname);
|
||||
} else {
|
||||
return _sendname(L, context, addrname);
|
||||
}
|
||||
break;
|
||||
}
|
||||
default:
|
||||
return luaL_error(L, "address must be number or string, got %s",lua_typename(L,addr_type));
|
||||
uint32_t dest = lua_tounsigned(L, 1);
|
||||
const char * dest_string = NULL;
|
||||
if (dest == 0) {
|
||||
dest_string = get_dest_string(L, 1);
|
||||
}
|
||||
|
||||
int type = luaL_checkinteger(L, 2);
|
||||
@@ -207,13 +169,21 @@ _send(lua_State *L) {
|
||||
if (len == 0) {
|
||||
msg = NULL;
|
||||
}
|
||||
session = skynet_send(context, 0, dest, type, session , msg, len);
|
||||
if (dest_string) {
|
||||
session = skynet_sendname(context, 0, dest_string, type, session , msg, len);
|
||||
} else {
|
||||
session = skynet_send(context, 0, dest, type, session , msg, len);
|
||||
}
|
||||
break;
|
||||
}
|
||||
case LUA_TLIGHTUSERDATA: {
|
||||
void * msg = lua_touserdata(L,4);
|
||||
int size = luaL_checkinteger(L,5);
|
||||
session = skynet_send(context, 0, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
|
||||
if (dest_string) {
|
||||
session = skynet_sendname(context, 0, dest_string, type | PTYPE_TAG_DONTCOPY, session, msg, size);
|
||||
} else {
|
||||
session = skynet_send(context, 0, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
|
||||
}
|
||||
break;
|
||||
}
|
||||
default:
|
||||
@@ -231,7 +201,11 @@ _send(lua_State *L) {
|
||||
static int
|
||||
_redirect(lua_State *L) {
|
||||
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
|
||||
uint32_t dest = luaL_checkunsigned(L,1);
|
||||
uint32_t dest = lua_tounsigned(L,1);
|
||||
const char * dest_string = NULL;
|
||||
if (dest == 0) {
|
||||
dest_string = get_dest_string(L, 1);
|
||||
}
|
||||
uint32_t source = luaL_checkunsigned(L,2);
|
||||
int type = luaL_checkinteger(L,3);
|
||||
int session = luaL_checkinteger(L,4);
|
||||
@@ -244,13 +218,21 @@ _redirect(lua_State *L) {
|
||||
if (len == 0) {
|
||||
msg = NULL;
|
||||
}
|
||||
session = skynet_send(context, source, dest, type, session , msg, len);
|
||||
if (dest_string) {
|
||||
session = skynet_sendname(context, source, dest_string, type, session , msg, len);
|
||||
} else {
|
||||
session = skynet_send(context, source, dest, type, session , msg, len);
|
||||
}
|
||||
break;
|
||||
}
|
||||
case LUA_TLIGHTUSERDATA: {
|
||||
void * msg = lua_touserdata(L,5);
|
||||
int size = luaL_checkinteger(L,6);
|
||||
session = skynet_send(context, source, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
|
||||
if (dest_string) {
|
||||
session = skynet_sendname(context, source, dest_string, type | PTYPE_TAG_DONTCOPY, session, msg, size);
|
||||
} else {
|
||||
session = skynet_send(context, source, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
|
||||
}
|
||||
break;
|
||||
}
|
||||
default:
|
||||
@@ -299,8 +281,28 @@ lpackstring(lua_State *L) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
ltrash(lua_State *L) {
|
||||
int t = lua_type(L,1);
|
||||
switch (t) {
|
||||
case LUA_TSTRING: {
|
||||
break;
|
||||
}
|
||||
case LUA_TLIGHTUSERDATA: {
|
||||
void * msg = lua_touserdata(L,1);
|
||||
luaL_checkinteger(L,2);
|
||||
skynet_free(msg);
|
||||
break;
|
||||
}
|
||||
default:
|
||||
luaL_error(L, "skynet.trash invalid param %s", lua_typename(L,t));
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
int
|
||||
luaopen_skynet_c(lua_State *L) {
|
||||
luaopen_skynet_core(lua_State *L) {
|
||||
luaL_checkversion(L);
|
||||
|
||||
luaL_Reg l[] = {
|
||||
@@ -314,6 +316,7 @@ luaopen_skynet_c(lua_State *L) {
|
||||
{ "pack", _luaseri_pack },
|
||||
{ "unpack", _luaseri_unpack },
|
||||
{ "packstring", lpackstring },
|
||||
{ "trash" , ltrash },
|
||||
{ "callback", _callback },
|
||||
{ NULL, NULL },
|
||||
};
|
||||
|
||||
244
lualib-src/lua-stm.c
Normal file
244
lualib-src/lua-stm.c
Normal file
@@ -0,0 +1,244 @@
|
||||
#include <lua.h>
|
||||
#include <lauxlib.h>
|
||||
#include <stdlib.h>
|
||||
#include <stdint.h>
|
||||
#include <assert.h>
|
||||
|
||||
#include "rwlock.h"
|
||||
#include "skynet_malloc.h"
|
||||
|
||||
struct stm_object {
|
||||
struct rwlock lock;
|
||||
int reference;
|
||||
struct stm_copy * copy;
|
||||
};
|
||||
|
||||
struct stm_copy {
|
||||
int reference;
|
||||
uint32_t sz;
|
||||
void * msg;
|
||||
};
|
||||
|
||||
// msg should alloc by skynet_malloc
|
||||
static struct stm_copy *
|
||||
stm_newcopy(void * msg, int32_t sz) {
|
||||
struct stm_copy * copy = skynet_malloc(sizeof(*copy));
|
||||
copy->reference = 1;
|
||||
copy->sz = sz;
|
||||
copy->msg = msg;
|
||||
|
||||
return copy;
|
||||
}
|
||||
|
||||
static struct stm_object *
|
||||
stm_new(void * msg, int32_t sz) {
|
||||
struct stm_object * obj = skynet_malloc(sizeof(*obj));
|
||||
rwlock_init(&obj->lock);
|
||||
obj->reference = 1;
|
||||
obj->copy = stm_newcopy(msg, sz);
|
||||
|
||||
return obj;
|
||||
}
|
||||
|
||||
static void
|
||||
stm_releasecopy(struct stm_copy *copy) {
|
||||
if (copy == NULL)
|
||||
return;
|
||||
if (__sync_sub_and_fetch(©->reference, 1) == 0) {
|
||||
skynet_free(copy->msg);
|
||||
skynet_free(copy);
|
||||
}
|
||||
}
|
||||
|
||||
static void
|
||||
stm_release(struct stm_object *obj) {
|
||||
assert(obj->copy);
|
||||
rwlock_wlock(&obj->lock);
|
||||
// writer release the stm object, so release the last copy .
|
||||
stm_releasecopy(obj->copy);
|
||||
obj->copy = NULL;
|
||||
if (--obj->reference > 0) {
|
||||
// stm object grab by readers, reset the copy to NULL.
|
||||
rwlock_wunlock(&obj->lock);
|
||||
return;
|
||||
}
|
||||
// no one grab the stm object, no need to unlock wlock.
|
||||
skynet_free(obj);
|
||||
}
|
||||
|
||||
static void
|
||||
stm_releasereader(struct stm_object *obj) {
|
||||
rwlock_rlock(&obj->lock);
|
||||
if (__sync_sub_and_fetch(&obj->reference,1) == 0) {
|
||||
// last reader, no writer. so no need to unlock
|
||||
assert(obj->copy == NULL);
|
||||
skynet_free(obj);
|
||||
return;
|
||||
}
|
||||
rwlock_runlock(&obj->lock);
|
||||
}
|
||||
|
||||
static void
|
||||
stm_grab(struct stm_object *obj) {
|
||||
rwlock_rlock(&obj->lock);
|
||||
int ref = __sync_fetch_and_add(&obj->reference,1);
|
||||
rwlock_runlock(&obj->lock);
|
||||
assert(ref > 0);
|
||||
}
|
||||
|
||||
static struct stm_copy *
|
||||
stm_copy(struct stm_object *obj) {
|
||||
rwlock_rlock(&obj->lock);
|
||||
struct stm_copy * ret = obj->copy;
|
||||
if (ret) {
|
||||
int ref = __sync_fetch_and_add(&ret->reference,1);
|
||||
assert(ref > 0);
|
||||
}
|
||||
rwlock_runlock(&obj->lock);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
static void
|
||||
stm_update(struct stm_object *obj, void *msg, int32_t sz) {
|
||||
struct stm_copy *copy = stm_newcopy(msg, sz);
|
||||
rwlock_wlock(&obj->lock);
|
||||
struct stm_copy *oldcopy = obj->copy;
|
||||
obj->copy = copy;
|
||||
rwlock_wunlock(&obj->lock);
|
||||
|
||||
stm_releasecopy(oldcopy);
|
||||
}
|
||||
|
||||
// lua binding
|
||||
|
||||
struct boxstm {
|
||||
struct stm_object * obj;
|
||||
};
|
||||
|
||||
static int
|
||||
lcopy(lua_State *L) {
|
||||
struct boxstm * box = lua_touserdata(L, 1);
|
||||
stm_grab(box->obj);
|
||||
lua_pushlightuserdata(L, box->obj);
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
lnewwriter(lua_State *L) {
|
||||
void * msg = lua_touserdata(L, 1);
|
||||
uint32_t sz = luaL_checkunsigned(L, 2);
|
||||
struct boxstm * box = lua_newuserdata(L, sizeof(*box));
|
||||
box->obj = stm_new(msg,sz);
|
||||
lua_pushvalue(L, lua_upvalueindex(1));
|
||||
lua_setmetatable(L, -2);
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
ldeletewriter(lua_State *L) {
|
||||
struct boxstm * box = lua_touserdata(L, 1);
|
||||
stm_release(box->obj);
|
||||
box->obj = NULL;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int
|
||||
lupdate(lua_State *L) {
|
||||
struct boxstm * box = lua_touserdata(L, 1);
|
||||
void * msg = lua_touserdata(L, 2);
|
||||
uint32_t sz = luaL_checkunsigned(L, 3);
|
||||
stm_update(box->obj, msg, sz);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
struct boxreader {
|
||||
struct stm_object *obj;
|
||||
struct stm_copy *lastcopy;
|
||||
};
|
||||
|
||||
static int
|
||||
lnewreader(lua_State *L) {
|
||||
struct boxreader * box = lua_newuserdata(L, sizeof(*box));
|
||||
box->obj = lua_touserdata(L, 1);
|
||||
box->lastcopy = NULL;
|
||||
lua_pushvalue(L, lua_upvalueindex(1));
|
||||
lua_setmetatable(L, -2);
|
||||
|
||||
return 1;
|
||||
}
|
||||
|
||||
static int
|
||||
ldeletereader(lua_State *L) {
|
||||
struct boxreader * box = lua_touserdata(L, 1);
|
||||
stm_releasereader(box->obj);
|
||||
box->obj = NULL;
|
||||
stm_releasecopy(box->lastcopy);
|
||||
box->lastcopy = NULL;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
static int
|
||||
lread(lua_State *L) {
|
||||
struct boxreader * box = lua_touserdata(L, 1);
|
||||
luaL_checktype(L, 2, LUA_TFUNCTION);
|
||||
struct stm_copy * copy = stm_copy(box->obj);
|
||||
if (copy == box->lastcopy) {
|
||||
// not update
|
||||
stm_releasecopy(copy);
|
||||
lua_pushboolean(L, 0);
|
||||
return 1;
|
||||
}
|
||||
|
||||
stm_releasecopy(box->lastcopy);
|
||||
box->lastcopy = copy;
|
||||
if (copy) {
|
||||
lua_settop(L, 2);
|
||||
lua_pushlightuserdata(L, copy->msg);
|
||||
lua_pushunsigned(L, copy->sz);
|
||||
lua_call(L, 2, LUA_MULTRET);
|
||||
lua_pushboolean(L, 1);
|
||||
lua_replace(L, 1);
|
||||
return lua_gettop(L);
|
||||
} else {
|
||||
lua_pushboolean(L, 0);
|
||||
return 1;
|
||||
}
|
||||
}
|
||||
|
||||
int
|
||||
luaopen_stm(lua_State *L) {
|
||||
luaL_checkversion(L);
|
||||
lua_createtable(L, 0, 3);
|
||||
|
||||
lua_pushcfunction(L, lcopy);
|
||||
lua_setfield(L, -2, "copy");
|
||||
|
||||
luaL_Reg writer[] = {
|
||||
{ "new", lnewwriter },
|
||||
{ NULL, NULL },
|
||||
};
|
||||
lua_createtable(L, 0, 2);
|
||||
lua_pushcfunction(L, ldeletewriter),
|
||||
lua_setfield(L, -2, "__gc");
|
||||
lua_pushcfunction(L, lupdate),
|
||||
lua_setfield(L, -2, "__call");
|
||||
luaL_setfuncs(L, writer, 1);
|
||||
|
||||
luaL_Reg reader[] = {
|
||||
{ "newcopy", lnewreader },
|
||||
{ NULL, NULL },
|
||||
};
|
||||
lua_createtable(L, 0, 2);
|
||||
lua_pushcfunction(L, ldeletereader),
|
||||
lua_setfield(L, -2, "__gc");
|
||||
lua_pushcfunction(L, lread),
|
||||
lua_setfield(L, -2, "__call");
|
||||
luaL_setfuncs(L, reader, 1);
|
||||
|
||||
return 1;
|
||||
}
|
||||
@@ -4,7 +4,7 @@ local clusterd
|
||||
local cluster = {}
|
||||
|
||||
function cluster.call(node, address, ...)
|
||||
-- skynet.pack(...) will free by cluster.c.packrequest
|
||||
-- skynet.pack(...) will free by cluster.core.packrequest
|
||||
return skynet.call(clusterd, "lua", "req", node, address, skynet.pack(...))
|
||||
end
|
||||
|
||||
@@ -20,6 +20,10 @@ function cluster.reload()
|
||||
skynet.call(clusterd, "lua", "reload")
|
||||
end
|
||||
|
||||
function cluster.proxy(node, name)
|
||||
return skynet.call(clusterd, "lua", "proxy", node, name)
|
||||
end
|
||||
|
||||
skynet.init(function()
|
||||
clusterd = skynet.uniqueservice("clusterd")
|
||||
end)
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
-- This is a deprecated module, use skynet.queue instead.
|
||||
|
||||
local skynet = require "skynet"
|
||||
local c = require "skynet.c"
|
||||
local c = require "skynet.core"
|
||||
|
||||
local mqueue = {}
|
||||
local init_once
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
local skynet = require "skynet"
|
||||
local mc = require "multicast.c"
|
||||
local mc = require "multicast.core"
|
||||
|
||||
local multicastd
|
||||
local multicast = {}
|
||||
|
||||
43
lualib/sharedata.lua
Normal file
43
lualib/sharedata.lua
Normal file
@@ -0,0 +1,43 @@
|
||||
local skynet = require "skynet"
|
||||
local sd = require "sharedata.corelib"
|
||||
|
||||
local service
|
||||
|
||||
skynet.init(function()
|
||||
service = skynet.uniqueservice "sharedatad"
|
||||
end)
|
||||
|
||||
local sharedata = {}
|
||||
|
||||
local function monitor(name, obj, cobj)
|
||||
local newobj = cobj
|
||||
while true do
|
||||
newobj = skynet.call(service, "lua", "monitor", name, newobj)
|
||||
if newobj == nil then
|
||||
break
|
||||
end
|
||||
sd.update(obj, newobj)
|
||||
end
|
||||
end
|
||||
|
||||
function sharedata.query(name)
|
||||
local obj = skynet.call(service, "lua", "query", name)
|
||||
local r = sd.box(obj)
|
||||
skynet.send(service, "lua", "confirm" , obj)
|
||||
skynet.fork(monitor,name, r, obj)
|
||||
return r
|
||||
end
|
||||
|
||||
function sharedata.new(name, v)
|
||||
skynet.call(service, "lua", "new", name, v)
|
||||
end
|
||||
|
||||
function sharedata.update(name, v)
|
||||
skynet.call(service, "lua", "update", name, v)
|
||||
end
|
||||
|
||||
function sharedata.delete(name)
|
||||
skynet.call(service, "lua", "delete", name)
|
||||
end
|
||||
|
||||
return sharedata
|
||||
134
lualib/sharedata/corelib.lua
Normal file
134
lualib/sharedata/corelib.lua
Normal file
@@ -0,0 +1,134 @@
|
||||
local core = require "sharedata.core"
|
||||
local type = type
|
||||
local next = next
|
||||
local rawget = rawget
|
||||
|
||||
local conf = {}
|
||||
|
||||
conf.host = {
|
||||
new = core.new,
|
||||
delete = core.delete,
|
||||
getref = core.getref,
|
||||
markdirty = core.markdirty,
|
||||
incref = core.incref,
|
||||
decref = core.decref,
|
||||
}
|
||||
|
||||
local meta = {}
|
||||
|
||||
local isdirty = core.isdirty
|
||||
local index = core.index
|
||||
local needupdate = core.needupdate
|
||||
local len = core.len
|
||||
|
||||
local function findroot(self)
|
||||
while self.__parent do
|
||||
self = self.__parent
|
||||
end
|
||||
return self
|
||||
end
|
||||
|
||||
local function update(root, cobj, gcobj)
|
||||
root.__obj = cobj
|
||||
root.__gcobj = gcobj
|
||||
-- don't use pairs
|
||||
for k,v in next, root do
|
||||
if type(v)=="table" and k~="__parent" then
|
||||
local pointer = index(cobj, k)
|
||||
if type(pointer) == "userdata" then
|
||||
update(v, pointer, gcobj)
|
||||
else
|
||||
root[k] = nil
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
local function genkey(self)
|
||||
local key = tostring(self.__key)
|
||||
while self.__parent do
|
||||
self = self.__parent
|
||||
key = self.__key .. "." .. key
|
||||
end
|
||||
return key
|
||||
end
|
||||
|
||||
function meta:__index(key)
|
||||
local obj = self.__obj
|
||||
if isdirty(obj) then
|
||||
local newobj, newtbl = needupdate(self.__gcobj)
|
||||
if newobj then
|
||||
local newgcobj = newtbl.__gcobj
|
||||
-- todo: update
|
||||
local root = findroot(self)
|
||||
update(root, newobj, newgcobj)
|
||||
if obj == self.__obj then
|
||||
error ("The key [" .. genkey(self) .. "] doesn't exist after update")
|
||||
end
|
||||
end
|
||||
end
|
||||
local v = index(self.__obj, key)
|
||||
if type(v) == "userdata" then
|
||||
local r = setmetatable({
|
||||
__obj = v,
|
||||
__gcobj = self.__gcobj,
|
||||
__parent = self,
|
||||
__key = key,
|
||||
}, meta)
|
||||
self[key] = r
|
||||
return r
|
||||
else
|
||||
return v
|
||||
end
|
||||
end
|
||||
|
||||
function meta:__len()
|
||||
return len(self.__obj)
|
||||
end
|
||||
|
||||
local function conf_ipairs(self, index)
|
||||
local obj = self.__obj
|
||||
index = index + 1
|
||||
local value = rawget(self, index)
|
||||
if value then
|
||||
return index, value
|
||||
end
|
||||
local sz = len(obj)
|
||||
if sz < index then
|
||||
return
|
||||
end
|
||||
return index, self[index]
|
||||
end
|
||||
|
||||
function meta:__ipairs()
|
||||
return conf_ipairs, self, 0
|
||||
end
|
||||
|
||||
function meta:__pairs()
|
||||
return conf.next, self, nil
|
||||
end
|
||||
|
||||
function conf.next(obj, key)
|
||||
local nextkey = core.nextkey(obj.__obj, key)
|
||||
if nextkey then
|
||||
return nextkey, obj[nextkey]
|
||||
end
|
||||
end
|
||||
|
||||
function conf.box(obj)
|
||||
local gcobj = core.box(obj)
|
||||
return setmetatable({
|
||||
__parent = false,
|
||||
__obj = obj,
|
||||
__gcobj = gcobj,
|
||||
__key = "",
|
||||
} , meta)
|
||||
end
|
||||
|
||||
function conf.update(self, pointer)
|
||||
local cobj = self.__obj
|
||||
assert(isdirty(cobj), "Obly dirty object can be update")
|
||||
core.update(self.__gcobj, pointer, { __gcobj = core.box(pointer) })
|
||||
end
|
||||
|
||||
return conf
|
||||
@@ -1,4 +1,4 @@
|
||||
local c = require "skynet.c"
|
||||
local c = require "skynet.core"
|
||||
local tostring = tostring
|
||||
local tonumber = tonumber
|
||||
local coroutine = coroutine
|
||||
@@ -43,12 +43,14 @@ end
|
||||
local session_id_coroutine = {}
|
||||
local session_coroutine_id = {}
|
||||
local session_coroutine_address = {}
|
||||
local session_response = {}
|
||||
|
||||
local wakeup_session = {}
|
||||
local sleep_session = {}
|
||||
|
||||
local watching_service = {}
|
||||
local watching_session = {}
|
||||
local dead_service = {}
|
||||
local error_queue = {}
|
||||
|
||||
-- suspend is function
|
||||
@@ -73,7 +75,9 @@ local function _error_dispatch(error_session, error_source)
|
||||
if error_session == 0 then
|
||||
-- service is down
|
||||
-- Don't remove from watching_service , because user may call dead service
|
||||
watching_service[error_source] = false
|
||||
if watching_service[error_source] then
|
||||
dead_service[error_source] = true
|
||||
end
|
||||
for session, srv in pairs(watching_session) do
|
||||
if srv == error_source then
|
||||
table.insert(error_queue, session)
|
||||
@@ -95,6 +99,7 @@ local coroutine_yield = coroutine.yield
|
||||
local function co_create(f)
|
||||
local co = table.remove(coroutine_pool)
|
||||
if co == nil then
|
||||
local print = print
|
||||
co = coroutine.create(function(...)
|
||||
f(...)
|
||||
while true do
|
||||
@@ -122,6 +127,18 @@ local function dispatch_wakeup()
|
||||
end
|
||||
end
|
||||
|
||||
local function release_watching(address)
|
||||
local ref = watching_service[address]
|
||||
if ref then
|
||||
ref = ref - 1
|
||||
if ref > 0 then
|
||||
watching_service[address] = ref
|
||||
else
|
||||
watching_service[address] = nil
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
-- suspend is local function
|
||||
function suspend(co, result, command, param, size)
|
||||
if not result then
|
||||
@@ -142,15 +159,67 @@ function suspend(co, result, command, param, size)
|
||||
elseif command == "RETURN" then
|
||||
local co_session = session_coroutine_id[co]
|
||||
local co_address = session_coroutine_address[co]
|
||||
if param == nil then
|
||||
if param == nil or session_response[co] then
|
||||
error(debug.traceback(co))
|
||||
end
|
||||
c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size)
|
||||
return suspend(co, coroutine.resume(co))
|
||||
session_response[co] = true
|
||||
local ret
|
||||
if not dead_service[co_address] then
|
||||
ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size) >= 0
|
||||
elseif size == nil then
|
||||
c.trash(param, size)
|
||||
ret = false
|
||||
end
|
||||
return suspend(co, coroutine.resume(co, ret))
|
||||
elseif command == "RESPONSE" then
|
||||
local co_session = session_coroutine_id[co]
|
||||
local co_address = session_coroutine_address[co]
|
||||
if session_response[co] then
|
||||
error(debug.traceback(co))
|
||||
end
|
||||
local f = param
|
||||
local function response(ok, ...)
|
||||
if ok == "TEST" then
|
||||
if dead_service[co_address] then
|
||||
release_watching(co_address)
|
||||
f = false
|
||||
return false
|
||||
else
|
||||
return true
|
||||
end
|
||||
end
|
||||
if not f then
|
||||
if f == false then
|
||||
f = nil
|
||||
return false
|
||||
end
|
||||
error "Can't response more than once"
|
||||
end
|
||||
|
||||
local ret
|
||||
if not dead_service[co_address] then
|
||||
if ok then
|
||||
ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, f(...)) >=0
|
||||
else
|
||||
ret = c.send(co_address, skynet.PTYPE_ERROR, co_session, "") >=0
|
||||
end
|
||||
else
|
||||
ret = false
|
||||
end
|
||||
release_watching(co_address)
|
||||
f = nil
|
||||
return ret
|
||||
end
|
||||
watching_service[co_address] = watching_service[co_address] + 1
|
||||
session_response[co] = response
|
||||
return suspend(co, coroutine.resume(co, response))
|
||||
elseif command == "EXIT" then
|
||||
-- coroutine exit
|
||||
local address = session_coroutine_address[co]
|
||||
release_watching(address)
|
||||
session_coroutine_id[co] = nil
|
||||
session_coroutine_address[co] = nil
|
||||
session_response[co] = nil
|
||||
elseif command == "QUIT" then
|
||||
-- service exit
|
||||
return
|
||||
@@ -259,13 +328,21 @@ end
|
||||
|
||||
function skynet.exit()
|
||||
skynet.send(".launcher","lua","REMOVE",skynet.self())
|
||||
-- report the sources that call me
|
||||
for co, session in pairs(session_coroutine_id) do
|
||||
local address = session_coroutine_address[co]
|
||||
local self = skynet.self()
|
||||
if session~=0 and address then
|
||||
skynet.redirect(address, self, "error", session, "")
|
||||
c.redirect(address, 0, skynet.PTYPE_ERROR, session, "")
|
||||
end
|
||||
end
|
||||
-- report the sources I call but haven't return
|
||||
local tmp = {}
|
||||
for session, address in pairs(watching_session) do
|
||||
tmp[address] = true
|
||||
end
|
||||
for address in pairs(tmp) do
|
||||
c.redirect(address, 0, skynet.PTYPE_ERROR, 0, "")
|
||||
end
|
||||
c.command("EXIT")
|
||||
-- quit service
|
||||
coroutine_yield "QUIT"
|
||||
@@ -294,9 +371,6 @@ end
|
||||
|
||||
function skynet.send(addr, typename, ...)
|
||||
local p = proto[typename]
|
||||
if watching_service[addr] == false then
|
||||
error("Service is dead")
|
||||
end
|
||||
return c.send(addr, p.id, 0 , p.pack(...))
|
||||
end
|
||||
|
||||
@@ -321,9 +395,6 @@ end
|
||||
|
||||
function skynet.call(addr, typename, ...)
|
||||
local p = proto[typename]
|
||||
if watching_service[addr] == false then
|
||||
error("Service is dead")
|
||||
end
|
||||
local session = c.send(addr, p.id , nil , p.pack(...))
|
||||
if session == nil then
|
||||
error("call to invalid address " .. skynet.address(addr))
|
||||
@@ -333,16 +404,18 @@ end
|
||||
|
||||
function skynet.rawcall(addr, typename, msg, sz)
|
||||
local p = proto[typename]
|
||||
if watching_service[addr] == false then
|
||||
error("Service is dead")
|
||||
end
|
||||
local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address")
|
||||
return yield_call(addr, session)
|
||||
end
|
||||
|
||||
function skynet.ret(msg, sz)
|
||||
msg = msg or ""
|
||||
coroutine_yield("RETURN", msg, sz)
|
||||
return coroutine_yield("RETURN", msg, sz)
|
||||
end
|
||||
|
||||
function skynet.response(pack)
|
||||
pack = pack or skynet.pack
|
||||
return coroutine_yield("RESPONSE", pack)
|
||||
end
|
||||
|
||||
function skynet.retpack(...)
|
||||
@@ -412,6 +485,12 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
|
||||
local p = assert(proto[prototype], prototype)
|
||||
local f = p.dispatch
|
||||
if f then
|
||||
local ref = watching_service[source]
|
||||
if ref then
|
||||
watching_service[source] = ref + 1
|
||||
else
|
||||
watching_service[source] = 1
|
||||
end
|
||||
local co = co_create(f)
|
||||
session_coroutine_id[co] = session
|
||||
session_coroutine_address[co] = source
|
||||
@@ -470,7 +549,7 @@ end
|
||||
|
||||
function skynet.address(addr)
|
||||
if type(addr) == "number" then
|
||||
return string.format(":%x",addr)
|
||||
return string.format(":%08x",addr)
|
||||
else
|
||||
return tostring(addr)
|
||||
end
|
||||
@@ -570,6 +649,21 @@ function skynet.filter(f ,start_func)
|
||||
end)
|
||||
end
|
||||
|
||||
function skynet.forward_type(map, start_func)
|
||||
c.callback(function(ptype, msg, sz, ...)
|
||||
local prototype = map[ptype]
|
||||
if prototype then
|
||||
dispatch_message(prototype, msg, sz, ...)
|
||||
else
|
||||
dispatch_message(ptype, msg, sz, ...)
|
||||
c.trash(msg, sz)
|
||||
end
|
||||
end, true)
|
||||
skynet.timeout(0, function()
|
||||
init_service(start_func)
|
||||
end)
|
||||
end
|
||||
|
||||
function skynet.endless()
|
||||
return c.command("ENDLESS")~=nil
|
||||
end
|
||||
@@ -607,6 +701,6 @@ end
|
||||
|
||||
-- Inject internal debug framework
|
||||
local debug = require "skynet.debug"
|
||||
debug(skynet)
|
||||
debug(skynet, dispatch_message)
|
||||
|
||||
return skynet
|
||||
|
||||
@@ -1,4 +1,8 @@
|
||||
return function (skynet)
|
||||
local io = io
|
||||
local table = table
|
||||
local debug = debug
|
||||
|
||||
return function (skynet, dispatch_func)
|
||||
|
||||
local internal_info_func
|
||||
|
||||
@@ -39,6 +43,17 @@ function dbgcmd.INFO()
|
||||
end
|
||||
end
|
||||
|
||||
function dbgcmd.EXIT()
|
||||
skynet.exit()
|
||||
end
|
||||
|
||||
function dbgcmd.RUN(source, filename)
|
||||
local inject = require "skynet.inject"
|
||||
local output = inject(source, filename , dispatch_func, skynet.register_protocol)
|
||||
collectgarbage "collect"
|
||||
skynet.ret(skynet.pack(table.concat(output, "\n")))
|
||||
end
|
||||
|
||||
local function _debug_dispatch(session, address, cmd, ...)
|
||||
local f = dbgcmd[cmd]
|
||||
assert(f, cmd)
|
||||
@@ -53,4 +68,4 @@ skynet.register_protocol {
|
||||
dispatch = _debug_dispatch,
|
||||
}
|
||||
|
||||
end
|
||||
end
|
||||
|
||||
@@ -4,15 +4,19 @@ local harbor = {}
|
||||
|
||||
function harbor.globalname(name, handle)
|
||||
handle = handle or skynet.self()
|
||||
skynet.send(".slave", "lua", "REGISTER", name, handle)
|
||||
skynet.send(".cslave", "lua", "REGISTER", name, handle)
|
||||
end
|
||||
|
||||
function harbor.queryname(name)
|
||||
return skynet.call(".cslave", "lua", "QUERYNAME", name)
|
||||
end
|
||||
|
||||
function harbor.link(id)
|
||||
skynet.call(".slave", "lua", "LINK", id)
|
||||
skynet.call(".cslave", "lua", "LINK", id)
|
||||
end
|
||||
|
||||
function harbor.connect(id)
|
||||
skynet.call(".slave", "lua", "CONNECT", id)
|
||||
skynet.call(".cslave", "lua", "CONNECT", id)
|
||||
end
|
||||
|
||||
return harbor
|
||||
|
||||
65
lualib/skynet/inject.lua
Normal file
65
lualib/skynet/inject.lua
Normal file
@@ -0,0 +1,65 @@
|
||||
local function getupvaluetable(u, func, unique)
|
||||
i = 1
|
||||
while true do
|
||||
local name, value = debug.getupvalue(func, i)
|
||||
if name == nil then
|
||||
return
|
||||
end
|
||||
local t = type(value)
|
||||
if t == "table" then
|
||||
u[name] = value
|
||||
elseif t == "function" then
|
||||
if not unique[value] then
|
||||
unique[value] = true
|
||||
getupvaluetable(u, value, unique)
|
||||
end
|
||||
end
|
||||
i=i+1
|
||||
end
|
||||
end
|
||||
|
||||
return function(source, filename , ...)
|
||||
if filename then
|
||||
filename = "@" .. filename
|
||||
else
|
||||
filename = "=(load)"
|
||||
end
|
||||
local output = {}
|
||||
|
||||
local function print(...)
|
||||
local value = { ... }
|
||||
for k,v in ipairs(value) do
|
||||
value[k] = tostring(v)
|
||||
end
|
||||
table.insert(output, table.concat(value, "\t"))
|
||||
end
|
||||
local u = {}
|
||||
local unique = {}
|
||||
local funcs = { ... }
|
||||
for k, func in ipairs(funcs) do
|
||||
getupvaluetable(u, func, unique)
|
||||
end
|
||||
local p = {}
|
||||
local proto = u.proto
|
||||
if proto then
|
||||
for k,v in pairs(proto) do
|
||||
local name, dispatch = v.name, v.dispatch
|
||||
if name and dispatch then
|
||||
local pp = {}
|
||||
p[name] = pp
|
||||
getupvaluetable(pp, dispatch, unique)
|
||||
end
|
||||
end
|
||||
end
|
||||
local env = setmetatable( { print = print , _U = u, _P = p}, { __index = _ENV })
|
||||
local func, err = load(source, filename, "bt", env)
|
||||
if not func then
|
||||
return { err }
|
||||
end
|
||||
local ok, err = xpcall(func, debug.traceback)
|
||||
if not ok then
|
||||
table.insert(output, err)
|
||||
end
|
||||
|
||||
return output
|
||||
end
|
||||
@@ -3,7 +3,21 @@ local io = io
|
||||
|
||||
local hotfix = {}
|
||||
|
||||
local function collect_uv(f , uv)
|
||||
local function envid(f)
|
||||
local i = 1
|
||||
while true do
|
||||
local name, value = debug.getupvalue(f, i)
|
||||
if name == nil then
|
||||
return
|
||||
end
|
||||
if name == "_ENV" then
|
||||
return debug.upvalueid(f, i)
|
||||
end
|
||||
i = i + 1
|
||||
end
|
||||
end
|
||||
|
||||
local function collect_uv(f , uv, env)
|
||||
local i = 1
|
||||
while true do
|
||||
local name, value = debug.getupvalue(f, i)
|
||||
@@ -18,7 +32,9 @@ local function collect_uv(f , uv)
|
||||
uv[name] = { func = f, index = i, id = id }
|
||||
|
||||
if type(value) == "function" then
|
||||
collect_uv(value, uv)
|
||||
if envid(value) == env then
|
||||
collect_uv(value, uv, env)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
@@ -30,7 +46,7 @@ local function collect_all_uv(funcs)
|
||||
local global = {}
|
||||
for _, v in pairs(funcs) do
|
||||
if v[4] then
|
||||
collect_uv(v[4], global)
|
||||
collect_uv(v[4], global, envid(v[4]))
|
||||
end
|
||||
end
|
||||
|
||||
|
||||
@@ -61,6 +61,8 @@ return function (name , G, loader)
|
||||
|
||||
setmetatable(G, { __index = env , __newindex = init_system })
|
||||
|
||||
local pattern
|
||||
|
||||
do
|
||||
local path = skynet.getenv "snax"
|
||||
|
||||
@@ -70,6 +72,7 @@ return function (name , G, loader)
|
||||
local filename = string.gsub(pat, "?", name)
|
||||
local f , err = loader(filename, "bt", G)
|
||||
if f then
|
||||
pattern = pat
|
||||
mainfunc = f
|
||||
break
|
||||
else
|
||||
@@ -90,5 +93,5 @@ return function (name , G, loader)
|
||||
G[k] = v
|
||||
end
|
||||
|
||||
return func
|
||||
return func, pattern
|
||||
end
|
||||
|
||||
@@ -143,16 +143,11 @@ local function launch_master(conf)
|
||||
local balance = 1
|
||||
|
||||
skynet.dispatch("lua", function(_,source,command, ...)
|
||||
if command == "register_slave" then
|
||||
table.insert(slave, source)
|
||||
skynet.ret(skynet.pack(nil))
|
||||
else
|
||||
skynet.ret(skynet.pack(conf.command_handler(command, ...)))
|
||||
end
|
||||
skynet.ret(skynet.pack(conf.command_handler(command, ...)))
|
||||
end)
|
||||
|
||||
for i=1,instance do
|
||||
skynet.newservice(SERVICE_NAME)
|
||||
table.insert(slave, skynet.newservice(SERVICE_NAME))
|
||||
end
|
||||
|
||||
skynet.error(string.format("login server listen at : %s %d", host, port))
|
||||
@@ -178,7 +173,6 @@ local function login(conf)
|
||||
skynet.start(function()
|
||||
local loginmaster = skynet.localname(name)
|
||||
if loginmaster then
|
||||
skynet.call(loginmaster, "lua", "register_slave")
|
||||
local auth_handler = assert(conf.auth_handler)
|
||||
launch_master = nil
|
||||
conf = nil
|
||||
|
||||
@@ -218,9 +218,9 @@ local function try_connect(self , once)
|
||||
if not once then
|
||||
skynet.error("socket: connect to", self.__host, self.__port)
|
||||
end
|
||||
return
|
||||
return true
|
||||
elseif once then
|
||||
error(string.format("Connect to %s:%d failed", self.__host, self.__port))
|
||||
return false
|
||||
end
|
||||
if t > 1000 then
|
||||
skynet.error("socket: try to reconnect", self.__host, self.__port)
|
||||
@@ -233,7 +233,7 @@ local function try_connect(self , once)
|
||||
end
|
||||
end
|
||||
|
||||
local function block_connect(self, once)
|
||||
local function check_connection(self)
|
||||
if self.__sock then
|
||||
local authco = self.__authcoroutine
|
||||
if not authco then
|
||||
@@ -247,26 +247,36 @@ local function block_connect(self, once)
|
||||
if self.__closed then
|
||||
return false
|
||||
end
|
||||
end
|
||||
|
||||
local function block_connect(self, once)
|
||||
local r = check_connection(self)
|
||||
if r ~= nil then
|
||||
return r
|
||||
end
|
||||
|
||||
if #self.__connecting > 0 then
|
||||
-- connecting in other coroutine
|
||||
local co = coroutine.running()
|
||||
table.insert(self.__connecting, co)
|
||||
skynet.wait()
|
||||
-- check connection again
|
||||
return block_connect(self, once)
|
||||
end
|
||||
self.__connecting[1] = true
|
||||
try_connect(self, once)
|
||||
self.__connecting[1] = nil
|
||||
for i=2, #self.__connecting do
|
||||
local co = self.__connecting[i]
|
||||
self.__connecting[i] = nil
|
||||
skynet.wakeup(co)
|
||||
else
|
||||
self.__connecting[1] = true
|
||||
try_connect(self, once)
|
||||
self.__connecting[1] = nil
|
||||
for i=2, #self.__connecting do
|
||||
local co = self.__connecting[i]
|
||||
self.__connecting[i] = nil
|
||||
skynet.wakeup(co)
|
||||
end
|
||||
end
|
||||
|
||||
-- check again
|
||||
return block_connect(self, once)
|
||||
r = check_connection(self)
|
||||
if r == nil then
|
||||
error(string.format("Connect to %s:%d failed", self.__host, self.__port))
|
||||
else
|
||||
return r
|
||||
end
|
||||
end
|
||||
|
||||
function channel:connect(once)
|
||||
@@ -296,7 +306,7 @@ local function wait_for_response(self, response)
|
||||
end
|
||||
|
||||
function channel:request(request, response)
|
||||
assert(block_connect(self))
|
||||
assert(block_connect(self, true)) -- connect once
|
||||
|
||||
if not socket.write(self.__sock[1], request) then
|
||||
close_channel_socket(self)
|
||||
|
||||
@@ -55,7 +55,7 @@ traceback (lua_State *L) {
|
||||
static void
|
||||
_report_launcher_error(struct skynet_context *ctx) {
|
||||
// sizeof "ERROR" == 5
|
||||
skynet_sendname(ctx, ".launcher", PTYPE_TEXT, 0, "ERROR", 5);
|
||||
skynet_sendname(ctx, 0, ".launcher", PTYPE_TEXT, 0, "ERROR", 5);
|
||||
}
|
||||
|
||||
static const char *
|
||||
|
||||
@@ -13,31 +13,31 @@ skynet.start(function()
|
||||
standalone = true
|
||||
skynet.setenv("standalone", "true")
|
||||
|
||||
local slave = skynet.newservice "cdummy"
|
||||
if slave == nil then
|
||||
local ok, slave = pcall(skynet.newservice, "cdummy")
|
||||
if not ok then
|
||||
skynet.abort()
|
||||
end
|
||||
skynet.name(".slave", slave)
|
||||
skynet.name(".cslave", slave)
|
||||
|
||||
else
|
||||
if standalone then
|
||||
if not skynet.newservice "cmaster" then
|
||||
if not pcall(skynet.newservice,"cmaster") then
|
||||
skynet.abort()
|
||||
end
|
||||
end
|
||||
|
||||
local slave = skynet.newservice "cslave"
|
||||
if slave == nil then
|
||||
local ok, slave = pcall(skynet.newservice, "cslave")
|
||||
if not ok then
|
||||
skynet.abort()
|
||||
end
|
||||
skynet.name(".slave", slave)
|
||||
skynet.name(".cslave", slave)
|
||||
end
|
||||
|
||||
if standalone then
|
||||
local datacenter = assert(skynet.newservice "datacenterd")
|
||||
local datacenter = skynet.newservice "datacenterd"
|
||||
skynet.name("DATACENTER", datacenter)
|
||||
end
|
||||
assert(skynet.newservice "service_mgr")
|
||||
assert(skynet.newservice(skynet.getenv "start" or "main"))
|
||||
skynet.newservice "service_mgr"
|
||||
pcall(skynet.newservice,skynet.getenv "start" or "main")
|
||||
skynet.exit()
|
||||
end)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
local skynet = require "skynet"
|
||||
|
||||
local globalname = {}
|
||||
local queryname = {}
|
||||
local harbor = {}
|
||||
|
||||
skynet.register_protocol {
|
||||
@@ -17,12 +18,43 @@ skynet.register_protocol {
|
||||
unpack = skynet.tostring,
|
||||
}
|
||||
|
||||
local function response_name(name)
|
||||
local address = globalname[name]
|
||||
if queryname[name] then
|
||||
local tmp = queryname[name]
|
||||
queryname[name] = nil
|
||||
for _,resp in ipairs(tmp) do
|
||||
resp(true, address)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
function harbor.REGISTER(name, handle)
|
||||
assert(globalname[name] == nil)
|
||||
globalname[name] = handle
|
||||
response_name(name)
|
||||
skynet.redirect(harbor_service, handle, "harbor", 0, "N " .. name)
|
||||
end
|
||||
|
||||
function harbor.QUERYNAME(name)
|
||||
if name:byte() == 46 then -- "." , local name
|
||||
skynet.ret(skynet.pack(skynet.localname(name)))
|
||||
return
|
||||
end
|
||||
local result = globalname[name]
|
||||
if result then
|
||||
skynet.ret(skynet.pack(result))
|
||||
return
|
||||
end
|
||||
local queue = queryname[name]
|
||||
if queue == nil then
|
||||
queue = { skynet.response() }
|
||||
queryname[name] = queue
|
||||
else
|
||||
table.insert(queue, skynet.response())
|
||||
end
|
||||
end
|
||||
|
||||
function harbor.LINK(id)
|
||||
skynet.ret()
|
||||
end
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
local skynet = require "skynet"
|
||||
local sc = require "socketchannel"
|
||||
local socket = require "socket"
|
||||
local cluster = require "cluster.c"
|
||||
local cluster = require "cluster.core"
|
||||
|
||||
local config_name = skynet.getenv "cluster"
|
||||
local node_address = {}
|
||||
@@ -51,13 +51,34 @@ function command.listen(source, addr, port)
|
||||
skynet.ret(skynet.pack(nil))
|
||||
end
|
||||
|
||||
function command.req(source, node, addr, msg, sz)
|
||||
local function send_request(source, node, addr, msg, sz)
|
||||
local request
|
||||
local c = node_channel[node]
|
||||
local session = node_session[node]
|
||||
-- msg is a local pointer, cluster.packrequest will free it
|
||||
request, node_session[node] = cluster.packrequest(addr, session , msg, sz)
|
||||
skynet.ret(c:request(request, session))
|
||||
|
||||
return c:request(request, session)
|
||||
end
|
||||
|
||||
function command.req(...)
|
||||
local ok, msg, sz = pcall(send_request, ...)
|
||||
if ok then
|
||||
skynet.ret(msg, sz)
|
||||
else
|
||||
skynet.error(msg)
|
||||
skynet.response()(false)
|
||||
end
|
||||
end
|
||||
|
||||
local proxy = {}
|
||||
|
||||
function command.proxy(source, node, name)
|
||||
local fullname = node .. "." .. name
|
||||
if proxy[fullname] == nil then
|
||||
proxy[fullname] = skynet.newservice("clusterproxy", node, name)
|
||||
end
|
||||
skynet.ret(skynet.pack(proxy[fullname]))
|
||||
end
|
||||
|
||||
local request_fd = {}
|
||||
|
||||
27
service/clusterproxy.lua
Normal file
27
service/clusterproxy.lua
Normal file
@@ -0,0 +1,27 @@
|
||||
local skynet = require "skynet"
|
||||
local cluster = require "cluster"
|
||||
|
||||
local node, address = ...
|
||||
|
||||
skynet.register_protocol {
|
||||
name = "system",
|
||||
id = skynet.PTYPE_SYSTEM,
|
||||
unpack = function (...) return ... end,
|
||||
}
|
||||
|
||||
local forward_map = {
|
||||
[skynet.PTYPE_LUA] = skynet.PTYPE_SYSTEM,
|
||||
[skynet.PTYPE_RESPONSE] = skynet.PTYPE_RESPONSE, -- don't free response message
|
||||
}
|
||||
|
||||
skynet.forward_type( forward_map ,function()
|
||||
local clusterd = skynet.uniqueservice("clusterd")
|
||||
local n = tonumber(address)
|
||||
if n then
|
||||
address = n
|
||||
end
|
||||
skynet.dispatch("system", function (session, source, msg, sz)
|
||||
local m,s = skynet.rawcall(clusterd, "lua", skynet.pack("req", node, address, msg, sz))
|
||||
skynet.ret(skynet.rawcall(clusterd, "lua", skynet.pack("req", node, address, msg, sz)))
|
||||
end)
|
||||
end)
|
||||
@@ -7,10 +7,7 @@ local function console_main_loop()
|
||||
while true do
|
||||
local cmdline = socket.readline(stdin, "\n")
|
||||
if cmdline ~= "" then
|
||||
local handle = skynet.newservice(cmdline)
|
||||
if handle == nil then
|
||||
print("Launch error:",cmdline)
|
||||
end
|
||||
pcall(skynet.newservice,cmdline)
|
||||
end
|
||||
end
|
||||
socket.unlock(stdin)
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
local skynet = require "skynet"
|
||||
local socket = require "socket"
|
||||
local table = table
|
||||
|
||||
local slaves = {}
|
||||
local connect_queue = {}
|
||||
local globalname = {}
|
||||
local queryname = {}
|
||||
local harbor = {}
|
||||
local harbor_service
|
||||
local monitor = {}
|
||||
@@ -28,7 +30,7 @@ local function monitor_clear(id)
|
||||
if v then
|
||||
monitor[id] = nil
|
||||
for _, v in ipairs(v) do
|
||||
skynet.redirect(v.address, 0, "response", v.session, "")
|
||||
v(true)
|
||||
end
|
||||
end
|
||||
end
|
||||
@@ -60,6 +62,17 @@ local function ready()
|
||||
end
|
||||
end
|
||||
|
||||
local function response_name(name)
|
||||
local address = globalname[name]
|
||||
if queryname[name] then
|
||||
local tmp = queryname[name]
|
||||
queryname[name] = nil
|
||||
for _,resp in ipairs(tmp) do
|
||||
resp(true, address)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
local function monitor_master(master_fd)
|
||||
while true do
|
||||
local ok, t, id_name, address = pcall(read_package,master_fd)
|
||||
@@ -72,6 +85,7 @@ local function monitor_master(master_fd)
|
||||
end
|
||||
elseif t == 'N' then
|
||||
globalname[id_name] = address
|
||||
response_name(id_name)
|
||||
if connect_queue == nil then
|
||||
skynet.redirect(harbor_service, address, "harbor", 0, "N " .. id_name)
|
||||
end
|
||||
@@ -79,6 +93,7 @@ local function monitor_master(master_fd)
|
||||
local fd = slaves[id_name]
|
||||
slaves[id_name] = false
|
||||
if fd then
|
||||
monitor_clear(id_name)
|
||||
socket.close(fd)
|
||||
end
|
||||
end
|
||||
@@ -129,14 +144,14 @@ local function monitor_harbor(master_fd)
|
||||
return function(session, source, command)
|
||||
local t = string.sub(command, 1, 1)
|
||||
local arg = string.sub(command, 3)
|
||||
if t == "Q" then
|
||||
if t == 'Q' then
|
||||
-- query name
|
||||
if globalname[arg] then
|
||||
skynet.redirect(harbor_service, globalname[arg], "harbor", 0, "N " .. arg)
|
||||
else
|
||||
socket.write(master_fd, pack_package("Q", arg))
|
||||
end
|
||||
elseif t == "D" then
|
||||
elseif t == 'D' then
|
||||
-- harbor down
|
||||
local id = tonumber(arg)
|
||||
if slaves[id] then
|
||||
@@ -149,46 +164,66 @@ local function monitor_harbor(master_fd)
|
||||
end
|
||||
end
|
||||
|
||||
function harbor.REGISTER(_,_, fd, name, handle)
|
||||
function harbor.REGISTER(fd, name, handle)
|
||||
assert(globalname[name] == nil)
|
||||
globalname[name] = handle
|
||||
response_name(name)
|
||||
socket.write(fd, pack_package("R", name, handle))
|
||||
skynet.redirect(harbor_service, handle, "harbor", 0, "N " .. name)
|
||||
end
|
||||
|
||||
function harbor.LINK(session, source, fd, id)
|
||||
function harbor.LINK(fd, id)
|
||||
if slaves[id] then
|
||||
if monitor[id] == nil then
|
||||
monitor[id] = {}
|
||||
end
|
||||
table.insert(monitor[id], { address = source, session = session })
|
||||
table.insert(monitor[id], skynet.response())
|
||||
else
|
||||
skynet.ret()
|
||||
end
|
||||
end
|
||||
|
||||
function harbor.CONNECT(session, source, fd, id)
|
||||
function harbor.CONNECT(fd, id)
|
||||
if not slaves[id] then
|
||||
if monitor[id] == nil then
|
||||
monitor[id] = {}
|
||||
end
|
||||
table.insert(monitor[id], { address = source, session = session })
|
||||
table.insert(monitor[id], skynet.response())
|
||||
else
|
||||
skynet.ret()
|
||||
end
|
||||
end
|
||||
|
||||
function harbor.QUERYNAME(fd, name)
|
||||
if name:byte() == 46 then -- "." , local name
|
||||
skynet.ret(skynet.pack(skynet.localname(name)))
|
||||
return
|
||||
end
|
||||
local result = globalname[name]
|
||||
if result then
|
||||
skynet.ret(skynet.pack(result))
|
||||
return
|
||||
end
|
||||
local queue = queryname[name]
|
||||
if queue == nil then
|
||||
queue = { skynet.response() }
|
||||
queryname[name] = queue
|
||||
else
|
||||
table.insert(queue, skynet.response())
|
||||
end
|
||||
end
|
||||
|
||||
skynet.start(function()
|
||||
local master_addr = skynet.getenv "master"
|
||||
local harbor_id = tonumber(skynet.getenv "harbor")
|
||||
local slave_address = assert(skynet.getenv "address")
|
||||
local slave_fd = socket.listen(slave_address)
|
||||
skynet.error("slave connect to master " .. tostring(master_addr))
|
||||
local master_fd = socket.open(master_addr)
|
||||
local master_fd = assert(socket.open(master_addr), "Can't connect to master")
|
||||
|
||||
skynet.dispatch("lua", function (session,source,command,...)
|
||||
skynet.dispatch("lua", function (_,_,command,...)
|
||||
local f = assert(harbor[command])
|
||||
f(session, source, master_fd, ...)
|
||||
f(master_fd, ...)
|
||||
end)
|
||||
skynet.dispatch("text", monitor_harbor(master_fd))
|
||||
|
||||
|
||||
@@ -45,10 +45,8 @@ local function wakeup(db, key1, key2, value, ...)
|
||||
db[key1] = nil
|
||||
if value then
|
||||
-- throw error because can't wake up a branch
|
||||
for _,v in ipairs(q) do
|
||||
local session = v[1]
|
||||
local source = v[2]
|
||||
skynet.redirect(source, 0, "error", session, "")
|
||||
for _,response in ipairs(q) do
|
||||
response(false)
|
||||
end
|
||||
else
|
||||
return q
|
||||
@@ -66,15 +64,13 @@ function command.UPDATE(...)
|
||||
end
|
||||
local q = wakeup(wait_queue, ...)
|
||||
if q then
|
||||
for _, v in ipairs(q) do
|
||||
local session = v[1]
|
||||
local source = v[2]
|
||||
skynet.redirect(source, 0, "response", session, skynet.pack(value))
|
||||
for _, response in ipairs(q) do
|
||||
response(true,value)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
local function waitfor(session, source, db, key1, key2, ...)
|
||||
local function waitfor(db, key1, key2, ...)
|
||||
if key2 == nil then
|
||||
-- push queue
|
||||
local q = db[key1]
|
||||
@@ -84,7 +80,7 @@ local function waitfor(session, source, db, key1, key2, ...)
|
||||
else
|
||||
assert(q[mode] == "queue")
|
||||
end
|
||||
table.insert(q, { session, source })
|
||||
table.insert(q, skynet.response())
|
||||
else
|
||||
local q = db[key1]
|
||||
if q == nil then
|
||||
@@ -93,18 +89,18 @@ local function waitfor(session, source, db, key1, key2, ...)
|
||||
else
|
||||
assert(q[mode] == "branch")
|
||||
end
|
||||
return waitfor(session, source, q, key2, ...)
|
||||
return waitfor(q, key2, ...)
|
||||
end
|
||||
end
|
||||
|
||||
skynet.start(function()
|
||||
skynet.dispatch("lua", function (session, source, cmd, ...)
|
||||
skynet.dispatch("lua", function (_, _, cmd, ...)
|
||||
if cmd == "WAIT" then
|
||||
local ret = command.QUERY(...)
|
||||
if ret then
|
||||
skynet.ret(skynet.pack(ret))
|
||||
else
|
||||
waitfor(session, source, wait_queue, ...)
|
||||
waitfor(wait_queue, ...)
|
||||
end
|
||||
else
|
||||
local f = assert(command[cmd])
|
||||
|
||||
@@ -36,6 +36,7 @@ local function dump_list(print, list)
|
||||
for _,v in ipairs(index) do
|
||||
dump_line(print, v, list[v])
|
||||
end
|
||||
print("OK")
|
||||
end
|
||||
|
||||
local function split_cmdline(cmdline)
|
||||
@@ -92,6 +93,9 @@ skynet.start(function()
|
||||
socket.start(listen_socket , function(id, addr)
|
||||
local function print(...)
|
||||
local t = { ... }
|
||||
for k,v in ipairs(t) do
|
||||
t[k] = tostring(v)
|
||||
end
|
||||
socket.write(id, table.concat(t,"\t"))
|
||||
socket.write(id, "\n")
|
||||
end
|
||||
@@ -106,6 +110,7 @@ function COMMAND.help()
|
||||
list = "List all the service",
|
||||
stat = "Dump all stats",
|
||||
info = "Info address : get service infomation",
|
||||
exit = "exit address : kill a lua service",
|
||||
kill = "kill address : kill service",
|
||||
mem = "mem : show memory status",
|
||||
gc = "gc : force every lua service do garbage collect",
|
||||
@@ -114,6 +119,7 @@ function COMMAND.help()
|
||||
clearcache = "clear lua code cache",
|
||||
service = "List unique service",
|
||||
task = "task address : show service task detail",
|
||||
inject = "inject address luascript.lua",
|
||||
}
|
||||
end
|
||||
|
||||
@@ -122,8 +128,8 @@ function COMMAND.clearcache()
|
||||
end
|
||||
|
||||
function COMMAND.start(...)
|
||||
local addr = skynet.newservice(...)
|
||||
if addr then
|
||||
local ok, addr = pcall(skynet.newservice, ...)
|
||||
if ok then
|
||||
return { [skynet.address(addr)] = ... }
|
||||
else
|
||||
return "Failed"
|
||||
@@ -131,8 +137,8 @@ function COMMAND.start(...)
|
||||
end
|
||||
|
||||
function COMMAND.snax(...)
|
||||
local s = snax.newservice(...)
|
||||
if s then
|
||||
local ok, s = pcall(snax.newservice, ...)
|
||||
if ok then
|
||||
local addr = s.handle
|
||||
return { [skynet.address(addr)] = ... }
|
||||
else
|
||||
@@ -143,3 +149,35 @@ end
|
||||
function COMMAND.service()
|
||||
return skynet.call("SERVICE", "lua", "LIST")
|
||||
end
|
||||
|
||||
local function adjust_address(address)
|
||||
if address:sub(1,1) ~= ":" then
|
||||
address = bit32.replace( tonumber("0x" .. address), skynet.harbor(skynet.self()), 24, 8)
|
||||
end
|
||||
return address
|
||||
end
|
||||
|
||||
function COMMAND.exit(address)
|
||||
skynet.send(adjust_address(address), "debug", "EXIT")
|
||||
end
|
||||
|
||||
function COMMAND.inject(address, filename)
|
||||
address = adjust_address(address)
|
||||
local f = io.open(filename, "rb")
|
||||
if not f then
|
||||
return "Can't open " .. filename
|
||||
end
|
||||
local source = f:read "*a"
|
||||
f:close()
|
||||
return skynet.call(address, "debug", "RUN", source, filename)
|
||||
end
|
||||
|
||||
function COMMAND.task(address)
|
||||
address = adjust_address(address)
|
||||
return skynet.call(address,"debug","TASK")
|
||||
end
|
||||
|
||||
function COMMAND.info(address)
|
||||
address = adjust_address(address)
|
||||
return skynet.call(address,"debug","INFO")
|
||||
end
|
||||
|
||||
@@ -31,7 +31,7 @@ end
|
||||
function handler.connect(fd, addr)
|
||||
local c = {
|
||||
fd = fd,
|
||||
ip = msg,
|
||||
ip = addr,
|
||||
}
|
||||
connection[fd] = c
|
||||
skynet.send(watchdog, "lua", "socket", "open", fd, addr)
|
||||
|
||||
@@ -28,27 +28,7 @@ function command.STAT()
|
||||
return list
|
||||
end
|
||||
|
||||
function command.INFO(_, _, handle)
|
||||
handle = handle_to_address(handle)
|
||||
if services[handle] == nil then
|
||||
return
|
||||
else
|
||||
local result = skynet.call(handle,"debug","INFO")
|
||||
return result
|
||||
end
|
||||
end
|
||||
|
||||
function command.TASK(_, _, handle)
|
||||
handle = handle_to_address(handle)
|
||||
if services[handle] == nil then
|
||||
return
|
||||
else
|
||||
local result = skynet.call(handle,"debug","TASK")
|
||||
return result
|
||||
end
|
||||
end
|
||||
|
||||
function command.KILL(_, _, handle)
|
||||
function command.KILL(_, handle)
|
||||
handle = handle_to_address(handle)
|
||||
skynet.kill(handle)
|
||||
local ret = { [skynet.address(handle)] = tostring(services[handle]) }
|
||||
@@ -72,18 +52,29 @@ function command.GC()
|
||||
return command.MEM()
|
||||
end
|
||||
|
||||
function command.REMOVE(_,_, handle)
|
||||
function command.REMOVE(_, handle)
|
||||
services[handle] = nil
|
||||
local response = instance[handle]
|
||||
if response then
|
||||
-- instance is dead
|
||||
response(false)
|
||||
instance[handle] = nil
|
||||
end
|
||||
|
||||
-- don't return (skynet.ret) because the handle may exit
|
||||
return NORET
|
||||
end
|
||||
|
||||
function command.LAUNCH(address, session, service, ...)
|
||||
local function return_string(str)
|
||||
return str
|
||||
end
|
||||
|
||||
function command.LAUNCH(_, service, ...)
|
||||
local param = table.concat({...}, " ")
|
||||
local inst = skynet.launch(service, param)
|
||||
if inst then
|
||||
services[inst] = service .. " " .. param
|
||||
instance[inst] = { session = session, address = address }
|
||||
instance[inst] = skynet.response(return_string)
|
||||
else
|
||||
skynet.ret("") -- launch failed
|
||||
end
|
||||
@@ -93,9 +84,9 @@ end
|
||||
function command.ERROR(address)
|
||||
-- see serivce-src/service_lua.c
|
||||
-- init failed
|
||||
local reply = instance[address]
|
||||
if reply then
|
||||
skynet.redirect(reply.address , 0, "response", reply.session, "")
|
||||
local response = instance[address]
|
||||
if response then
|
||||
response(false)
|
||||
instance[address] = nil
|
||||
end
|
||||
services[address] = nil
|
||||
@@ -104,9 +95,9 @@ end
|
||||
|
||||
function command.LAUNCHOK(address)
|
||||
-- init notice
|
||||
local reply = instance[address]
|
||||
if reply then
|
||||
skynet.redirect(reply.address , 0, "response", reply.session, skynet.address(address))
|
||||
local response = instance[address]
|
||||
if response then
|
||||
response(true, skynet.address(address))
|
||||
instance[address] = nil
|
||||
end
|
||||
|
||||
@@ -127,7 +118,7 @@ skynet.register_protocol {
|
||||
else
|
||||
-- launch request
|
||||
local service, param = string.match(cmd,"([^ ]+) (.*)")
|
||||
command.LAUNCH(address, session, service, param)
|
||||
command.LAUNCH(_, service, param)
|
||||
end
|
||||
end,
|
||||
}
|
||||
@@ -136,7 +127,7 @@ skynet.dispatch("lua", function(session, address, cmd , ...)
|
||||
cmd = string.upper(cmd)
|
||||
local f = command[cmd]
|
||||
if f then
|
||||
local ret = f(address, session, ...)
|
||||
local ret = f(address, ...)
|
||||
if ret ~= NORET then
|
||||
skynet.ret(skynet.pack(ret))
|
||||
end
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
local skynet = require "skynet"
|
||||
local mc = require "multicast.c"
|
||||
local mc = require "multicast.core"
|
||||
local datacenter = require "datacenter"
|
||||
|
||||
local harbor_id = skynet.harbor(skynet.self())
|
||||
|
||||
@@ -117,7 +117,6 @@ local function register_global()
|
||||
|
||||
function cmd.REPORT(m)
|
||||
mgr[m] = true
|
||||
skynet.watch(m)
|
||||
end
|
||||
|
||||
local function add_list(all, m)
|
||||
|
||||
118
service/sharedatad.lua
Normal file
118
service/sharedatad.lua
Normal file
@@ -0,0 +1,118 @@
|
||||
local skynet = require "skynet"
|
||||
local sharedata = require "sharedata.corelib"
|
||||
local table = table
|
||||
|
||||
local NORET = {}
|
||||
local pool = {}
|
||||
local objmap = {}
|
||||
|
||||
local function newobj(name, tbl)
|
||||
assert(pool[name] == nil)
|
||||
local cobj = sharedata.host.new(tbl)
|
||||
sharedata.host.incref(cobj)
|
||||
local v = { value = tbl , obj = cobj, watch = {} }
|
||||
objmap[cobj] = v
|
||||
pool[name] = v
|
||||
end
|
||||
|
||||
local function collectobj()
|
||||
while true do
|
||||
skynet.sleep(600 * 100) -- sleep 10 min
|
||||
collectgarbage()
|
||||
for obj, v in pairs(objmap) do
|
||||
if v == true then
|
||||
if sharedata.host.getref(obj) <= 0 then
|
||||
objmap[obj] = nil
|
||||
sharedata.host.delete(obj)
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
local CMD = {}
|
||||
|
||||
function CMD.new(name, t)
|
||||
local dt = type(t)
|
||||
local value
|
||||
if dt == "table" then
|
||||
value = t
|
||||
elseif dt == "string" then
|
||||
value = {}
|
||||
local f = load(t, "=" .. name, "t", env)
|
||||
f()
|
||||
elseif dt == "nil" then
|
||||
value = {}
|
||||
else
|
||||
error ("Unknown data type " .. dt)
|
||||
end
|
||||
newobj(name, value)
|
||||
end
|
||||
|
||||
function CMD.delete(name)
|
||||
local v = assert(pool[name])
|
||||
pool[name] = nil
|
||||
assert(objmap[v.obj])
|
||||
objmap[v.obj] = true
|
||||
sharedata.host.decref(v.obj)
|
||||
for _,response in ipairs(v.watch) do
|
||||
response(true)
|
||||
end
|
||||
end
|
||||
|
||||
function CMD.query(name)
|
||||
local v = assert(pool[name])
|
||||
local obj = v.obj
|
||||
sharedata.host.incref(obj)
|
||||
return v.obj
|
||||
end
|
||||
|
||||
function CMD.confirm(cobj)
|
||||
if objmap[cobj] then
|
||||
sharedata.host.decref(cobj)
|
||||
end
|
||||
return NORET
|
||||
end
|
||||
|
||||
function CMD.update(name, t)
|
||||
local v = pool[name]
|
||||
local watch, oldcobj
|
||||
if v then
|
||||
watch = v.watch
|
||||
oldcobj = v.obj
|
||||
objmap[oldcobj] = true
|
||||
sharedata.host.decref(oldcobj)
|
||||
pool[name] = nil
|
||||
end
|
||||
CMD.new(name, t)
|
||||
local newobj = pool[name].obj
|
||||
if watch then
|
||||
sharedata.host.markdirty(oldcobj)
|
||||
for _,response in ipairs(watch) do
|
||||
response(true, newobj)
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
function CMD.monitor(name, obj)
|
||||
local v = assert(pool[name])
|
||||
if obj ~= v.obj then
|
||||
return v.obj
|
||||
end
|
||||
|
||||
table.insert(v.watch, skynet.response())
|
||||
|
||||
return NORET
|
||||
end
|
||||
|
||||
skynet.start(function()
|
||||
skynet.fork(collectobj)
|
||||
skynet.dispatch("lua", function (session, source ,cmd, ...)
|
||||
local f = assert(CMD[cmd])
|
||||
local r = f(...)
|
||||
if r ~= NORET then
|
||||
skynet.ret(skynet.pack(r))
|
||||
end
|
||||
end)
|
||||
end)
|
||||
|
||||
@@ -1,10 +1,17 @@
|
||||
local skynet = require "skynet"
|
||||
local c = require "skynet.c"
|
||||
local c = require "skynet.core"
|
||||
local snax_interface = require "snax.interface"
|
||||
local profile = require "profile"
|
||||
local snax = require "snax"
|
||||
|
||||
local func = snax_interface(tostring(...), _ENV)
|
||||
local snax_name = tostring(...)
|
||||
local func, pattern = snax_interface(snax_name, _ENV)
|
||||
local snax_path = pattern:sub(1,pattern:find("?", 1, true)-1) .. snax_name .. "/"
|
||||
package.path = snax_path .. "?.lua;" .. package.path
|
||||
|
||||
SERVICE_NAME = snax_name
|
||||
SERVICE_PATH = snax_path
|
||||
|
||||
local mode
|
||||
local thread_id
|
||||
local message_queue = {}
|
||||
|
||||
@@ -30,7 +30,7 @@ void skynet_error(struct skynet_context * context, const char *msg, ...);
|
||||
const char * skynet_command(struct skynet_context * context, const char * cmd , const char * parm);
|
||||
uint32_t skynet_queryname(struct skynet_context * context, const char * name);
|
||||
int skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int type, int session, void * msg, size_t sz);
|
||||
int skynet_sendname(struct skynet_context * context, const char * destination , int type, int session, void * msg, size_t sz);
|
||||
int skynet_sendname(struct skynet_context * context, uint32_t source, const char * destination , int type, int session, void * msg, size_t sz);
|
||||
|
||||
int skynet_isremote(struct skynet_context *, uint32_t handle, int * harbor);
|
||||
|
||||
|
||||
@@ -49,7 +49,6 @@ skynet_handle_register(struct skynet_context *ctx) {
|
||||
rwlock_wunlock(&s->lock);
|
||||
|
||||
handle |= s->harbor;
|
||||
skynet_context_init(ctx, handle);
|
||||
return handle;
|
||||
}
|
||||
}
|
||||
@@ -67,8 +66,9 @@ skynet_handle_register(struct skynet_context *ctx) {
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
int
|
||||
skynet_handle_retire(uint32_t handle) {
|
||||
int ret = 0;
|
||||
struct handle_storage *s = H;
|
||||
|
||||
rwlock_wlock(&s->lock);
|
||||
@@ -79,6 +79,7 @@ skynet_handle_retire(uint32_t handle) {
|
||||
if (ctx != NULL && skynet_context_handle(ctx) == handle) {
|
||||
skynet_context_release(ctx);
|
||||
s->slot[hash] = NULL;
|
||||
ret = 1;
|
||||
int i;
|
||||
int j=0, n=s->name_count;
|
||||
for (i=0; i<n; ++i) {
|
||||
@@ -94,6 +95,8 @@ skynet_handle_retire(uint32_t handle) {
|
||||
}
|
||||
|
||||
rwlock_wunlock(&s->lock);
|
||||
|
||||
return ret;
|
||||
}
|
||||
|
||||
void
|
||||
@@ -105,10 +108,14 @@ skynet_handle_retireall() {
|
||||
for (i=0;i<s->slot_size;i++) {
|
||||
rwlock_rlock(&s->lock);
|
||||
struct skynet_context * ctx = s->slot[i];
|
||||
uint32_t handle = 0;
|
||||
if (ctx)
|
||||
handle = skynet_context_handle(ctx);
|
||||
rwlock_runlock(&s->lock);
|
||||
if (ctx != NULL) {
|
||||
++n;
|
||||
skynet_handle_retire(skynet_context_handle(ctx));
|
||||
if (handle != 0) {
|
||||
if (skynet_handle_retire(handle)) {
|
||||
++n;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (n==0)
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
struct skynet_context;
|
||||
|
||||
uint32_t skynet_handle_register(struct skynet_context *);
|
||||
void skynet_handle_retire(uint32_t handle);
|
||||
int skynet_handle_retire(uint32_t handle);
|
||||
struct skynet_context * skynet_handle_grab(uint32_t handle);
|
||||
void skynet_handle_retireall();
|
||||
|
||||
|
||||
@@ -97,9 +97,13 @@ static const char * load_config = "\
|
||||
|
||||
int
|
||||
main(int argc, char *argv[]) {
|
||||
const char * config_file = "config";
|
||||
const char * config_file = NULL ;
|
||||
if (argc > 1) {
|
||||
config_file = argv[1];
|
||||
} else {
|
||||
fprintf(stderr, "Need a config file. Please read skynet wiki : https://github.com/cloudwu/skynet/wiki/Config\n"
|
||||
"usage: skynet configfilename\n");
|
||||
return 1;
|
||||
}
|
||||
skynet_globalinit();
|
||||
skynet_env_init();
|
||||
|
||||
@@ -132,6 +132,8 @@ skynet_context_new(const char * name, const char *param) {
|
||||
|
||||
ctx->init = false;
|
||||
ctx->endless = false;
|
||||
// Should set to 0 first to avoid skynet_handle_retireall get an uninitialized handle
|
||||
ctx->handle = 0;
|
||||
ctx->handle = skynet_handle_register(ctx);
|
||||
struct message_queue * queue = ctx->queue = skynet_mq_create(ctx->handle);
|
||||
// init function maybe use ctx->handle, so it must init at last
|
||||
@@ -230,7 +232,7 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
|
||||
size_t sz = msg->sz & HANDLE_MASK;
|
||||
if (!ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz)) {
|
||||
skynet_free(msg->data);
|
||||
}
|
||||
}
|
||||
CHECKCALLING_END(ctx)
|
||||
}
|
||||
|
||||
@@ -612,8 +614,10 @@ skynet_send(struct skynet_context * context, uint32_t source, uint32_t destinati
|
||||
}
|
||||
|
||||
int
|
||||
skynet_sendname(struct skynet_context * context, const char * addr , int type, int session, void * data, size_t sz) {
|
||||
uint32_t source = context->handle;
|
||||
skynet_sendname(struct skynet_context * context, uint32_t source, const char * addr , int type, int session, void * data, size_t sz) {
|
||||
if (source == 0) {
|
||||
source = context->handle;
|
||||
}
|
||||
uint32_t des = 0;
|
||||
if (addr[0] == ':') {
|
||||
des = strtoul(addr+1, NULL, 16);
|
||||
@@ -646,11 +650,6 @@ skynet_context_handle(struct skynet_context *ctx) {
|
||||
return ctx->handle;
|
||||
}
|
||||
|
||||
void
|
||||
skynet_context_init(struct skynet_context *ctx, uint32_t handle) {
|
||||
ctx->handle = handle;
|
||||
}
|
||||
|
||||
void
|
||||
skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb) {
|
||||
context->cb = cb;
|
||||
|
||||
@@ -12,7 +12,6 @@ struct skynet_context * skynet_context_new(const char * name, const char * parm)
|
||||
void skynet_context_grab(struct skynet_context *);
|
||||
struct skynet_context * skynet_context_release(struct skynet_context *);
|
||||
uint32_t skynet_context_handle(struct skynet_context *);
|
||||
void skynet_context_init(struct skynet_context *, uint32_t handle);
|
||||
int skynet_context_push(uint32_t handle, struct skynet_message *message);
|
||||
void skynet_context_send(struct skynet_context * context, void * msg, size_t sz, uint32_t source, int type, int session);
|
||||
int skynet_context_newsession(struct skynet_context *);
|
||||
|
||||
@@ -1,10 +0,0 @@
|
||||
local skynet = require "skynet"
|
||||
|
||||
skynet.start(function()
|
||||
local ping = skynet.newservice("pingserver")
|
||||
skynet.timeout(0,function()
|
||||
print(skynet.call(ping,"lua","PING","ping"))
|
||||
end)
|
||||
|
||||
print(skynet.blockcall(ping,"lua","HELLO"))
|
||||
end)
|
||||
@@ -11,6 +11,15 @@ skynet.start(function()
|
||||
end)
|
||||
end)
|
||||
|
||||
elseif mode == "dead" then
|
||||
|
||||
skynet.start(function()
|
||||
skynet.dispatch("lua", function (...)
|
||||
skynet.sleep(100)
|
||||
print("return", skynet.ret "")
|
||||
end)
|
||||
end)
|
||||
|
||||
else
|
||||
|
||||
skynet.start(function()
|
||||
@@ -20,6 +29,9 @@ else
|
||||
skynet.call(test,"lua", "dead call")
|
||||
end))
|
||||
|
||||
skynet.exit()
|
||||
local dead = skynet.newservice(SERVICE_NAME, "dead") -- launch self in dead mode
|
||||
|
||||
skynet.timeout(0, skynet.exit) -- exit after a while, so the call never return
|
||||
skynet.call(dead, "lua", "whould not return")
|
||||
end)
|
||||
end
|
||||
end
|
||||
|
||||
44
test/testecho.lua
Normal file
44
test/testecho.lua
Normal file
@@ -0,0 +1,44 @@
|
||||
local skynet = require "skynet"
|
||||
|
||||
local mode = ...
|
||||
|
||||
if mode == "slave" then
|
||||
|
||||
skynet.start(function()
|
||||
skynet.dispatch("lua", function(_,_, ...)
|
||||
skynet.ret(skynet.pack(...))
|
||||
end)
|
||||
end)
|
||||
|
||||
else
|
||||
|
||||
skynet.start(function()
|
||||
local slave = skynet.newservice(SERVICE_NAME, "slave")
|
||||
local n = 100000
|
||||
local start = skynet.now()
|
||||
print("call salve", n, "times in queue")
|
||||
for i=1,n do
|
||||
skynet.call(slave, "lua")
|
||||
end
|
||||
print("qps = ", n/ (skynet.now() - start) * 100)
|
||||
|
||||
start = skynet.now()
|
||||
|
||||
local worker = 10
|
||||
local task = n/worker
|
||||
print("call salve", n, "times in parallel, worker = ", worker)
|
||||
|
||||
for i=1,worker do
|
||||
skynet.fork(function()
|
||||
for i=1,task do
|
||||
skynet.call(slave, "lua")
|
||||
end
|
||||
worker = worker -1
|
||||
if worker == 0 then
|
||||
print("qps = ", n/ (skynet.now() - start) * 100)
|
||||
end
|
||||
end)
|
||||
end
|
||||
end)
|
||||
|
||||
end
|
||||
@@ -6,6 +6,7 @@ skynet.start(function()
|
||||
print("run skynet examples/config_log please")
|
||||
harbor.connect(2)
|
||||
print("harbor 2 connected")
|
||||
print("LOG =", skynet.address(harbor.queryname "LOG"))
|
||||
harbor.link(2)
|
||||
print("disconnected")
|
||||
end)
|
||||
|
||||
36
test/teststm.lua
Normal file
36
test/teststm.lua
Normal file
@@ -0,0 +1,36 @@
|
||||
local skynet = require "skynet"
|
||||
local stm = require "stm"
|
||||
|
||||
local mode = ...
|
||||
|
||||
if mode == "slave" then
|
||||
|
||||
skynet.start(function()
|
||||
skynet.dispatch("lua", function (_,_, obj)
|
||||
local obj = stm.newcopy(obj)
|
||||
print("read:", obj(skynet.unpack))
|
||||
skynet.ret()
|
||||
skynet.error("sleep and read")
|
||||
for i=1,10 do
|
||||
skynet.sleep(10)
|
||||
print("read:", obj(skynet.unpack))
|
||||
end
|
||||
skynet.exit()
|
||||
end)
|
||||
end)
|
||||
|
||||
else
|
||||
|
||||
skynet.start(function()
|
||||
local slave = skynet.newservice(SERVICE_NAME, "slave")
|
||||
local obj = stm.new(skynet.pack(1,2,3,4,5))
|
||||
local copy = stm.copy(obj)
|
||||
skynet.call(slave, "lua", copy)
|
||||
for i=1,5 do
|
||||
skynet.sleep(20)
|
||||
print("write", i)
|
||||
obj(skynet.pack("hello world", i))
|
||||
end
|
||||
skynet.exit()
|
||||
end)
|
||||
end
|
||||
Reference in New Issue
Block a user