Compare commits

...

43 Commits

Author SHA1 Message Date
Cloud Wu
f393b64124 remove watch from service_mgr 2014-08-18 17:40:18 +08:00
Cloud Wu
9ea756ce4f fix typo 2014-08-18 16:49:14 +08:00
Cloud Wu
f9438461eb check dest type 2014-08-18 16:44:19 +08:00
Cloud Wu
0fb23f9ab8 remove default config name 2014-08-18 16:23:49 +08:00
Cloud Wu
0390818a02 release v0.6.0 2014-08-18 11:31:05 +08:00
云风
e002796d2e Merge pull request #153 from cloudwu/dev
prerelease v0.6.0
2014-08-18 11:28:18 +08:00
Cloud Wu
b8a0e35f21 bugfix: mongo driver 2014-08-16 15:19:01 +08:00
Cloud Wu
bd52c5fece bugfix: socketchannel connect once 2014-08-14 17:03:09 +08:00
Cloud Wu
d2cea9b70f socketchannel request only try connect once 2014-08-14 13:49:49 +08:00
Cloud Wu
7e24d46047 clusterproxy 2014-08-14 11:11:01 +08:00
Cloud Wu
8078d777ab split skynet.debug to skynet.inject 2014-08-13 21:24:48 +08:00
Cloud Wu
939ef564ce debug run extract proto's upvalues 2014-08-13 21:12:07 +08:00
Cloud Wu
6bf8dd9a31 add debug command inject 2014-08-13 21:02:56 +08:00
Cloud Wu
809e6c3a89 add debug command: exit 2014-08-13 19:37:26 +08:00
Cloud Wu
845acd3314 cluster servicename is not stable, so don't cache address 2014-08-13 19:10:51 +08:00
Cloud Wu
2cf69a6ef0 use localname .simpledb 2014-08-13 18:05:11 +08:00
Cloud Wu
978e010e67 add cluster.proxy and cluster.ncall 2014-08-13 17:49:30 +08:00
Cloud Wu
cde423cb73 add skynet.harbor.queryname 2014-08-13 14:32:28 +08:00
Cloud Wu
cfe8b19c8a change .slave to .cslave 2014-08-12 21:41:28 +08:00
Cloud Wu
f69f6e8e70 add stm object support. (STM : Software transactional memory) 2014-08-12 14:33:36 +08:00
Cloud Wu
d7b228f342 use .core for lua c module name 2014-08-11 15:45:16 +08:00
Cloud Wu
0954a906f4 skynet.redirect support string address 2014-08-11 15:29:48 +08:00
Cloud Wu
77d53cf5f1 release v0.5.2 2014-08-11 09:45:37 +08:00
云风
5621c52a44 Merge pull request #150 from cloudwu/http
Http
2014-08-11 09:43:10 +08:00
Cloud Wu
f6fa6c7ded timer support more than 497 days 2014-08-08 20:33:38 +08:00
Cloud Wu
023e4abd87 reset current_point when time error 2014-08-08 15:38:06 +08:00
Cloud Wu
cab10edd1d snaxd change add package.path 2014-08-07 19:34:48 +08:00
Cloud Wu
87a02ffe72 don't check imported function in snax.hotfix 2014-08-07 18:06:08 +08:00
Cloud Wu
e739df0a32 fix log 2014-08-04 18:17:35 +08:00
Cloud Wu
b18962929b bugfix: chunked mode 2014-08-04 18:11:11 +08:00
Cloud Wu
dec2e6fe4c add httpc 2014-08-04 18:06:16 +08:00
Cloud Wu
7beed39b1d bugfix: use temp array for each request 2014-08-04 16:34:29 +08:00
Cloud Wu
ae5a9bb855 Merge branch 'master' into dev 2014-08-04 14:23:13 +08:00
Cloud Wu
c48e0e4c32 Merge branch 'master' into dev 2014-08-02 18:33:08 +08:00
Cloud Wu
0747a7d7e1 remove register_slave 2014-07-31 20:30:22 +08:00
Cloud Wu
4b0a648b21 Merge branch 'master' into dev 2014-07-31 20:13:57 +08:00
Cloud Wu
40519e9ee1 add response test mode 2014-07-30 14:54:18 +08:00
Cloud Wu
ce5adba5b2 add skynet.response for delay response 2014-07-30 12:10:30 +08:00
Cloud Wu
c78466b486 remove debug print 2014-07-29 21:56:36 +08:00
Cloud Wu
deed51d309 fix typo 2014-07-29 21:30:18 +08:00
Cloud Wu
68a952ff17 bugfix: update until delete 2014-07-29 21:22:49 +08:00
Cloud Wu
434cd1ca79 add sharedata 2014-07-29 20:56:35 +08:00
Cloud Wu
8e09fad233 remove unused skynet_context_init 2014-07-29 10:31:45 +08:00
58 changed files with 2451 additions and 404 deletions

View File

@@ -1,3 +1,33 @@
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
* Bugifx : http chunked mode
* Add : httpc
* timer support more than 497 days
v0.5.1 (2014-8-4)
-----------
* Bugfix : http module
* Bugfix : multicast local channel delete
* Bugfix : socket.read(fd)
v0.5.0 (2014-7-28)
-----------
* skynet.exit will quit service immediately.

View File

@@ -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

View File

@@ -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)

View File

@@ -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)

View File

@@ -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
View 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

View File

@@ -39,6 +39,11 @@ skynet.start(function()
table.insert(tmp, string.format("query: %s= %s", k,v))
end
end
table.insert(tmp, "-----header----")
for k,v in pairs(header) do
table.insert(tmp, string.format("%s = %s",k,v))
end
table.insert(tmp, "-----body----\n" .. body)
response(id, code, table.concat(tmp,"\n"))
end
else

View File

@@ -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 },

View File

@@ -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);
}
}

View File

@@ -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
View 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;
}

View File

@@ -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
View 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(&copy->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;
}

View File

@@ -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)

114
lualib/http/httpc.lua Normal file
View File

@@ -0,0 +1,114 @@
local socket = require "http.sockethelper"
local url = require "http.url"
local internal = require "http.internal"
local string = string
local table = table
local httpc = {}
local function request(fd, method, host, url, recvheader, header, content)
local read = socket.readfunc(fd)
local write = socket.writefunc(fd)
local header_content = ""
if header then
for k,v in pairs(header) do
header_content = string.format("%s%s:%s\r\n", header_content, k, v)
end
end
if content then
local data = string.format("%s %s HTTP/1.1\r\nhost:%s\r\ncontent-length:%d\r\n%s\r\n%s", method, url, host, #content, header_content, content)
write(data)
else
local request_header = string.format("%s %s HTTP/1.1\r\nhost:%s\r\ncontent-length:0\r\n%s\r\n", method, url, host, header_content)
write(request_header)
end
local tmpline = {}
local body = internal.recvheader(read, tmpline, "")
if not body then
error(socket.socket_error)
end
local statusline = tmpline[1]
local code, info = statusline:match "HTTP/[%d%.]+%s+([%d]+)%s+(.*)$"
code = assert(tonumber(code))
local header = internal.parseheader(tmpline,2,recvheader or {})
if not header then
error("Invalid HTTP response header")
end
local length = header["content-length"]
if length then
length = tonumber(length)
end
local mode = header["transfer-encoding"]
if mode then
if mode ~= "identity" and mode ~= "chunked" then
error ("Unsupport transfer-encoding")
end
end
if mode == "chunked" then
body, header = internal.recvchunkedbody(read, nil, header, body)
if not body then
error("Invalid response body")
end
else
-- identity mode
if length then
if #body >= length then
body = body:sub(1,length)
else
local padding = read(length - #body)
body = body .. padding
end
else
body = nil
end
end
return code, body
end
function httpc.request(method, host, url, recvheader, header, content)
local hostname, port = host:match"([^:]+):?(%d*)$"
if port == "" then
port = 80
else
port = tonumber(port)
end
local fd = socket.connect(hostname, port)
local ok , statuscode, body = pcall(request, fd,method, host, url, recvheader, header, content)
if ok then
return statuscode, body
else
socket.close(fd)
error(statuscode)
end
end
function httpc.get(...)
return httpc.request("GET", ...)
end
local function escape(s)
return (string.gsub(s, "([^A-Za-z0-9_])", function(c)
return string.format("%%%02X", string.byte(c))
end))
end
function httpc.post(host, url, form, recvheader)
local header = {
["content-type"] = "application/x-www-form-urlencoded"
}
local body = {}
for k,v in pairs(form) do
table.insert(body, string.format("%s=%s",escape(k),escape(v)))
end
return httpc.request("POST", host, url, recvheader, header, table.concat(body , "&"))
end
return httpc

View File

@@ -1,3 +1,5 @@
local internal = require "http.internal"
local table = table
local httpd = {}
@@ -45,140 +47,9 @@ local http_status_msg = {
[505] = "HTTP Version not supported",
}
local function recvheader(readbytes, limit, lines, header)
local result
local e = header:find("\r\n\r\n", 1, true)
if e then
result = header:sub(e+4)
else
while true do
local bytes = readbytes()
header = header .. bytes
if #header > limit then
return
end
e = header:find("\r\n\r\n", -#bytes-3, true)
if e then
result = header:sub(e+4)
break
end
end
end
local idx = 1
for v in header:gmatch("(.-)\r\n") do
if v == "" then
break
end
lines[idx] = v
idx = idx + 1
end
for i = idx, #lines do
lines[i] = nil
end
return result
end
local function parseheader(lines, from, header)
local name, value
for i=from,#lines do
local line = lines[i]
if line:byte(1) == 9 then -- tab, append last line
if name == nil then
return
end
header[name] = header[name] .. line:sub(2)
else
name, value = line:match "^(.-):%s*(.*)"
if name == nil or value == nil then
return
end
name = name:lower()
if header[name] then
header[name] = header[name] .. ", " .. value
else
header[name] = value
end
end
end
return header
end
local function chunksize(readbytes, body)
while true do
if #body > 128 then
return
end
body = body .. readbytes()
local f,e = body:find("\r\n",1,true)
if f then
return tonumber(body:sub(1,f-1),16), body:sub(e+1)
end
end
end
local function readcrln(readbytes, body)
if #body > 2 then
if body:sub(1,2) ~= "\r\n" then
return
end
return body:sub(3)
else
body = body .. readbytes(2-#body)
if body ~= "\r\n" then
return
end
return ""
end
end
local tmpline = {}
local function recvchunkedbody(readbytes, bodylimit, header, body)
local result = ""
local size = 0
while true do
local sz
sz , body = chunksize(readbytes, body)
if not sz then
return
end
if sz == 0 then
break
end
size = size + sz
if bodylimit and size > bodylimit then
return
end
if #body >= sz then
result = result .. body:sub(1,sz)
body = body:sub(sz+1)
else
result = result .. body .. readbytes(sz - #body)
body = ""
end
body = readcrln(readbytes, body)
if not body then
return
end
end
body = readcrln(readbytes, body)
if not body then
return
end
body = recvheader(readbytes, 8192, tmpline, body)
if not body then
return
end
header = parseheader(tmpline,1,header)
return result, header
end
local function readall(readbytes, bodylimit)
local body = recvheader(readbytes, 8192, tmpline, "")
local tmpline = {}
local body = internal.recvheader(readbytes, tmpline, "")
if not body then
return 413 -- Request Entity Too Large
end
@@ -189,7 +60,7 @@ local function readall(readbytes, bodylimit)
if httpver < 1.0 or httpver > 1.1 then
return 505 -- HTTP Version not supported
end
local header = parseheader(tmpline,2,{})
local header = internal.parseheader(tmpline,2,{})
if not header then
return 400 -- Bad request
end
@@ -199,13 +70,13 @@ local function readall(readbytes, bodylimit)
end
local mode = header["transfer-encoding"]
if mode then
if mode ~= "identity" or mode ~= "chunked" then
if mode ~= "identity" and mode ~= "chunked" then
return 501 -- Not Implemented
end
end
if mode == "chunked" then
body, header = recvchunkedbody(readbytes, bodylimit, header, body)
body, header = internal.recvchunkedbody(readbytes, bodylimit, header, body)
if not body then
return 413
end

135
lualib/http/internal.lua Normal file
View File

@@ -0,0 +1,135 @@
local M = {}
local LIMIT = 8192
local function chunksize(readbytes, body)
while true do
if #body > 128 then
return
end
body = body .. readbytes()
local f,e = body:find("\r\n",1,true)
if f then
return tonumber(body:sub(1,f-1),16), body:sub(e+1)
end
end
end
local function readcrln(readbytes, body)
if #body >= 2 then
if body:sub(1,2) ~= "\r\n" then
return
end
return body:sub(3)
else
body = body .. readbytes(2-#body)
if body ~= "\r\n" then
return
end
return ""
end
end
function M.recvheader(readbytes, lines, header)
if #header >= 2 then
if header:find "^\r\n" then
return header:sub(3)
end
end
local result
local e = header:find("\r\n\r\n", 1, true)
if e then
result = header:sub(e+4)
else
while true do
local bytes = readbytes()
header = header .. bytes
if #header > LIMIT then
return
end
e = header:find("\r\n\r\n", -#bytes-3, true)
if e then
result = header:sub(e+4)
break
end
if header:find "^\r\n" then
return header:sub(3)
end
end
end
for v in header:gmatch("(.-)\r\n") do
if v == "" then
break
end
table.insert(lines, v)
end
return result
end
function M.parseheader(lines, from, header)
local name, value
for i=from,#lines do
local line = lines[i]
if line:byte(1) == 9 then -- tab, append last line
if name == nil then
return
end
header[name] = header[name] .. line:sub(2)
else
name, value = line:match "^(.-):%s*(.*)"
if name == nil or value == nil then
return
end
name = name:lower()
if header[name] then
header[name] = header[name] .. ", " .. value
else
header[name] = value
end
end
end
return header
end
function M.recvchunkedbody(readbytes, bodylimit, header, body)
local result = ""
local size = 0
while true do
local sz
sz , body = chunksize(readbytes, body)
if not sz then
return
end
if sz == 0 then
break
end
size = size + sz
if bodylimit and size > bodylimit then
return
end
if #body >= sz then
result = result .. body:sub(1,sz)
body = body:sub(sz+1)
else
result = result .. body .. readbytes(sz - #body)
body = ""
end
body = readcrln(readbytes, body)
if not body then
return
end
end
local tmpline = {}
body = M.recvheader(readbytes, tmpline, body)
if not body then
return
end
header = M.parseheader(tmpline,1,header)
return result, header
end
return M

View File

@@ -28,4 +28,16 @@ function sockethelper.writefunc(fd)
end
end
function sockethelper.connect(host, port)
local fd = socket.open(host, port)
if fd then
return fd
end
error(socket_error)
end
function sockethelper.close(fd)
socket.close(fd)
end
return sockethelper

View File

@@ -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

View File

@@ -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
View 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

View 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

View File

@@ -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

View File

@@ -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

View File

@@ -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
View 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

View File

@@ -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

View File

@@ -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

View File

@@ -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

View File

@@ -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)

View File

@@ -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 *

View File

@@ -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)

View File

@@ -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

View File

@@ -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
View 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)

View File

@@ -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)

View File

@@ -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))

View File

@@ -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])

View File

@@ -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

View File

@@ -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)

View File

@@ -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

View File

@@ -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())

View File

@@ -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
View 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)

View File

@@ -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 = {}

View File

@@ -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);

View File

@@ -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)

View File

@@ -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();

View File

@@ -89,7 +89,6 @@ static const char * load_config = "\
local code = assert(f:read \'*a\')\
local function getenv(name) return assert(os.getenv(name), name) end\
code = string.gsub(code, \'%$([%w_%d]+)\', getenv)\
print(code)\
f:close()\
local result = {}\
assert(load(code,\'=(load)\',\'t\',result))()\
@@ -98,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();

View File

@@ -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;

View File

@@ -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 *);

View File

@@ -9,6 +9,7 @@
#include <assert.h>
#include <string.h>
#include <stdlib.h>
#include <stdint.h>
#if defined(__APPLE__)
#include <sys/time.h>
@@ -33,7 +34,7 @@ struct timer_event {
struct timer_node {
struct timer_node *next;
int expire;
uint32_t expire;
};
struct link_list {
@@ -43,9 +44,9 @@ struct link_list {
struct timer {
struct link_list near[TIME_NEAR];
struct link_list t[4][TIME_LEVEL-1];
struct link_list t[4][TIME_LEVEL];
int lock;
int time;
uint32_t time;
uint32_t current;
uint32_t starttime;
uint64_t current_point;
@@ -72,21 +73,22 @@ link(struct link_list *list,struct timer_node *node) {
static void
add_node(struct timer *T,struct timer_node *node) {
int time=node->expire;
int current_time=T->time;
uint32_t time=node->expire;
uint32_t current_time=T->time;
if ((time|TIME_NEAR_MASK)==(current_time|TIME_NEAR_MASK)) {
link(&T->near[time&TIME_NEAR_MASK],node);
} else {
int i;
int mask=TIME_NEAR << TIME_LEVEL_SHIFT;
uint32_t mask=TIME_NEAR << TIME_LEVEL_SHIFT;
for (i=0;i<3;i++) {
if ((time|(mask-1))==(current_time|(mask-1))) {
break;
}
mask <<= TIME_LEVEL_SHIFT;
}
link(&T->t[i][((time>>(TIME_NEAR_SHIFT + i*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)-1],node);
link(&T->t[i][((time>>(TIME_NEAR_SHIFT + i*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node);
}
}
@@ -103,29 +105,38 @@ timer_add(struct timer *T,void *arg,size_t sz,int time) {
UNLOCK(T);
}
static void
move_list(struct timer *T, int level, int idx) {
struct timer_node *current = link_clear(&T->t[level][idx]);
while (current) {
struct timer_node *temp=current->next;
add_node(T,current);
current=temp;
}
}
static void
timer_shift(struct timer *T) {
LOCK(T);
int mask = TIME_NEAR;
int time = (++T->time) >> TIME_NEAR_SHIFT;
int i=0;
while ((T->time & (mask-1))==0) {
int idx=time & TIME_LEVEL_MASK;
if (idx!=0) {
--idx;
struct timer_node *current = link_clear(&T->t[i][idx]);
while (current) {
struct timer_node *temp=current->next;
add_node(T,current);
current=temp;
uint32_t ct = ++T->time;
if (ct == 0) {
move_list(T, 3, 0);
} else {
uint32_t time = ct >> TIME_NEAR_SHIFT;
int i=0;
while ((ct & (mask-1))==0) {
int idx=time & TIME_LEVEL_MASK;
if (idx!=0) {
move_list(T, i, idx);
break;
}
break;
mask <<= TIME_LEVEL_SHIFT;
time >>= TIME_LEVEL_SHIFT;
++i;
}
mask <<= TIME_LEVEL_SHIFT;
time >>= TIME_LEVEL_SHIFT;
++i;
}
}
UNLOCK(T);
}
@@ -187,7 +198,7 @@ timer_create_timer() {
}
for (i=0;i<4;i++) {
for (j=0;j<TIME_LEVEL-1;j++) {
for (j=0;j<TIME_LEVEL;j++) {
link_clear(&r->t[i][j]);
}
}
@@ -265,6 +276,7 @@ skynet_updatetime(void) {
uint64_t cp = gettime();
if(cp < TI->current_point) {
skynet_error(NULL, "time diff error: change from %lld to %lld", cp, TI->current_point);
TI->current_point = cp;
} else if (cp != TI->current_point) {
uint32_t diff = (uint32_t)(cp - TI->current_point);
TI->current_point = cp;

View File

@@ -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)

View File

@@ -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
View 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

View File

@@ -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)

16
test/testhttp.lua Normal file
View File

@@ -0,0 +1,16 @@
local skynet = require "skynet"
local httpc = require "http.httpc"
skynet.start(function()
print("GET baidu.com")
local header = {}
local status, body = httpc.get("baidu.com", "/", header)
print("[header] =====>")
for k,v in pairs(header) do
print(k,v)
end
print("[body] =====>", status)
print(body)
skynet.exit()
end)

36
test/teststm.lua Normal file
View 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