diff --git a/HISTORY.md b/HISTORY.md index 375687fd..aca45755 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -1,3 +1,20 @@ +Dev version +----------- +* 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 diff --git a/Makefile b/Makefile index 63d9085d..ed8d1c7d 100644 --- a/Makefile +++ b/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 diff --git a/examples/cluster1.lua b/examples/cluster1.lua index 551b0f38..6bb3b68d 100644 --- a/examples/cluster1.lua +++ b/examples/cluster1.lua @@ -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) diff --git a/examples/cluster2.lua b/examples/cluster2.lua index a61ff5c4..a49aa718 100644 --- a/examples/cluster2.lua +++ b/examples/cluster2.lua @@ -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) diff --git a/examples/globallog.lua b/examples/globallog.lua index ffd0a0f3..2939756c 100644 --- a/examples/globallog.lua +++ b/examples/globallog.lua @@ -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) diff --git a/examples/share.lua b/examples/share.lua new file mode 100644 index 00000000..21c3bf54 --- /dev/null +++ b/examples/share.lua @@ -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 diff --git a/lualib-src/lua-cluster.c b/lualib-src/lua-cluster.c index 29ebb08a..8ba1b7d6 100644 --- a/lualib-src/lua-cluster.c +++ b/lualib-src/lua-cluster.c @@ -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 }, diff --git a/lualib-src/lua-mongo.c b/lualib-src/lua-mongo.c index 855cc385..af63d0b0 100644 --- a/lualib-src/lua-mongo.c +++ b/lualib-src/lua-mongo.c @@ -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); } } diff --git a/lualib-src/lua-multicast.c b/lualib-src/lua-multicast.c index 25b9d936..866821d4 100644 --- a/lualib-src/lua-multicast.c +++ b/lualib-src/lua-multicast.c @@ -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 }, diff --git a/lualib-src/lua-sharedata.c b/lualib-src/lua-sharedata.c new file mode 100644 index 00000000..1a0f4f74 --- /dev/null +++ b/lualib-src/lua-sharedata.c @@ -0,0 +1,789 @@ +// build: gcc -O2 -Wall --shared -o conf.so luaconf.c + +#include +#include +#include +#include +#include +#include + +#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;ihash[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;iarraytype[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;ihash[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;isizearray;i++) { + if (tbl->arraytype[i] == VALUETYPE_TABLE) { + delete_tbl(tbl->array[i].tbl); + } + } + for (i=0;isizehash;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;isizearray;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;iarraytype[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; +} diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c index 0189b61b..644c1b2e 100644 --- a/lualib-src/lua-skynet.c +++ b/lualib-src/lua-skynet.c @@ -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,46 +126,6 @@ _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); - } - - 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; -} - /* unsigned address string address @@ -167,28 +138,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 = lua_tostring(L, 1); } int type = luaL_checkinteger(L, 2); @@ -207,13 +160,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 +192,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 = lua_tostring(L,1); + } uint32_t source = luaL_checkunsigned(L,2); int type = luaL_checkinteger(L,3); int session = luaL_checkinteger(L,4); @@ -244,13 +209,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 +272,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 +307,7 @@ luaopen_skynet_c(lua_State *L) { { "pack", _luaseri_pack }, { "unpack", _luaseri_unpack }, { "packstring", lpackstring }, + { "trash" , ltrash }, { "callback", _callback }, { NULL, NULL }, }; diff --git a/lualib-src/lua-stm.c b/lualib-src/lua-stm.c new file mode 100644 index 00000000..afe26f07 --- /dev/null +++ b/lualib-src/lua-stm.c @@ -0,0 +1,244 @@ +#include +#include +#include +#include +#include + +#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; +} diff --git a/lualib/cluster.lua b/lualib/cluster.lua index d53e6d9b..ddb00d13 100644 --- a/lualib/cluster.lua +++ b/lualib/cluster.lua @@ -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) diff --git a/lualib/mqueue.lua b/lualib/mqueue.lua index f21921ca..13c2f327 100644 --- a/lualib/mqueue.lua +++ b/lualib/mqueue.lua @@ -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 diff --git a/lualib/multicast.lua b/lualib/multicast.lua index 502f8d21..1f5fbd15 100644 --- a/lualib/multicast.lua +++ b/lualib/multicast.lua @@ -1,5 +1,5 @@ local skynet = require "skynet" -local mc = require "multicast.c" +local mc = require "multicast.core" local multicastd local multicast = {} diff --git a/lualib/sharedata.lua b/lualib/sharedata.lua new file mode 100644 index 00000000..58ecf28d --- /dev/null +++ b/lualib/sharedata.lua @@ -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 diff --git a/lualib/sharedata/corelib.lua b/lualib/sharedata/corelib.lua new file mode 100644 index 00000000..d9a5488a --- /dev/null +++ b/lualib/sharedata/corelib.lua @@ -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 \ No newline at end of file diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 5e8e95b0..6dd32a46 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -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 diff --git a/lualib/skynet/debug.lua b/lualib/skynet/debug.lua index c8f51f0b..5b7b5afa 100644 --- a/lualib/skynet/debug.lua +++ b/lualib/skynet/debug.lua @@ -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 \ No newline at end of file +end diff --git a/lualib/skynet/harbor.lua b/lualib/skynet/harbor.lua index 5f470be8..b5863921 100644 --- a/lualib/skynet/harbor.lua +++ b/lualib/skynet/harbor.lua @@ -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 diff --git a/lualib/skynet/inject.lua b/lualib/skynet/inject.lua new file mode 100644 index 00000000..68e3954b --- /dev/null +++ b/lualib/skynet/inject.lua @@ -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 diff --git a/lualib/snax/hotfix.lua b/lualib/snax/hotfix.lua index e0cc2a7f..1a60383a 100644 --- a/lualib/snax/hotfix.lua +++ b/lualib/snax/hotfix.lua @@ -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 diff --git a/lualib/snax/interface.lua b/lualib/snax/interface.lua index 24d52c08..a2f8aa64 100644 --- a/lualib/snax/interface.lua +++ b/lualib/snax/interface.lua @@ -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 diff --git a/lualib/snax/loginserver.lua b/lualib/snax/loginserver.lua index c5de769f..c099ddb8 100644 --- a/lualib/snax/loginserver.lua +++ b/lualib/snax/loginserver.lua @@ -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 diff --git a/lualib/socketchannel.lua b/lualib/socketchannel.lua index 45aeea54..1086cd79 100644 --- a/lualib/socketchannel.lua +++ b/lualib/socketchannel.lua @@ -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) diff --git a/service-src/service_snlua.c b/service-src/service_snlua.c index 33e0a5ff..50530e33 100644 --- a/service-src/service_snlua.c +++ b/service-src/service_snlua.c @@ -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 * diff --git a/service/bootstrap.lua b/service/bootstrap.lua index 11303577..e1df7170 100644 --- a/service/bootstrap.lua +++ b/service/bootstrap.lua @@ -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) diff --git a/service/cdummy.lua b/service/cdummy.lua index fddb2c01..52b70d7c 100644 --- a/service/cdummy.lua +++ b/service/cdummy.lua @@ -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 diff --git a/service/clusterd.lua b/service/clusterd.lua index f7a9e90f..677f84fa 100644 --- a/service/clusterd.lua +++ b/service/clusterd.lua @@ -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 = {} diff --git a/service/clusterproxy.lua b/service/clusterproxy.lua new file mode 100644 index 00000000..6edcb674 --- /dev/null +++ b/service/clusterproxy.lua @@ -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) diff --git a/service/console.lua b/service/console.lua index 86bf6cc1..c0eba879 100644 --- a/service/console.lua +++ b/service/console.lua @@ -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) diff --git a/service/cslave.lua b/service/cslave.lua index 6f005b2f..fdbe9088 100644 --- a/service/cslave.lua +++ b/service/cslave.lua @@ -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)) diff --git a/service/datacenterd.lua b/service/datacenterd.lua index 1a0fee65..5929b0e3 100644 --- a/service/datacenterd.lua +++ b/service/datacenterd.lua @@ -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]) diff --git a/service/debug_console.lua b/service/debug_console.lua index a587f69d..b627eb7a 100644 --- a/service/debug_console.lua +++ b/service/debug_console.lua @@ -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 diff --git a/service/launcher.lua b/service/launcher.lua index 5c1c1a8b..e4197035 100644 --- a/service/launcher.lua +++ b/service/launcher.lua @@ -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 diff --git a/service/multicastd.lua b/service/multicastd.lua index dba56359..5a7f9d2a 100644 --- a/service/multicastd.lua +++ b/service/multicastd.lua @@ -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()) diff --git a/service/sharedatad.lua b/service/sharedatad.lua new file mode 100644 index 00000000..eb3346c4 --- /dev/null +++ b/service/sharedatad.lua @@ -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) + diff --git a/service/snaxd.lua b/service/snaxd.lua index de41d6c3..0ff9baa0 100644 --- a/service/snaxd.lua +++ b/service/snaxd.lua @@ -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 = {} diff --git a/skynet-src/skynet.h b/skynet-src/skynet.h index 9ae1759c..fc8235e0 100644 --- a/skynet-src/skynet.h +++ b/skynet-src/skynet.h @@ -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); diff --git a/skynet-src/skynet_handle.c b/skynet-src/skynet_handle.c index 58cabc0b..dc20dcf2 100644 --- a/skynet-src/skynet_handle.c +++ b/skynet-src/skynet_handle.c @@ -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; ilock); + + return ret; } void @@ -105,10 +108,14 @@ skynet_handle_retireall() { for (i=0;islot_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) diff --git a/skynet-src/skynet_handle.h b/skynet-src/skynet_handle.h index a39b47bb..e067eed0 100644 --- a/skynet-src/skynet_handle.h +++ b/skynet-src/skynet_handle.h @@ -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(); diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index 3e902cc9..eaf6e14d 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -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; diff --git a/skynet-src/skynet_server.h b/skynet-src/skynet_server.h index 5d07ba70..27b6b768 100644 --- a/skynet-src/skynet_server.h +++ b/skynet-src/skynet_server.h @@ -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 *); diff --git a/test/testblockcall.lua b/test/testblockcall.lua deleted file mode 100644 index b0d7e76f..00000000 --- a/test/testblockcall.lua +++ /dev/null @@ -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) diff --git a/test/testdeadcall.lua b/test/testdeadcall.lua index 9017d380..b34dccb8 100644 --- a/test/testdeadcall.lua +++ b/test/testdeadcall.lua @@ -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 \ No newline at end of file +end diff --git a/test/testecho.lua b/test/testecho.lua new file mode 100644 index 00000000..a6b265ef --- /dev/null +++ b/test/testecho.lua @@ -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 diff --git a/test/testharborlink.lua b/test/testharborlink.lua index 031e38ec..dfdb8613 100644 --- a/test/testharborlink.lua +++ b/test/testharborlink.lua @@ -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) diff --git a/test/teststm.lua b/test/teststm.lua new file mode 100644 index 00000000..7416a9f2 --- /dev/null +++ b/test/teststm.lua @@ -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