Compare commits

...

120 Commits

Author SHA1 Message Date
Cloud Wu
2ee85a91f9 Release 1.2.0 2018-11-06 11:11:29 +08:00
Cloud Wu
8a3a0c1b71 register logger name in framework, see #909 2018-10-21 13:21:51 +08:00
Sean Feng
b4752f2394 multiresultset spelling mistake 2018-10-17 19:54:38 +08:00
Cloud Wu
f110b51290 add socketchannel.overload 2018-10-17 14:54:44 +08:00
Cloud Wu
e0f16b8df5 remove dead_service table, address may reused 2018-10-15 10:05:23 +08:00
Cloud Wu
08e56f4316 fix #793 2018-10-11 17:38:49 +08:00
Cloud Wu
7e57e498f6 remove dead_service table, address may reused 2018-10-08 15:44:12 +08:00
Cloud Wu
98b5c7c7c6 address 0 is reserved 2018-10-08 14:20:53 +08:00
sundream
02d5f4fbe9 rediscluster: support eval/evalshal + subscribe/publish 2018-09-27 10:00:30 +08:00
Hanlei Qin
818ff8314d remove unused code 2018-09-27 09:59:18 +08:00
Hanlei Qin
d23e6f8d8e 1. remove unused code; 2. weird indents fixed. 2018-09-20 00:31:14 +08:00
Hanlei Qin
f217a56f6e Uniform indent to tab and clear up. 2018-09-20 00:29:19 +08:00
Hanlei Qin
46b49adf9e remove useless file 2018-09-16 18:41:30 +08:00
Hanlei Qin
d9fb4d06ec remove useless local function 2018-09-16 18:41:06 +08:00
Cloud Wu
34a9979255 check launching address 2018-09-13 18:59:03 +08:00
Cloud Wu
89e7a06deb add more info when service init blocked 2018-09-13 18:27:05 +08:00
Cloud Wu
43d02ac456 update jemalloc to 5.1.0 2018-09-12 15:23:59 +08:00
Cloud Wu
c839159470 remove netpack from gate service 2018-09-12 11:45:41 +08:00
Cloud Wu
823b773051 remove unused require 2018-09-12 11:31:12 +08:00
Cloud Wu
f8434d34b6 report last accept time 2018-09-07 16:26:56 +08:00
Cloud Wu
61980c8a1f add socket.netstat 2018-09-07 16:11:27 +08:00
Cloud Wu
35176383b7 add skynet.context 2018-09-06 19:01:10 +08:00
Cloud Wu
c96e20e6c3 add skynet.context and enhance debug console command service 2018-09-06 18:59:33 +08:00
LiuYang
ce0cdc7bd3 add mongo test case: auth 2018-08-30 11:42:47 +08:00
Cloud Wu
7e63080c03 add mongo_db:auth 2018-08-30 10:50:52 +08:00
Cloud Wu
1c8ed8a46e add mongo config option: authdb, see issue #884 2018-08-30 10:50:52 +08:00
Cloud Wu
c4334be9c3 add mongo_db:auth 2018-08-30 09:51:55 +08:00
Cloud Wu
e6e69ae805 add mongo config option: authdb, see issue #884 2018-08-30 09:38:18 +08:00
LiuYang
8ce816de7f fix mongodb testcase 2018-08-29 23:33:17 +08:00
Cloud Wu
2fb52f6fce cleanup 2018-08-29 15:29:26 +08:00
Cloud Wu
c72d4d2134 support string address name, see #883 2018-08-28 15:15:38 +08:00
btt
18a36c171f debug console inject cmd support args 2018-08-23 17:02:06 +08:00
Cloud Wu
7b3c1631d1 more stable 2018-08-17 21:27:14 +08:00
Cloud Wu
4613fde654 fix #877 2018-08-17 21:16:35 +08:00
Cloud Wu
143baec319 remove duplicate code 2018-08-13 14:12:15 +08:00
fanyh
168a07d658 sleep time 1s-> 1m, sleep 1s prodcut more message 2018-08-09 19:28:59 +08:00
Cloud Wu
281acf4c71 Use fork instead of timeout 0, see #873 2018-08-07 14:55:26 +08:00
Cloud Wu
d2ab6863e3 fix #872 2018-08-07 10:49:21 +08:00
Cloud Wu
e55ad63d3c fix #873 2018-08-07 09:43:50 +08:00
Cloud Wu
ec2eaa70a4 Merge branch 'master' of https://github.com/cloudwu/skynet 2018-08-03 11:29:12 +08:00
Cloud Wu
d402619347 see #870 2018-08-03 11:28:18 +08:00
Cloud Wu
c101d9df17 see #870 2018-08-03 11:24:14 +08:00
hong
b5653a68ea runnging拼写错误 2018-08-02 21:30:03 +08:00
hong
c9942deaa9 queryname并发导致消息乱序
当有1个以上的co进入queryname,后面的协程有在第1个以后某个的co先执行(第一个已经返回),导致cluster消息乱序
2018-08-02 21:30:03 +08:00
yenshan98
2df1446c37 Rearranged sentence to be reader friendly. 2018-07-30 15:10:57 +08:00
Cloud Wu
9421815327 add posix_memalign hook 2018-07-23 15:32:35 +08:00
Cloud Wu
1a0c9997ec fix #863 2018-07-23 14:53:01 +08:00
Cloud Wu
16f060f864 hook aligned_alloc 2018-07-20 14:46:53 +08:00
Cloud Wu
ea7c6402dd update lua to 5.3.5 2018-07-10 22:11:56 +08:00
zixun
07db2580c4 fix kqueue error flag 2018-07-05 18:55:51 +08:00
zixun
1df72c0afe fix kqueue eof event 2018-07-05 18:37:50 +08:00
Cloud Wu
0e41231298 fix #857 2018-07-04 15:10:32 +08:00
Cloud Wu
8224b5a62a udp_connect may raise error 2018-06-26 09:49:18 +08:00
Cloud Wu
d8ab9dea1b fix #852 2018-06-25 14:24:04 +08:00
Cloud Wu
a8e5c31381 check udp address, see #852 2018-06-25 12:10:30 +08:00
Cloud Wu
3a1556e1fd add new debug command trace 2018-06-08 11:24:27 +08:00
Cloud Wu
4b5fbab129 recycle coroutine bugfix 2018-06-08 10:23:36 +08:00
Cloud Wu
b1ec1f4d18 token bugfix 2018-06-08 10:23:36 +08:00
Cloud Wu
16d8f4c774 cleanup 2018-06-08 10:23:36 +08:00
Cloud Wu
f8b5ec48e8 support user tag 2018-05-28 23:13:05 +08:00
Cloud Wu
7a26926965 resume trace log 2018-05-28 21:45:41 +08:00
Cloud Wu
2990fbc7cf ignore response, fix #841 2018-05-28 20:38:10 +08:00
Cloud Wu
e8e38f1682 add session log 2018-05-28 18:16:31 +08:00
Cloud Wu
35b4a09c3d unique cluster trace tag 2018-05-28 18:09:34 +08:00
Cloud Wu
9035fbe96b cluster trace 2018-05-28 16:39:04 +08:00
Cloud Wu
a41a5fadd2 ignore ret in clientagent 2018-05-28 11:24:55 +08:00
Cloud Wu
d6f265e703 call skynet.trace() in example 2018-05-26 17:37:19 +08:00
Cloud Wu
596c77afcd trace bugfix 2018-05-26 17:17:20 +08:00
Cloud Wu
e53d985c4f add skynet.trace() 2018-05-26 16:26:50 +08:00
Cloud Wu
348e53245b skynet.wait use token instead of coroutine.running() 2018-05-25 10:42:58 +08:00
Cloud Wu
2fde0a0ae4 log error when forget response a session. Add skynet.ignoreret() 2018-05-24 16:53:27 +08:00
Cloud Wu
85ec428f3f remove assert 2018-05-22 20:04:05 +08:00
Cloud Wu
a8ab809928 add __nowating flag to cluster config, see issue #803 2018-05-22 19:49:07 +08:00
caijietao
6c7cb5313d set backlog to 128 2018-05-22 13:08:00 +08:00
spin6lock
188e38198a add ipv6 host support to service_gate.c
service_gate的地址解析用strchr找到第一个':',并将后续的字符串作为
端口号解析。填ipv6地址"0:0:0:0:0:0:0:0:9999"的时候会出错,现在改成
strrchr,从右面找到的第一个':'
2018-05-22 13:07:02 +08:00
Cloud Wu
294f557268 lua 5.3.4 bug report 6 2018-05-17 12:03:03 +08:00
Cloud Wu
4dd622e180 socket channel order mode support padding, see issue #835 2018-05-09 17:38:32 +08:00
Cloud Wu
5f8e5f9be5 add random for seed 2018-05-07 11:10:45 +08:00
Cloud Wu
e561896966 Merge branch 'master' of github.com:cloudwu/skynet 2018-05-07 10:55:34 +08:00
Cloud Wu
ea25e20c88 more random seed 2018-05-07 10:55:18 +08:00
云风
003b0fdb02 Add logo 2018-05-03 23:26:53 +08:00
xjdrew
8de6500677 update mongo driver: safe_insert, safe_update, safe_delete 2018-05-02 01:36:00 +08:00
Cloud Wu
7df13e922b add testtimeout 2018-04-20 11:44:05 +08:00
Cloud Wu
182d7c7dc7 support @name, see #823 2018-04-20 10:11:16 +08:00
Cloud Wu
f4d5bc7aa4 forward fd to agent 2018-04-18 19:32:15 +08:00
Cloud Wu
d53d98b625 Merge branch 'clusteragent' 2018-04-18 19:30:24 +08:00
Cloud Wu
2bfba716ef forward fd to agent 2018-04-18 19:29:41 +08:00
Cloud Wu
4cbe33d15d remove unused sz 2018-04-17 10:47:41 +08:00
Cloud Wu
a9852273cb add addresscommand 2018-04-17 10:47:41 +08:00
Cloud Wu
eb8a2b5e65 Improve cluster package module 2018-04-17 10:47:41 +08:00
Cloud Wu
7304a990cb use clusteragent to dispatch cluster request 2018-04-17 10:47:41 +08:00
Cloud Wu
311d7c8f6a remove unused sz 2018-04-16 12:56:34 +08:00
Cloud Wu
5a3d2fdac7 add addresscommand 2018-04-14 20:30:47 +08:00
Cloud Wu
4b88f68ab8 Improve cluster package module 2018-04-14 18:14:44 +08:00
Cloud Wu
70c224c3e7 use clusteragent to dispatch cluster request 2018-04-14 16:11:26 +08:00
Cloud Wu
1932d33b41 set node = false when node is down, see #803 2018-02-24 21:26:24 +08:00
Cloud Wu
9cf4525a8f avoid error circulation 2018-02-12 18:07:16 +08:00
Cloud Wu
0846a6533d fix a legacy problem. make harbor service more stable 2018-02-12 18:07:16 +08:00
Cloud Wu
484404295e Merge branch 'hongling0-master' 2018-02-05 11:38:22 +08:00
云风
3e54becb95 Merge branch 'master' into master 2018-02-05 11:34:52 +08:00
hong
ab13305b64 monitor中不增加ref 2018-02-05 10:18:35 +08:00
Cloud Wu
6fdb123df4 bugfix, See #800 2018-02-04 21:32:41 +08:00
Cloud Wu
6cdc1cf95c collect used response, see #800 2018-02-04 16:25:19 +08:00
hong
2dd5a06616 抄写错误 2018-02-03 15:49:01 +08:00
hong
e7436bfa95 skynet.response() leak
datasheet在monitor接口里面产生的reponse在release接口里面没有释放
2018-02-03 15:43:33 +08:00
Cloud Wu
084aab4bcd dns server auto retire 2018-01-25 15:56:47 +08:00
Cloud Wu
cd99557368 udp send don't need inc/dec sending ref, see issue #794 2018-01-25 15:55:45 +08:00
Cloud Wu
04f4004934 sproto bugfix 2018-01-23 11:24:44 +08:00
Cloud Wu
bbf46f935a dec sending count asap, this may fix #794 2018-01-19 10:41:35 +08:00
Cloud Wu
b792c7144d clusterd wait for cluster.reload when the name no exist 2018-01-04 14:18:29 +08:00
Cloud Wu
5d5112257c use 32bit sending, see issue #777 2017-12-06 20:58:36 +08:00
云风
747e6c7848 Merge pull request #775 from cloudwu/tcpwriteorder
Tcpwriteorder, Fix #744
2017-12-04 11:45:49 +08:00
Cloud Wu
45d848c910 bugfix: inc sending ref, See issue #774 2017-12-04 10:55:47 +08:00
Cloud Wu
f12fa7e8f9 May fix issue #774 2017-12-03 00:26:12 +08:00
Cloud Wu
568a435e9e fix issue #773 2017-11-30 16:13:58 +08:00
Cloud Wu
c46f2633be fork may not execute when the service exit, and __gc would send message release without monitor 2017-11-16 10:47:18 +08:00
Cloud Wu
b51e28a3e0 set nodelay, see issue #763 2017-11-14 18:01:08 +08:00
云风
badfab8561 Merge pull request #761 from cloudwu/mongolimit
positive limit support
2017-11-07 10:17:04 +08:00
Cloud Wu
d5c0d687b3 positive limit support 2017-11-06 20:51:33 +08:00
Cloud Wu
57f08426ee concurrent cluster link, fix issue #757 2017-11-02 22:31:54 +08:00
126 changed files with 2229 additions and 720 deletions

View File

@@ -1,5 +1,5 @@
/*
** $Id: lapi.c,v 2.259 2016/02/29 14:27:14 roberto Exp $
** $Id: lapi.c,v 2.259.1.2 2017/12/06 18:35:12 roberto Exp $
** Lua API
** See Copyright Notice in lua.h
*/
@@ -534,6 +534,7 @@ LUA_API void lua_pushcclosure (lua_State *L, lua_CFunction fn, int n) {
lua_lock(L);
if (n == 0) {
setfvalue(L->top, fn);
api_incr_top(L);
}
else {
CClosure *cl;
@@ -547,9 +548,9 @@ LUA_API void lua_pushcclosure (lua_State *L, lua_CFunction fn, int n) {
/* does not need barrier because closure is white */
}
setclCvalue(L, L->top, cl);
api_incr_top(L);
luaC_checkGC(L);
}
api_incr_top(L);
luaC_checkGC(L);
lua_unlock(L);
}

View File

@@ -1,5 +1,5 @@
/*
** $Id: lapi.h,v 2.9 2015/03/06 19:49:50 roberto Exp $
** $Id: lapi.h,v 2.9.1.1 2017/04/19 17:20:42 roberto Exp $
** Auxiliary functions from Lua API
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lauxlib.c,v 1.289 2016/12/20 18:37:00 roberto Exp $
** $Id: lauxlib.c,v 1.289.1.1 2017/04/19 17:20:42 roberto Exp $
** Auxiliary functions for building Lua libraries
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lauxlib.h,v 1.131 2016/12/06 14:54:31 roberto Exp $
** $Id: lauxlib.h,v 1.131.1.1 2017/04/19 17:20:42 roberto Exp $
** Auxiliary functions for building Lua libraries
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lbaselib.c,v 1.314 2016/09/05 19:06:34 roberto Exp $
** $Id: lbaselib.c,v 1.314.1.1 2017/04/19 17:39:34 roberto Exp $
** Basic library
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lbitlib.c,v 1.30 2015/11/11 19:08:09 roberto Exp $
** $Id: lbitlib.c,v 1.30.1.1 2017/04/19 17:20:42 roberto Exp $
** Standard library for bitwise operations
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lcode.c,v 2.112 2016/12/22 13:08:50 roberto Exp $
** $Id: lcode.c,v 2.112.1.1 2017/04/19 17:20:42 roberto Exp $
** Code generator for Lua
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lcode.h,v 1.64 2016/01/05 16:22:37 roberto Exp $
** $Id: lcode.h,v 1.64.1.1 2017/04/19 17:20:42 roberto Exp $
** Code generator for Lua
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lcorolib.c,v 1.10 2016/04/11 19:19:55 roberto Exp $
** $Id: lcorolib.c,v 1.10.1.1 2017/04/19 17:20:42 roberto Exp $
** Coroutine Library
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lctype.c,v 1.12 2014/11/02 19:19:04 roberto Exp $
** $Id: lctype.c,v 1.12.1.1 2017/04/19 17:20:42 roberto Exp $
** 'ctype' functions for Lua
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lctype.h,v 1.12 2011/07/15 12:50:29 roberto Exp $
** $Id: lctype.h,v 1.12.1.1 2013/04/12 18:48:47 roberto Exp $
** 'ctype' functions for Lua
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: ldblib.c,v 1.151 2015/11/23 11:29:43 roberto Exp $
** $Id: ldblib.c,v 1.151.1.1 2017/04/19 17:20:42 roberto Exp $
** Interface from Lua to its debug API
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: ldebug.c,v 2.121 2016/10/19 12:32:10 roberto Exp $
** $Id: ldebug.c,v 2.121.1.2 2017/07/10 17:21:50 roberto Exp $
** Debug Interface
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: ldebug.h,v 2.14 2015/05/22 17:45:56 roberto Exp $
** $Id: ldebug.h,v 2.14.1.1 2017/04/19 17:20:42 roberto Exp $
** Auxiliary functions from Debug Interface module
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: ldo.c,v 2.157 2016/12/13 15:52:21 roberto Exp $
** $Id: ldo.c,v 2.157.1.1 2017/04/19 17:20:42 roberto Exp $
** Stack and Call structure of Lua
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: ldo.h,v 2.29 2015/12/21 13:02:14 roberto Exp $
** $Id: ldo.h,v 2.29.1.1 2017/04/19 17:20:42 roberto Exp $
** Stack and Call structure of Lua
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: ldump.c,v 2.37 2015/10/08 15:53:49 roberto Exp $
** $Id: ldump.c,v 2.37.1.1 2017/04/19 17:20:42 roberto Exp $
** save precompiled Lua chunks
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lfunc.c,v 2.45 2014/11/02 19:19:04 roberto Exp $
** $Id: lfunc.c,v 2.45.1.1 2017/04/19 17:39:34 roberto Exp $
** Auxiliary functions to manipulate prototypes and closures
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lfunc.h,v 2.15 2015/01/13 15:49:11 roberto Exp $
** $Id: lfunc.h,v 2.15.1.1 2017/04/19 17:39:34 roberto Exp $
** Auxiliary functions to manipulate prototypes and closures
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lgc.c,v 2.215 2016/12/22 13:08:50 roberto Exp $
** $Id: lgc.c,v 2.215.1.2 2017/08/31 16:15:27 roberto Exp $
** Garbage Collector
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lgc.h,v 2.91 2015/12/21 13:02:14 roberto Exp $
** $Id: lgc.h,v 2.91.1.1 2017/04/19 17:39:34 roberto Exp $
** Garbage Collector
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: linit.c,v 1.39 2016/12/04 20:17:24 roberto Exp $
** $Id: linit.c,v 1.39.1.1 2017/04/19 17:20:42 roberto Exp $
** Initialization of libraries for lua.c and other clients
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: liolib.c,v 2.151 2016/12/20 18:37:00 roberto Exp $
** $Id: liolib.c,v 2.151.1.1 2017/04/19 17:29:57 roberto Exp $
** Standard I/O (and system) library
** See Copyright Notice in lua.h
*/
@@ -206,11 +206,16 @@ static int aux_close (lua_State *L) {
}
static int f_close (lua_State *L) {
tofile(L); /* make sure argument is an open stream */
return aux_close(L);
}
static int io_close (lua_State *L) {
if (lua_isnone(L, 1)) /* no argument? */
lua_getfield(L, LUA_REGISTRYINDEX, IO_OUTPUT); /* use standard output */
tofile(L); /* make sure argument is an open stream */
return aux_close(L);
return f_close(L);
}
@@ -712,7 +717,7 @@ static const luaL_Reg iolib[] = {
** methods for file handles
*/
static const luaL_Reg flib[] = {
{"close", io_close},
{"close", f_close},
{"flush", f_flush},
{"lines", f_lines},
{"read", f_read},

View File

@@ -1,5 +1,5 @@
/*
** $Id: llex.c,v 2.96 2016/05/02 14:02:12 roberto Exp $
** $Id: llex.c,v 2.96.1.1 2017/04/19 17:20:42 roberto Exp $
** Lexical Analyzer
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: llex.h,v 1.79 2016/05/02 14:02:12 roberto Exp $
** $Id: llex.h,v 1.79.1.1 2017/04/19 17:20:42 roberto Exp $
** Lexical Analyzer
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: llimits.h,v 1.141 2015/11/19 19:16:22 roberto Exp $
** $Id: llimits.h,v 1.141.1.1 2017/04/19 17:20:42 roberto Exp $
** Limits, basic types, and some other 'installation-dependent' definitions
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lmathlib.c,v 1.119 2016/12/22 13:08:50 roberto Exp $
** $Id: lmathlib.c,v 1.119.1.1 2017/04/19 17:20:42 roberto Exp $
** Standard mathematical library
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lmem.c,v 1.91 2015/03/06 19:45:54 roberto Exp $
** $Id: lmem.c,v 1.91.1.1 2017/04/19 17:20:42 roberto Exp $
** Interface to Memory Manager
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lmem.h,v 1.43 2014/12/19 17:26:14 roberto Exp $
** $Id: lmem.h,v 1.43.1.1 2017/04/19 17:20:42 roberto Exp $
** Interface to Memory Manager
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: loadlib.c,v 1.130 2017/01/12 17:14:26 roberto Exp $
** $Id: loadlib.c,v 1.130.1.1 2017/04/19 17:20:42 roberto Exp $
** Dynamic library loader for Lua
** See Copyright Notice in lua.h
**

View File

@@ -1,5 +1,5 @@
/*
** $Id: lobject.c,v 2.113 2016/12/22 13:08:50 roberto Exp $
** $Id: lobject.c,v 2.113.1.1 2017/04/19 17:29:57 roberto Exp $
** Some generic functions over Lua objects
** See Copyright Notice in lua.h
*/
@@ -435,7 +435,8 @@ const char *luaO_pushvfstring (lua_State *L, const char *fmt, va_list argp) {
}
case 'p': { /* a pointer */
char buff[4*sizeof(void *) + 8]; /* should be enough space for a '%p' */
int l = l_sprintf(buff, sizeof(buff), "%p", va_arg(argp, void *));
void *p = va_arg(argp, void *);
int l = lua_pointer2str(buff, sizeof(buff), p);
pushstr(L, buff, l);
break;
}

View File

@@ -1,5 +1,5 @@
/*
** $Id: lobject.h,v 2.117 2016/08/01 19:51:24 roberto Exp $
** $Id: lobject.h,v 2.117.1.1 2017/04/19 17:39:34 roberto Exp $
** Type definitions for Lua objects
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lopcodes.c,v 1.55 2015/01/05 13:48:33 roberto Exp $
** $Id: lopcodes.c,v 1.55.1.1 2017/04/19 17:20:42 roberto Exp $
** Opcodes for Lua virtual machine
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lopcodes.h,v 1.149 2016/07/19 17:12:21 roberto Exp $
** $Id: lopcodes.h,v 1.149.1.1 2017/04/19 17:20:42 roberto Exp $
** Opcodes for Lua virtual machine
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: loslib.c,v 1.65 2016/07/18 17:58:58 roberto Exp $
** $Id: loslib.c,v 1.65.1.1 2017/04/19 17:29:57 roberto Exp $
** Standard Operating System library
** See Copyright Notice in lua.h
*/
@@ -293,7 +293,8 @@ static int os_date (lua_State *L) {
else
stm = l_localtime(&t, &tmr);
if (stm == NULL) /* invalid date? */
luaL_error(L, "time result cannot be represented in this installation");
return luaL_error(L,
"time result cannot be represented in this installation");
if (strcmp(s, "*t") == 0) {
lua_createtable(L, 0, 9); /* 9 = number of fields */
setallfields(L, stm);
@@ -340,7 +341,8 @@ static int os_time (lua_State *L) {
setallfields(L, &ts); /* update fields with normalized values */
}
if (t != (time_t)(l_timet)t || t == (time_t)(-1))
luaL_error(L, "time result cannot be represented in this installation");
return luaL_error(L,
"time result cannot be represented in this installation");
l_pushtime(L, t);
return 1;
}

View File

@@ -1,5 +1,5 @@
/*
** $Id: lparser.c,v 2.155 2016/08/01 19:51:24 roberto Exp $
** $Id: lparser.c,v 2.155.1.2 2017/04/29 18:11:40 roberto Exp $
** Lua Parser
** See Copyright Notice in lua.h
*/
@@ -1395,7 +1395,7 @@ static void test_then_block (LexState *ls, int *escapelist) {
luaK_goiffalse(ls->fs, &v); /* will jump to label if condition is true */
enterblock(fs, &bl, 0); /* must enter block before 'goto' */
gotostat(ls, v.t); /* handle goto/break */
while (testnext(ls, ';')) {} /* skip semicolons */
while (testnext(ls, ';')) {} /* skip colons */
if (block_follow(ls, 0)) { /* 'goto' is the entire block? */
leaveblock(fs);
return; /* and that is it */

View File

@@ -1,5 +1,5 @@
/*
** $Id: lparser.h,v 1.76 2015/12/30 18:16:13 roberto Exp $
** $Id: lparser.h,v 1.76.1.1 2017/04/19 17:20:42 roberto Exp $
** Lua Parser
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lprefix.h,v 1.2 2014/12/29 16:54:13 roberto Exp $
** $Id: lprefix.h,v 1.2.1.1 2017/04/19 17:20:42 roberto Exp $
** Definitions for Lua code that must come before any other header file
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lstate.c,v 2.133 2015/11/13 12:16:51 roberto Exp $
** $Id: lstate.c,v 2.133.1.1 2017/04/19 17:39:34 roberto Exp $
** Global State
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lstate.h,v 2.133 2016/12/22 13:08:50 roberto Exp $
** $Id: lstate.h,v 2.133.1.1 2017/04/19 17:39:34 roberto Exp $
** Global State
** See Copyright Notice in lua.h
*/
@@ -26,6 +26,24 @@
** 'tobefnz': all objects ready to be finalized;
** 'fixedgc': all objects that are not to be collected (currently
** only small strings, such as reserved words).
**
** Moreover, there is another set of lists that control gray objects.
** These lists are linked by fields 'gclist'. (All objects that
** can become gray have such a field. The field is not the same
** in all objects, but it always has this name.) Any gray object
** must belong to one of these lists, and all objects in these lists
** must be gray:
**
** 'gray': regular gray objects, still waiting to be visited.
** 'grayagain': objects that must be revisited at the atomic phase.
** That includes
** - black objects got in a write barrier;
** - all kinds of weak tables during propagation phase;
** - all threads.
** 'weak': tables with weak values to be cleared;
** 'ephemeron': ephemeron tables with white->white entries;
** 'allweak': tables with weak keys and/or weak values to be cleared.
** The last three lists are used only during the atomic phase.
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lstring.c,v 2.56 2015/11/23 11:32:51 roberto Exp $
** $Id: lstring.c,v 2.56.1.1 2017/04/19 17:20:42 roberto Exp $
** String table (keeps all strings handled by Lua)
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lstring.h,v 1.61 2015/11/03 15:36:01 roberto Exp $
** $Id: lstring.h,v 1.61.1.1 2017/04/19 17:20:42 roberto Exp $
** String table (keep all strings handled by Lua)
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lstrlib.c,v 1.254 2016/12/22 13:08:50 roberto Exp $
** $Id: lstrlib.c,v 1.254.1.1 2017/04/19 17:29:57 roberto Exp $
** Standard library for string operations and pattern-matching
** See Copyright Notice in lua.h
*/
@@ -879,7 +879,7 @@ static int lua_number2strx (lua_State *L, char *buff, int sz,
buff[i] = toupper(uchar(buff[i]));
}
else if (fmt[SIZELENMOD] != 'a')
luaL_error(L, "modifiers for format '%%a'/'%%A' not implemented");
return luaL_error(L, "modifiers for format '%%a'/'%%A' not implemented");
return n;
}
@@ -1199,8 +1199,8 @@ static int getnum (const char **fmt, int df) {
static int getnumlimit (Header *h, const char **fmt, int df) {
int sz = getnum(fmt, df);
if (sz > MAXINTSIZE || sz <= 0)
luaL_error(h->L, "integral size (%d) out of limits [1,%d]",
sz, MAXINTSIZE);
return luaL_error(h->L, "integral size (%d) out of limits [1,%d]",
sz, MAXINTSIZE);
return sz;
}

View File

@@ -1,5 +1,5 @@
/*
** $Id: ltable.c,v 2.118 2016/11/07 12:38:35 roberto Exp $
** $Id: ltable.c,v 2.118.1.4 2018/06/08 16:22:51 roberto Exp $
** Lua tables (hash)
** See Copyright Notice in lua.h
*/
@@ -223,7 +223,9 @@ static unsigned int computesizes (unsigned int nums[], unsigned int *pna) {
unsigned int na = 0; /* number of elements to go to array part */
unsigned int optimal = 0; /* optimal size for array part */
/* loop while keys can fill more than half of total size */
for (i = 0, twotoi = 1; *pna > twotoi / 2; i++, twotoi *= 2) {
for (i = 0, twotoi = 1;
twotoi > 0 && *pna > twotoi / 2;
i++, twotoi *= 2) {
if (nums[i] > 0) {
a += nums[i];
if (a > twotoi/2) { /* more than half elements present? */
@@ -330,17 +332,34 @@ static void setnodevector (lua_State *L, Table *t, unsigned int size) {
}
typedef struct {
Table *t;
unsigned int nhsize;
} AuxsetnodeT;
static void auxsetnode (lua_State *L, void *ud) {
AuxsetnodeT *asn = cast(AuxsetnodeT *, ud);
setnodevector(L, asn->t, asn->nhsize);
}
void luaH_resize (lua_State *L, Table *t, unsigned int nasize,
unsigned int nhsize) {
unsigned int i;
int j;
AuxsetnodeT asn;
unsigned int oldasize = t->sizearray;
int oldhsize = allocsizenode(t);
Node *nold = t->node; /* save old hash ... */
if (nasize > oldasize) /* array part must grow? */
setarrayvector(L, t, nasize);
/* create new hash part with appropriate size */
setnodevector(L, t, nhsize);
asn.t = t; asn.nhsize = nhsize;
if (luaD_rawrunprotected(L, auxsetnode, &asn) != LUA_OK) { /* mem. error? */
setarrayvector(L, t, oldasize); /* array back to its original size */
luaD_throw(L, LUA_ERRMEM); /* rethrow memory error */
}
if (nasize < oldasize) { /* array part must shrink? */
t->sizearray = nasize;
/* re-insert elements from vanishing slice */
@@ -610,13 +629,13 @@ void luaH_setint (lua_State *L, Table *t, lua_Integer key, TValue *value) {
}
static int unbound_search (Table *t, unsigned int j) {
unsigned int i = j; /* i is zero or a present index */
static lua_Unsigned unbound_search (Table *t, lua_Unsigned j) {
lua_Unsigned i = j; /* i is zero or a present index */
j++;
/* find 'i' and 'j' such that i is present and j is not */
while (!ttisnil(luaH_getint(t, j))) {
i = j;
if (j > cast(unsigned int, MAX_INT)/2) { /* overflow? */
if (j > l_castS2U(LUA_MAXINTEGER) / 2) { /* overflow? */
/* table was built with bad purposes: resort to linear search */
i = 1;
while (!ttisnil(luaH_getint(t, i))) i++;
@@ -626,7 +645,7 @@ static int unbound_search (Table *t, unsigned int j) {
}
/* now do a binary search between them */
while (j - i > 1) {
unsigned int m = (i+j)/2;
lua_Unsigned m = (i+j)/2;
if (ttisnil(luaH_getint(t, m))) j = m;
else i = m;
}
@@ -638,7 +657,7 @@ static int unbound_search (Table *t, unsigned int j) {
** Try to find a boundary in table 't'. A 'boundary' is an integer index
** such that t[i] is non-nil and t[i+1] is nil (and 0 if t[1] is nil).
*/
int luaH_getn (Table *t) {
lua_Unsigned luaH_getn (Table *t) {
unsigned int j = t->sizearray;
if (j > 0 && ttisnil(&t->array[j - 1])) {
/* there is a boundary in the array part: (binary) search for it */

View File

@@ -1,5 +1,5 @@
/*
** $Id: ltable.h,v 2.23 2016/12/22 13:08:50 roberto Exp $
** $Id: ltable.h,v 2.23.1.2 2018/05/24 19:39:05 roberto Exp $
** Lua tables (hash)
** See Copyright Notice in lua.h
*/
@@ -54,7 +54,7 @@ LUAI_FUNC void luaH_resize (lua_State *L, Table *t, unsigned int nasize,
LUAI_FUNC void luaH_resizearray (lua_State *L, Table *t, unsigned int nasize);
LUAI_FUNC void luaH_free (lua_State *L, Table *t);
LUAI_FUNC int luaH_next (lua_State *L, Table *t, StkId key);
LUAI_FUNC int luaH_getn (Table *t);
LUAI_FUNC lua_Unsigned luaH_getn (Table *t);
#if defined(LUA_DEBUG)

View File

@@ -1,5 +1,5 @@
/*
** $Id: ltablib.c,v 1.93 2016/02/25 19:41:54 roberto Exp $
** $Id: ltablib.c,v 1.93.1.1 2017/04/19 17:20:42 roberto Exp $
** Library for Table Manipulation
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: ltm.c,v 2.38 2016/12/22 13:08:50 roberto Exp $
** $Id: ltm.c,v 2.38.1.1 2017/04/19 17:39:34 roberto Exp $
** Tag methods
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: ltm.h,v 2.22 2016/02/26 19:20:15 roberto Exp $
** $Id: ltm.h,v 2.22.1.1 2017/04/19 17:20:42 roberto Exp $
** Tag methods
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lua.c,v 1.230 2017/01/12 17:14:26 roberto Exp $
** $Id: lua.c,v 1.230.1.1 2017/04/19 17:29:57 roberto Exp $
** Lua stand-alone interpreter
** See Copyright Notice in lua.h
*/
@@ -139,7 +139,7 @@ static void print_usage (const char *badoption) {
"Available options are:\n"
" -e stat execute string 'stat'\n"
" -i enter interactive mode after executing 'script'\n"
" -l name require library 'name'\n"
" -l name require library 'name' into global 'name'\n"
" -v show version information\n"
" -E ignore environment variables\n"
" -- stop handling options\n"

View File

@@ -1,5 +1,5 @@
/*
** $Id: lua.h,v 1.332 2016/12/22 15:51:20 roberto Exp $
** $Id: lua.h,v 1.332.1.2 2018/06/13 16:58:17 roberto Exp $
** Lua - A Scripting Language
** Lua.org, PUC-Rio, Brazil (http://www.lua.org)
** See Copyright Notice at the end of this file
@@ -19,11 +19,11 @@
#define LUA_VERSION_MAJOR "5"
#define LUA_VERSION_MINOR "3"
#define LUA_VERSION_NUM 503
#define LUA_VERSION_RELEASE "4"
#define LUA_VERSION_RELEASE "5"
#define LUA_VERSION "Lua " LUA_VERSION_MAJOR "." LUA_VERSION_MINOR
#define LUA_RELEASE LUA_VERSION "." LUA_VERSION_RELEASE
#define LUA_COPYRIGHT LUA_RELEASE " Copyright (C) 1994-2017 Lua.org, PUC-Rio"
#define LUA_COPYRIGHT LUA_RELEASE " Copyright (C) 1994-2018 Lua.org, PUC-Rio"
#define LUA_AUTHORS "R. Ierusalimschy, L. H. de Figueiredo, W. Celes"
@@ -467,7 +467,7 @@ LUA_API void (lua_checksig_)(lua_State *L);
#define lua_checksig(L) if (skynet_sig_L) { lua_checksig_(L); }
/******************************************************************************
* Copyright (C) 1994-2017 Lua.org, PUC-Rio.
* Copyright (C) 1994-2018 Lua.org, PUC-Rio.
*
* Permission is hereby granted, free of charge, to any person obtaining
* a copy of this software and associated documentation files (the

View File

@@ -1,5 +1,5 @@
/*
** $Id: luac.c,v 1.75 2015/03/12 01:58:27 lhf Exp $
** $Id: luac.c,v 1.76 2018/06/19 01:32:02 lhf Exp $
** Lua compiler (saves bytecodes to files; also lists bytecodes)
** See Copyright Notice in lua.h
*/
@@ -208,7 +208,7 @@ int main(int argc, char* argv[])
}
/*
** $Id: luac.c,v 1.75 2015/03/12 01:58:27 lhf Exp $
** $Id: luac.c,v 1.76 2018/06/19 01:32:02 lhf Exp $
** print bytecodes
** See Copyright Notice in lua.h
*/
@@ -350,6 +350,7 @@ static void PrintCode(const Proto* f)
case OP_ADD:
case OP_SUB:
case OP_MUL:
case OP_MOD:
case OP_POW:
case OP_DIV:
case OP_IDIV:

View File

@@ -1,5 +1,5 @@
/*
** $Id: luaconf.h,v 1.259 2016/12/22 13:08:50 roberto Exp $
** $Id: luaconf.h,v 1.259.1.1 2017/04/19 17:29:57 roberto Exp $
** Configuration file for Lua
** See Copyright Notice in lua.h
*/
@@ -620,6 +620,13 @@
#endif
/*
@@ lua_pointer2str converts a pointer to a readable string in a
** non-specified way.
*/
#define lua_pointer2str(buff,sz,p) l_sprintf(buff,sz,"%p",p)
/*
@@ lua_number2strx converts a float to an hexadecimal numeric string.
** In C99, 'sprintf' (with format specifiers '%a'/'%A') does that.

View File

@@ -1,5 +1,5 @@
/*
** $Id: lualib.h,v 1.45 2017/01/12 17:14:26 roberto Exp $
** $Id: lualib.h,v 1.45.1.1 2017/04/19 17:20:42 roberto Exp $
** Lua standard libraries
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lundump.c,v 2.44 2015/11/02 16:09:30 roberto Exp $
** $Id: lundump.c,v 2.44.1.1 2017/04/19 17:20:42 roberto Exp $
** load precompiled Lua chunks
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lundump.h,v 1.45 2015/09/08 15:41:05 roberto Exp $
** $Id: lundump.h,v 1.45.1.1 2017/04/19 17:20:42 roberto Exp $
** load precompiled Lua chunks
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lutf8lib.c,v 1.16 2016/12/22 13:08:50 roberto Exp $
** $Id: lutf8lib.c,v 1.16.1.1 2017/04/19 17:29:57 roberto Exp $
** Standard library for UTF-8 manipulation
** See Copyright Notice in lua.h
*/
@@ -171,7 +171,7 @@ static int byteoffset (lua_State *L) {
}
else {
if (iscont(s + posi))
luaL_error(L, "initial position is a continuation byte");
return luaL_error(L, "initial position is a continuation byte");
if (n < 0) {
while (n < 0 && posi > 0) { /* move back */
do { /* find beginning of previous character */

View File

@@ -1,5 +1,5 @@
/*
** $Id: lvm.c,v 2.268 2016/02/05 19:59:14 roberto Exp $
** $Id: lvm.c,v 2.268.1.1 2017/04/19 17:39:34 roberto Exp $
** Lua virtual machine
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lvm.h,v 2.41 2016/12/22 13:08:50 roberto Exp $
** $Id: lvm.h,v 2.41.1.1 2017/04/19 17:20:42 roberto Exp $
** Lua virtual machine
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lzio.c,v 1.37 2015/09/08 15:41:05 roberto Exp $
** $Id: lzio.c,v 1.37.1.1 2017/04/19 17:20:42 roberto Exp $
** Buffered streams
** See Copyright Notice in lua.h
*/

View File

@@ -1,5 +1,5 @@
/*
** $Id: lzio.h,v 1.31 2015/09/08 15:41:05 roberto Exp $
** $Id: lzio.h,v 1.31.1.1 2017/04/19 17:20:42 roberto Exp $
** Buffered streams
** See Copyright Notice in lua.h
*/

View File

@@ -1,3 +1,22 @@
v1.2.0 (2018-11-6)
-----------
* Improve cluster support
* Improve mongodb driver
* Improve redis driver
* Improve socket concurrent write
* Improve socket channel
* Improve service gate
* Improve udp support
* Add skynet.ignoreret
* Add skynet.trace
* Add skynet.context
* Improve skynet.wait/wakeup
* Add socket.netstat
* Add socketchannel.overload
* Fix memory leak for dead service
* lua update to 5.3.5
* jemalloc update to 5.1.0
v1.1.0 (2017-10-31)
-----------
* add socket.disconnected()

View File

@@ -1,4 +1,4 @@
## Skynet
## ![skynet logo](https://github.com/cloudwu/skynet/wiki/image/skynet_metro.jpg)
Skynet is a lightweight online game framework, and it can be used in many other fields.
@@ -32,11 +32,11 @@ Run these in different consoles:
## About Lua version
Skynet now uses a modified version of lua 5.3.4 ( https://github.com/ejoy/lua/tree/skynet ) for multiple lua states.
Skynet now uses a modified version of lua 5.3.5 ( https://github.com/ejoy/lua/tree/skynet ) for multiple lua states.
You can also use official Lua versions, just edit the Makefile by yourself.
## How To Use (Sorry, Only in Chinese now)
## How To Use (Sorry, currently only available in Chinese)
* Read Wiki for documents https://github.com/cloudwu/skynet/wiki
* The FAQ in wiki https://github.com/cloudwu/skynet/wiki/FAQ

View File

@@ -1,5 +1,4 @@
local skynet = require "skynet"
local netpack = require "skynet.netpack"
local socket = require "skynet.socket"
local sproto = require "sproto"
local sprotoloader = require "sprotoloader"
@@ -50,7 +49,10 @@ skynet.register_protocol {
unpack = function (msg, sz)
return host:dispatch(msg, sz)
end,
dispatch = function (_, _, type, ...)
dispatch = function (fd, _, type, ...)
assert(fd == client_fd) -- You can use fd to reply message
skynet.ignoreret() -- session is fd, don't call skynet.ret
skynet.trace()
if type == "REQUEST" then
local ok, result = pcall(request, ...)
if ok then
@@ -92,6 +94,7 @@ end
skynet.start(function()
skynet.dispatch("lua", function(_,_, command, ...)
skynet.trace()
local f = CMD[command]
skynet.ret(skynet.pack(f(...)))
end)

View File

@@ -2,22 +2,30 @@ local skynet = require "skynet"
local cluster = require "skynet.cluster"
skynet.start(function()
-- query name "sdb" of cluster db.
local sdb = cluster.query("db", "sdb")
print("db.sbd=",sdb)
local proxy = cluster.proxy("db", sdb)
local proxy = cluster.proxy "db@sdb" -- cluster.proxy("db", "@sdb")
local largekey = string.rep("X", 128*1024)
local largevalue = string.rep("R", 100 * 1024)
print(skynet.call(proxy, "lua", "SET", largekey, largevalue))
skynet.call(proxy, "lua", "SET", largekey, largevalue)
local v = skynet.call(proxy, "lua", "GET", largekey)
assert(largevalue == v)
skynet.send(proxy, "lua", "PING", "proxy")
print(cluster.call("db", sdb, "GET", "a"))
print(cluster.call("db2", sdb, "GET", "b"))
cluster.send("db2", sdb, "PING", "db2:longstring" .. largevalue)
skynet.fork(function()
skynet.trace("cluster")
print(cluster.call("db", "@sdb", "GET", "a"))
print(cluster.call("db2", "@sdb", "GET", "b"))
cluster.send("db2", "@sdb", "PING", "db2:longstring" .. largevalue)
end)
-- test snax service
local pingserver = cluster.snax("db", "pingserver")
skynet.timeout(300,function()
cluster.reload {
db = false, -- db is down
db3 = "127.0.0.1:2529"
}
print(pcall(cluster.call, "db", "@sdb", "GET", "a")) -- db is down
end)
cluster.reload { __nowaiting = false }
local pingserver = cluster.snax("db3", "pingserver")
print(pingserver.req.ping "hello")
end)

View File

@@ -1,2 +1,4 @@
__nowaiting = true -- If you turn this flag off, cluster.call would block when node name is absent
db = "127.0.0.1:2528"
db2 = "127.0.0.1:2529"

11
examples/config.mongodb Normal file
View File

@@ -0,0 +1,11 @@
root = "./"
thread = 8
logger = nil
harbor = 0
start = "main_mongodb" -- main script
bootstrap = "snlua bootstrap" -- The service for bootstrap
luaservice = root.."service/?.lua;"..root.."test/?.lua;"..root.."examples/?.lua"
lualoader = "lualib/loader.lua"
snax = root.."examples/?.lua;"..root.."test/?.lua"
cpath = root.."cservice/?.so"
-- daemon = "./skynet.pid"

12
examples/main_mongodb.lua Normal file
View File

@@ -0,0 +1,12 @@
local skynet = require "skynet"
skynet.start(function()
print("Main Server start")
local console = skynet.newservice(
"testmongodb", "127.0.0.1", 27017, "testdb", "test", "test"
)
print("Main Server exit")
skynet.exit()
end)

View File

@@ -33,5 +33,6 @@ skynet.start(function()
error(string.format("Unknown command %s", tostring(cmd)))
end
end)
-- skynet.traceproto("lua", false) -- true off tracelog
skynet.register "SIMPLEDB"
end)

View File

@@ -1,6 +1,7 @@
local skynet = require "skynet"
require "skynet.manager"
-- register protocol text before skynet.start would be better.
skynet.register_protocol {
name = "text",
id = skynet.PTYPE_TEXT,
@@ -21,5 +22,4 @@ skynet.register_protocol {
}
skynet.start(function()
skynet.register ".logger"
end)

View File

@@ -4,6 +4,7 @@
#include <lauxlib.h>
#include <string.h>
#include <assert.h>
#include <unistd.h>
#include "skynet.h"
@@ -73,6 +74,11 @@ fill_header(lua_State *L, uint8_t *buf, int sz) {
BYTE 2/3 ; 2:multipart, 3:multipart end
DWORD SESSION
PADDING msgpart(sz)
trace
WORD stringsz + 1
BYTE 4
STRING tag
*/
static int
packreq_number(lua_State *L, int session, void * msg, uint32_t sz, int is_push) {
@@ -177,14 +183,14 @@ packrequest(lua_State *L, int is_push) {
} else {
multipak = packreq_string(L, session, msg, sz, is_push);
}
int current_session = session;
if (++session < 0) {
session = 1;
uint32_t new_session = (uint32_t)session + 1;
if (new_session > INT32_MAX) {
new_session = 1;
}
lua_pushinteger(L, session);
lua_pushinteger(L, new_session);
if (multipak) {
lua_createtable(L, multipak, 0);
packreq_multi(L, current_session, msg, sz);
packreq_multi(L, session, msg, sz);
skynet_free(msg);
return 3;
} else {
@@ -203,13 +209,30 @@ lpackpush(lua_State *L) {
return packrequest(L, 1);
}
static int
lpacktrace(lua_State *L) {
size_t sz;
const char * tag = luaL_checklstring(L, 1, &sz);
if (sz > 0x8000) {
return luaL_error(L, "trace tag is too long : %d", (int) sz);
}
uint8_t buf[TEMP_LENGTH];
buf[2] = 4;
fill_header(L, buf, sz+1);
memcpy(buf+3, tag, sz);
lua_pushlstring(L, (const char *)buf, sz+3);
return 1;
}
/*
string packed message
return
uint32_t or string addr
int session
string msg
lightuserdata msg
int sz
boolean padding
boolean is_push
*/
static inline uint32_t
@@ -217,6 +240,14 @@ unpack_uint32(const uint8_t * buf) {
return buf[0] | buf[1]<<8 | buf[2]<<16 | buf[3]<<24;
}
static void
return_buffer(lua_State *L, const char * buffer, int sz) {
void * ptr = skynet_malloc(sz);
memcpy(ptr, buffer, sz);
lua_pushlightuserdata(L, ptr);
lua_pushinteger(L, sz);
}
static int
unpackreq_number(lua_State *L, const uint8_t * buf, int sz) {
if (sz < 9) {
@@ -226,14 +257,15 @@ unpackreq_number(lua_State *L, const uint8_t * buf, int sz) {
uint32_t session = unpack_uint32(buf+5);
lua_pushinteger(L, address);
lua_pushinteger(L, session);
lua_pushlstring(L, (const char *)buf+9, sz-9);
return_buffer(L, (const char *)buf+9, sz-9);
if (session == 0) {
lua_pushnil(L);
lua_pushboolean(L,1); // is_push, no reponse
return 5;
return 6;
}
return 3;
return 4;
}
static int
@@ -246,11 +278,12 @@ unpackmreq_number(lua_State *L, const uint8_t * buf, int sz, int is_push) {
uint32_t size = unpack_uint32(buf+9);
lua_pushinteger(L, address);
lua_pushinteger(L, session);
lua_pushnil(L);
lua_pushinteger(L, size);
lua_pushboolean(L, 1); // padding multi part
lua_pushboolean(L, is_push);
return 5;
return 6;
}
static int
@@ -262,10 +295,16 @@ unpackmreq_part(lua_State *L, const uint8_t * buf, int sz) {
uint32_t session = unpack_uint32(buf+1);
lua_pushboolean(L, 0); // no address
lua_pushinteger(L, session);
lua_pushlstring(L, (const char *)buf+5, sz-5);
return_buffer(L, (const char *)buf+5, sz-5);
lua_pushboolean(L, padding);
return 4;
return 5;
}
static int
unpacktrace(lua_State *L, const char * buf, int sz) {
lua_pushlstring(L, buf + 1, sz - 1);
return 1;
}
static int
@@ -280,14 +319,14 @@ unpackreq_string(lua_State *L, const uint8_t * buf, int sz) {
lua_pushlstring(L, (const char *)buf+2, namesz);
uint32_t session = unpack_uint32(buf + namesz + 2);
lua_pushinteger(L, (uint32_t)session);
lua_pushlstring(L, (const char *)buf+2+namesz+4, sz - namesz - 6);
return_buffer(L, (const char *)buf+2+namesz+4, sz - namesz - 6);
if (session == 0) {
lua_pushnil(L);
lua_pushboolean(L,1); // is_push, no reponse
return 5;
return 6;
}
return 3;
return 4;
}
static int
@@ -303,18 +342,26 @@ unpackmreq_string(lua_State *L, const uint8_t * buf, int sz, int is_push) {
uint32_t session = unpack_uint32(buf + namesz + 2);
uint32_t size = unpack_uint32(buf + namesz + 6);
lua_pushinteger(L, session);
lua_pushnil(L);
lua_pushinteger(L, size);
lua_pushboolean(L, 1); // padding multipart
lua_pushboolean(L, is_push);
return 5;
return 6;
}
static int
lunpackrequest(lua_State *L) {
size_t ssz;
const char *msg = luaL_checklstring(L,1,&ssz);
int sz = (int)ssz;
int sz;
const char *msg;
if (lua_type(L, 1) == LUA_TLIGHTUSERDATA) {
msg = (const char *)lua_touserdata(L, 1);
sz = luaL_checkinteger(L, 2);
} else {
size_t ssz;
msg = luaL_checklstring(L,1,&ssz);
sz = (int)ssz;
}
switch (msg[0]) {
case 0:
return unpackreq_number(L, (const uint8_t *)msg, sz);
@@ -325,6 +372,8 @@ lunpackrequest(lua_State *L) {
case 2:
case 3:
return unpackmreq_part(L, (const uint8_t *)msg, sz);
case 4:
return unpacktrace(L, msg, sz);
case '\x80':
return unpackreq_string(L, (const uint8_t *)msg, sz);
case '\x81':
@@ -474,6 +523,32 @@ lunpackresponse(lua_State *L) {
}
}
/*
table
pointer
sz
push (pointer/sz) as string into table, and free pointer
*/
static int
lappend(lua_State *L) {
luaL_checktype(L, 1, LUA_TTABLE);
int n = lua_rawlen(L, 1);
if (lua_isnil(L, 2)) {
lua_settop(L, 3);
lua_seti(L, 1, n + 1);
return 0;
}
void * buffer = lua_touserdata(L, 2);
if (buffer == NULL)
return luaL_error(L, "Need lightuserdata");
int sz = luaL_checkinteger(L, 3);
lua_pushlstring(L, (const char *)buffer, sz);
skynet_free((void *)buffer);
lua_seti(L, 1, n+1);
return 0;
}
static int
lconcat(lua_State *L) {
if (!lua_istable(L,1))
@@ -507,15 +582,41 @@ lconcat(lua_State *L) {
return 2;
}
static int
lisname(lua_State *L) {
const char * name = lua_tostring(L, 1);
if (name && name[0] == '@') {
lua_pushboolean(L, 1);
return 1;
}
return 0;
}
static int
lnodename(lua_State *L) {
pid_t pid = getpid();
char hostname[256];
if (gethostname(hostname, sizeof(hostname))==0) {
lua_pushfstring(L, "%s%d", hostname, (int)pid);
} else {
lua_pushfstring(L, "noname%d", (int)pid);
}
return 1;
}
LUAMOD_API int
luaopen_skynet_cluster_core(lua_State *L) {
luaL_Reg l[] = {
{ "packrequest", lpackrequest },
{ "packpush", lpackpush },
{ "packtrace", lpacktrace },
{ "unpackrequest", lunpackrequest },
{ "packresponse", lpackresponse },
{ "unpackresponse", lunpackresponse },
{ "append", lappend },
{ "concat", lconcat },
{ "isname", lisname },
{ "nodename", lnodename },
{ NULL, NULL },
};
luaL_checkversion(L);

View File

@@ -7,6 +7,7 @@
#include <stdint.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#define SMALL_CHUNK 256
@@ -960,7 +961,7 @@ luaopen_skynet_crypt(lua_State *L) {
if (!init) {
// Don't need call srandom more than once.
init = 1 ;
srandom(time(NULL));
srandom((random() << 8) ^ (time(NULL) << 16) ^ getpid());
}
luaL_Reg l[] = {
{ "hashkey", lhashkey },

View File

@@ -11,6 +11,29 @@
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <inttypes.h>
#include <time.h>
#if defined(__APPLE__)
#include <sys/time.h>
#endif
#include "skynet.h"
// return nsec
static int64_t
get_time() {
#if !defined(__APPLE__) || defined(AVAILABLE_MAC_OS_X_VERSION_10_12_AND_LATER)
struct timespec ti;
clock_gettime(CLOCK_MONOTONIC, &ti);
return (int64_t)1000000000 * ti.tv_sec + ti.tv_nsec;
#else
struct timeval tv;
gettimeofday(&tv, NULL);
return (int64_t)1000000000 * tv.tv_sec + tv.tv_usec * 1000;
#endif
}
struct snlua {
lua_State * L;
@@ -120,6 +143,38 @@ lcommand(lua_State *L) {
return 0;
}
static int
laddresscommand(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
const char * cmd = luaL_checkstring(L,1);
const char * result;
const char * parm = NULL;
if (lua_gettop(L) == 2) {
parm = luaL_checkstring(L,2);
}
result = skynet_command(context, cmd, parm);
if (result && result[0] == ':') {
int i;
uint32_t addr = 0;
for (i=1;result[i];i++) {
int c = result[i];
if (c>='0' && c<='9') {
c = c - '0';
} else if (c>='a' && c<='f') {
c = c - 'a' + 10;
} else if (c>='A' && c<='F') {
c = c - 'A' + 10;
} else {
return 0;
}
addr = addr * 16 + c;
}
lua_pushinteger(L, addr);
return 1;
}
return 0;
}
static int
lintcommand(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
@@ -223,6 +278,11 @@ send_message(lua_State *L, int source, int idx_type) {
luaL_error(L, "invalid param %s", lua_typename(L, lua_type(L,idx_type+2)));
}
if (session < 0) {
if (session == -2) {
// package is too large
lua_pushboolean(L, 0);
return 1;
}
// send to invalid address
// todo: maybe throw an error would be better
return 0;
@@ -346,6 +406,79 @@ lnow(lua_State *L) {
return 1;
}
static int
lhpc(lua_State *L) {
lua_pushinteger(L, get_time());
return 1;
}
#define MAX_LEVEL 3
struct source_info {
const char * source;
int line;
};
/*
string tag
string userstring
thread co (default nil/current L)
integer level (default nil)
*/
static int
ltrace(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
const char * tag = luaL_checkstring(L, 1);
const char * user = luaL_checkstring(L, 2);
if (!lua_isnoneornil(L, 3)) {
lua_State * co = L;
int level;
if (lua_isthread(L, 3)) {
co = lua_tothread (L, 3);
level = luaL_optinteger(L, 4, 1);
} else {
level = luaL_optinteger(L, 3, 1);
}
struct source_info si[MAX_LEVEL];
lua_Debug d;
int index = 0;
do {
if (!lua_getstack(co, level, &d))
break;
lua_getinfo(co, "Sl", &d);
level++;
si[index].source = d.source;
si[index].line = d.currentline;
if (d.currentline >= 0)
++index;
} while (index < MAX_LEVEL);
switch (index) {
case 1:
skynet_error(context, "<TRACE %s> %" PRId64 " %s : %s:%d", tag, get_time(), user, si[0].source, si[0].line);
break;
case 2:
skynet_error(context, "<TRACE %s> %" PRId64 " %s : %s:%d %s:%d", tag, get_time(), user,
si[0].source, si[0].line,
si[1].source, si[1].line
);
break;
case 3:
skynet_error(context, "<TRACE %s> %" PRId64 " %s : %s:%d %s:%d %s:%d", tag, get_time(), user,
si[0].source, si[0].line,
si[1].source, si[1].line,
si[2].source, si[2].line
);
break;
default:
skynet_error(context, "<TRACE %s> %" PRId64 " %s", tag, get_time(), user);
break;
}
return 0;
}
skynet_error(context, "<TRACE %s> %" PRId64 " %s", tag, get_time(), user);
return 0;
}
LUAMOD_API int
luaopen_skynet_core(lua_State *L) {
luaL_checkversion(L);
@@ -356,19 +489,27 @@ luaopen_skynet_core(lua_State *L) {
{ "redirect", lredirect },
{ "command" , lcommand },
{ "intcommand", lintcommand },
{ "addresscommand", laddresscommand },
{ "error", lerror },
{ "tostring", ltostring },
{ "harbor", lharbor },
{ "callback", lcallback },
{ "trace", ltrace },
{ NULL, NULL },
};
// functions without skynet_context
luaL_Reg l2[] = {
{ "tostring", ltostring },
{ "pack", luaseri_pack },
{ "unpack", luaseri_unpack },
{ "packstring", lpackstring },
{ "trash" , ltrash },
{ "callback", lcallback },
{ "now", lnow },
{ "hpc", lhpc }, // getHPCounter
{ NULL, NULL },
};
luaL_newlibtable(L, l);
lua_createtable(L, 0, sizeof(l)/sizeof(l[0]) + sizeof(l2)/sizeof(l2[0]) -2);
lua_getfield(L, LUA_REGISTRYINDEX, "skynet_context");
struct skynet_context *ctx = lua_touserdata(L,-1);
@@ -376,7 +517,10 @@ luaopen_skynet_core(lua_State *L) {
return luaL_error(L, "Init skynet context first");
}
luaL_setfuncs(L,l,1);
luaL_setfuncs(L,l2,0);
return 1;
}

View File

@@ -680,6 +680,72 @@ ludp_address(lua_State *L) {
return 2;
}
static void
getinfo(lua_State *L, struct socket_info *si) {
lua_newtable(L);
lua_pushinteger(L, si->id);
lua_setfield(L, -2, "id");
lua_pushinteger(L, si->opaque);
lua_setfield(L, -2, "address");
switch(si->type) {
case SOCKET_INFO_LISTEN:
lua_pushstring(L, "LISTEN");
lua_setfield(L, -2, "type");
lua_pushinteger(L, si->read);
lua_setfield(L, -2, "accept");
lua_pushinteger(L, si->rtime);
lua_setfield(L, -2, "rtime");
if (si->name[0]) {
lua_pushstring(L, si->name);
lua_setfield(L, -2, "sock");
}
return;
case SOCKET_INFO_TCP:
lua_pushstring(L, "TCP");
break;
case SOCKET_INFO_UDP:
lua_pushstring(L, "UDP");
break;
case SOCKET_INFO_BIND:
lua_pushstring(L, "BIND");
break;
default:
lua_pushstring(L, "UNKNOWN");
lua_setfield(L, -2, "type");
return;
}
lua_setfield(L, -2, "type");
lua_pushinteger(L, si->read);
lua_setfield(L, -2, "read");
lua_pushinteger(L, si->write);
lua_setfield(L, -2, "write");
lua_pushinteger(L, si->wbuffer);
lua_setfield(L, -2, "wbuffer");
lua_pushinteger(L, si->rtime);
lua_setfield(L, -2, "rtime");
lua_pushinteger(L, si->wtime);
lua_setfield(L, -2, "wtime");
if (si->name[0]) {
lua_pushstring(L, si->name);
lua_setfield(L, -2, "peer");
}
}
static int
linfo(lua_State *L) {
lua_newtable(L);
struct socket_info * si = skynet_socket_info();
struct socket_info * temp = si;
int n = 0;
while (temp) {
getinfo(L, temp);
lua_seti(L, -2, ++n);
temp = temp->next;
}
socket_info_release(si);
return 1;
}
LUAMOD_API int
luaopen_skynet_socketdriver(lua_State *L) {
luaL_checkversion(L);
@@ -693,6 +759,7 @@ luaopen_skynet_socketdriver(lua_State *L) {
{ "readline", lreadline },
{ "str2p", lstr2p },
{ "header", lheader },
{ "info", linfo },
{ "unpack", lunpack },
{ NULL, NULL },

View File

@@ -2,6 +2,7 @@
#include <string.h>
#include <stdlib.h>
#include <math.h>
#include "msvcint.h"
#include "lua.h"
@@ -172,7 +173,7 @@ encode(const struct sproto_arg *args) {
// It's decimal.
lua_Number vn = lua_tonumber(L, -1);
// use 64bit integer for 32bit architecture.
v = (int64_t)(vn * args->extra + 0.5);
v = (int64_t)(round(vn * args->extra));
} else {
v = lua_tointegerx(L, -1, &isnum);
if(!isnum) {

View File

@@ -11,6 +11,8 @@
#define SIZEOF_LENGTH 4
#define SIZEOF_HEADER 2
#define SIZEOF_FIELD 2
#define SIZEOF_INT64 ((int)sizeof(uint64_t))
#define SIZEOF_INT32 ((int)sizeof(uint32_t))
struct field {
int tag;
@@ -758,7 +760,7 @@ encode_integer_array(sproto_callback cb, struct sproto_arg *args, uint8_t *buffe
return NULL;
buffer++;
size--;
intlen = sizeof(uint32_t);
intlen = SIZEOF_INT32;
index = 1;
*noarray = 0;
@@ -781,36 +783,38 @@ encode_integer_array(sproto_callback cb, struct sproto_arg *args, uint8_t *buffe
}
return NULL; // sz == SPROTO_CB_ERROR
}
if (size < sizeof(uint64_t))
// notice: sizeof(uint64_t) is size_t (unsigned) , size may be negative. See issue #75
// so use MACRO SIZOF_INT64 instead
if (size < SIZEOF_INT64)
return NULL;
if (sz == sizeof(uint32_t)) {
if (sz == SIZEOF_INT32) {
uint32_t v = u.u32;
buffer[0] = v & 0xff;
buffer[1] = (v >> 8) & 0xff;
buffer[2] = (v >> 16) & 0xff;
buffer[3] = (v >> 24) & 0xff;
if (intlen == sizeof(uint64_t)) {
if (intlen == SIZEOF_INT64) {
uint32_to_uint64(v & 0x80000000, buffer);
}
} else {
uint64_t v;
if (sz != sizeof(uint64_t))
if (sz != SIZEOF_INT64)
return NULL;
if (intlen == sizeof(uint32_t)) {
if (intlen == SIZEOF_INT32) {
int i;
// rearrange
size -= (index-1) * sizeof(uint32_t);
if (size < sizeof(uint64_t))
size -= (index-1) * SIZEOF_INT32;
if (size < SIZEOF_INT64)
return NULL;
buffer += (index-1) * sizeof(uint32_t);
buffer += (index-1) * SIZEOF_INT32;
for (i=index-2;i>=0;i--) {
int negative;
memcpy(header+1+i*sizeof(uint64_t), header+1+i*sizeof(uint32_t), sizeof(uint32_t));
negative = header[1+i*sizeof(uint64_t)+3] & 0x80;
uint32_to_uint64(negative, header+1+i*sizeof(uint64_t));
memcpy(header+1+i*SIZEOF_INT64, header+1+i*SIZEOF_INT32, SIZEOF_INT32);
negative = header[1+i*SIZEOF_INT64+3] & 0x80;
uint32_to_uint64(negative, header+1+i*SIZEOF_INT64);
}
intlen = sizeof(uint64_t);
intlen = SIZEOF_INT64;
}
v = u.u64;
@@ -956,14 +960,14 @@ sproto_encode(const struct sproto_type *st, void * buffer, int size, sproto_call
return 0;
return -1; // sz == SPROTO_CB_ERROR
}
if (sz == sizeof(uint32_t)) {
if (sz == SIZEOF_INT32) {
if (u.u32 < 0x7fff) {
value = (u.u32+1) * 2;
sz = 2; // sz can be any number > 0
} else {
sz = encode_integer(u.u32, data, size);
}
} else if (sz == sizeof(uint64_t)) {
} else if (sz == SIZEOF_INT64) {
sz= encode_uint64(u.u64, data, size);
} else {
return -1;
@@ -1066,22 +1070,22 @@ decode_array(sproto_callback cb, struct sproto_arg *args, uint8_t * stream) {
int len = *stream;
++stream;
--sz;
if (len == sizeof(uint32_t)) {
if (sz % sizeof(uint32_t) != 0)
if (len == SIZEOF_INT32) {
if (sz % SIZEOF_INT32 != 0)
return -1;
for (i=0;i<sz/sizeof(uint32_t);i++) {
uint64_t value = expand64(todword(stream + i*sizeof(uint32_t)));
for (i=0;i<sz/SIZEOF_INT32;i++) {
uint64_t value = expand64(todword(stream + i*SIZEOF_INT32));
args->index = i+1;
args->value = &value;
args->length = sizeof(value);
cb(args);
}
} else if (len == sizeof(uint64_t)) {
if (sz % sizeof(uint64_t) != 0)
} else if (len == SIZEOF_INT64) {
if (sz % SIZEOF_INT64 != 0)
return -1;
for (i=0;i<sz/sizeof(uint64_t);i++) {
uint64_t low = todword(stream + i*sizeof(uint64_t));
uint64_t hi = todword(stream + i*sizeof(uint64_t) + sizeof(uint32_t));
for (i=0;i<sz/SIZEOF_INT64;i++) {
uint64_t low = todword(stream + i*SIZEOF_INT64);
uint64_t hi = todword(stream + i*SIZEOF_INT64 + SIZEOF_INT32);
uint64_t value = low | hi << 32;
args->index = i+1;
args->value = &value;
@@ -1175,16 +1179,16 @@ sproto_decode(const struct sproto_type *st, const void * data, int size, sproto_
switch (f->type) {
case SPROTO_TINTEGER: {
uint32_t sz = todword(currentdata);
if (sz == sizeof(uint32_t)) {
if (sz == SIZEOF_INT32) {
uint64_t v = expand64(todword(currentdata + SIZEOF_LENGTH));
args.value = &v;
args.length = sizeof(v);
cb(&args);
} else if (sz != sizeof(uint64_t)) {
} else if (sz != SIZEOF_INT64) {
return -1;
} else {
uint32_t low = todword(currentdata + SIZEOF_LENGTH);
uint32_t hi = todword(currentdata + SIZEOF_LENGTH + sizeof(uint32_t));
uint32_t hi = todword(currentdata + SIZEOF_LENGTH + SIZEOF_INT32);
uint64_t v = (uint64_t)low | (uint64_t) hi << 32;
args.value = &v;
args.length = sizeof(v);

View File

@@ -1,7 +1,6 @@
-- read https://github.com/cloudwu/skynet/wiki/FAQ for the module "skynet.core"
local c = require "skynet.core"
local tostring = tostring
local tonumber = tonumber
local coroutine = coroutine
local assert = assert
local pairs = pairs
@@ -10,8 +9,16 @@ local table = table
local profile = require "skynet.profile"
local coroutine_resume = profile.resume
local cresume = profile.resume
local running_thread = nil
local init_thread = nil
local function coroutine_resume(co, ...)
running_thread = co
return cresume(co, ...)
end
local coroutine_yield = profile.yield
local coroutine_create = coroutine.create
local proto = {}
local skynet = {
@@ -28,6 +35,7 @@ local skynet = {
PTYPE_DEBUG = 9,
PTYPE_LUA = 10,
PTYPE_SNAX = 11,
PTYPE_TRACE = 12, -- use for debug trace
}
-- code cache
@@ -45,24 +53,19 @@ end
local session_id_coroutine = {}
local session_coroutine_id = {}
local session_coroutine_address = {}
local session_response = {}
local session_coroutine_tracetag = {}
local unresponse = {}
local wakeup_queue = {}
local sleep_session = {}
local watching_service = {}
local watching_session = {}
local dead_service = {}
local error_queue = {}
local fork_queue = {}
-- suspend is function
local suspend
local function string_to_handle(str)
return tonumber("0x" .. string.sub(str , 2))
end
----- monitor exit
@@ -76,11 +79,13 @@ local function dispatch_error_queue()
end
local function _error_dispatch(error_session, error_source)
skynet.ignoreret() -- don't return for error
if error_session == 0 then
-- service is down
-- Don't remove from watching_service , because user may call dead service
if watching_service[error_source] then
dead_service[error_source] = true
-- error_source is down, clear unreponse set
for resp, address in pairs(unresponse) do
if error_source == address then
unresponse[resp] = nil
end
end
for session, srv in pairs(watching_session) do
if srv == error_source then
@@ -102,148 +107,81 @@ local coroutine_pool = setmetatable({}, { __mode = "kv" })
local function co_create(f)
local co = table.remove(coroutine_pool)
if co == nil then
co = coroutine.create(function(...)
co = coroutine_create(function(...)
f(...)
while true do
local session = session_coroutine_id[co]
if session and session ~= 0 then
local source = debug.getinfo(f,"S")
skynet.error(string.format("Maybe forgot response session %s from %s : %s:%d",
session,
skynet.address(session_coroutine_address[co]),
source.source, source.linedefined))
end
-- coroutine exit
local tag = session_coroutine_tracetag[co]
if tag ~= nil then
if tag then c.trace(tag, "end") end
session_coroutine_tracetag[co] = nil
end
local address = session_coroutine_address[co]
if address then
session_coroutine_id[co] = nil
end
-- recycle co into pool
f = nil
coroutine_pool[#coroutine_pool+1] = co
f = coroutine_yield "EXIT"
-- recv new main function f
f = coroutine_yield "SUSPEND"
f(coroutine_yield())
end
end)
else
-- pass the main function f to coroutine, and restore running thread
local running = running_thread
coroutine_resume(co, f)
running_thread = running
end
return co
end
local function dispatch_wakeup()
local co = table.remove(wakeup_queue,1)
if co then
local session = sleep_session[co]
local token = table.remove(wakeup_queue,1)
if token then
local session = sleep_session[token]
if session then
local co = session_id_coroutine[session]
local tag = session_coroutine_tracetag[co]
if tag then c.trace(tag, "resume") end
session_id_coroutine[session] = "BREAK"
return suspend(co, coroutine_resume(co, false, "BREAK"))
end
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)
function suspend(co, result, command)
if not result then
local session = session_coroutine_id[co]
if session then -- coroutine may fork by others (session is nil)
local addr = session_coroutine_address[co]
if session ~= 0 then
-- only call response error
local tag = session_coroutine_tracetag[co]
if tag then c.trace(tag, "error") end
c.send(addr, skynet.PTYPE_ERROR, session, "")
end
session_coroutine_id[co] = nil
session_coroutine_address[co] = nil
session_coroutine_tracetag[co] = nil
end
skynet.fork(function() end) -- trigger command "SUSPEND"
error(debug.traceback(co,tostring(command)))
end
if command == "CALL" then
session_id_coroutine[param] = co
elseif command == "SLEEP" then
session_id_coroutine[param] = co
sleep_session[co] = param
elseif command == "RETURN" then
local co_session = session_coroutine_id[co]
if co_session == 0 then
if size ~= nil then
c.trash(param, size)
end
return suspend(co, coroutine_resume(co, false)) -- send don't need ret
end
local co_address = session_coroutine_address[co]
if param == nil or session_response[co] then
error(debug.traceback(co))
end
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) ~= nil
if not ret then
-- If the package is too large, returns nil. so we should report error back
c.send(co_address, skynet.PTYPE_ERROR, co_session, "")
end
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)
unresponse[response] = nil
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
-- do not response when session == 0 (send)
if co_session ~= 0 and not dead_service[co_address] then
if ok then
ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, f(...)) ~= nil
if not ret then
-- If the package is too large, returns false. so we should report error back
c.send(co_address, skynet.PTYPE_ERROR, co_session, "")
end
else
ret = c.send(co_address, skynet.PTYPE_ERROR, co_session, "") ~= nil
end
else
ret = false
end
release_watching(co_address)
unresponse[response] = nil
f = nil
return ret
end
watching_service[co_address] = watching_service[co_address] + 1
session_response[co] = true
unresponse[response] = true
return suspend(co, coroutine_resume(co, response))
elseif command == "EXIT" then
-- coroutine exit
local address = session_coroutine_address[co]
if address then
release_watching(address)
session_coroutine_id[co] = nil
session_coroutine_address[co] = nil
session_response[co] = nil
end
if command == "SUSPEND" then
dispatch_wakeup()
dispatch_error_queue()
elseif command == "QUIT" then
-- service exit
return
@@ -256,8 +194,6 @@ function suspend(co, result, command, param, size)
else
error("Unknown command : " .. command .. "\n" .. debug.traceback(co))
end
dispatch_wakeup()
dispatch_error_queue()
end
function skynet.timeout(ti, func)
@@ -266,13 +202,25 @@ function skynet.timeout(ti, func)
local co = co_create(func)
assert(session_id_coroutine[session] == nil)
session_id_coroutine[session] = co
return co -- for debug
end
function skynet.sleep(ti)
local function suspend_sleep(session, token)
local tag = session_coroutine_tracetag[running_thread]
if tag then c.trace(tag, "sleep", 2) end
session_id_coroutine[session] = running_thread
assert(sleep_session[token] == nil, "token duplicative")
sleep_session[token] = session
return coroutine_yield "SUSPEND"
end
function skynet.sleep(ti, token)
local session = c.intcommand("TIMEOUT",ti)
assert(session)
local succ, ret = coroutine_yield("SLEEP", session)
sleep_session[coroutine.running()] = nil
token = token or coroutine.running()
local succ, ret = suspend_sleep(session, token)
sleep_session[token] = nil
if succ then
return
end
@@ -287,31 +235,46 @@ function skynet.yield()
return skynet.sleep(0)
end
function skynet.wait(co)
function skynet.wait(token)
local session = c.genid()
local ret, msg = coroutine_yield("SLEEP", session)
co = co or coroutine.running()
sleep_session[co] = nil
token = token or coroutine.running()
local ret, msg = suspend_sleep(session, token)
sleep_session[token] = nil
session_id_coroutine[session] = nil
end
local self_handle
function skynet.self()
if self_handle then
return self_handle
end
self_handle = string_to_handle(c.command("REG"))
return self_handle
return c.addresscommand "REG"
end
function skynet.localname(name)
local addr = c.command("QUERY", name)
if addr then
return string_to_handle(addr)
end
return c.addresscommand("QUERY", name)
end
skynet.now = c.now
skynet.hpc = c.hpc -- high performance counter
local traceid = 0
function skynet.trace(info)
skynet.error("TRACE", session_coroutine_tracetag[running_thread])
if session_coroutine_tracetag[running_thread] == false then
-- force off trace log
return
end
traceid = traceid + 1
local tag = string.format(":%08x-%d",skynet.self(), traceid)
session_coroutine_tracetag[running_thread] = tag
if info then
c.trace(tag, "trace " .. info)
else
c.trace(tag, "trace")
end
end
function skynet.tracetag()
return session_coroutine_tracetag[running_thread]
end
local starttime
@@ -385,7 +348,8 @@ skynet.trash = assert(c.trash)
local function yield_call(service, session)
watching_session[session] = service
local succ, msg, sz = coroutine_yield("CALL", session)
session_id_coroutine[session] = running_thread
local succ, msg, sz = coroutine_yield "SUSPEND"
watching_session[session] = nil
if not succ then
error "call failed"
@@ -394,6 +358,12 @@ local function yield_call(service, session)
end
function skynet.call(addr, typename, ...)
local tag = session_coroutine_tracetag[running_thread]
if tag then
c.trace(tag, "call", 2)
c.send(addr, skynet.PTYPE_TRACE, 0, tag)
end
local p = proto[typename]
local session = c.send(addr, p.id , nil , p.pack(...))
if session == nil then
@@ -403,28 +373,112 @@ function skynet.call(addr, typename, ...)
end
function skynet.rawcall(addr, typename, msg, sz)
local tag = session_coroutine_tracetag[running_thread]
if tag then
c.trace(tag, "call", 2)
c.send(addr, skynet.PTYPE_TRACE, 0, tag)
end
local p = proto[typename]
local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address")
return yield_call(addr, session)
end
function skynet.tracecall(tag, addr, typename, msg, sz)
c.trace(tag, "tracecall begin")
c.send(addr, skynet.PTYPE_TRACE, 0, tag)
local p = proto[typename]
local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address")
local msg, sz = yield_call(addr, session)
c.trace(tag, "tracecall end")
return msg, sz
end
function skynet.ret(msg, sz)
msg = msg or ""
return coroutine_yield("RETURN", msg, sz)
local tag = session_coroutine_tracetag[running_thread]
if tag then c.trace(tag, "response") end
local co_session = session_coroutine_id[running_thread]
session_coroutine_id[running_thread] = nil
if co_session == 0 then
if sz ~= nil then
c.trash(msg, sz)
end
return false -- send don't need ret
end
local co_address = session_coroutine_address[running_thread]
if not co_session then
error "No session"
end
local ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, msg, sz)
if ret then
return true
elseif ret == false then
-- If the package is too large, returns false. so we should report error back
c.send(co_address, skynet.PTYPE_ERROR, co_session, "")
end
return false
end
function skynet.context()
local co_session = session_coroutine_id[running_thread]
local co_address = session_coroutine_address[running_thread]
return co_session, co_address
end
function skynet.ignoreret()
-- We use session for other uses
session_coroutine_id[running_thread] = nil
end
function skynet.response(pack)
pack = pack or skynet.pack
return coroutine_yield("RESPONSE", pack)
local co_session = assert(session_coroutine_id[running_thread], "no session")
session_coroutine_id[running_thread] = nil
local co_address = session_coroutine_address[running_thread]
if co_session == 0 then
-- do not response when session == 0 (send)
return function() end
end
local function response(ok, ...)
if ok == "TEST" then
return unresponse[response] ~= nil
end
if not pack then
error "Can't response more than once"
end
local ret
if unresponse[response] then
if ok then
ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, pack(...))
if ret == false then
-- If the package is too large, returns false. so we should report error back
c.send(co_address, skynet.PTYPE_ERROR, co_session, "")
end
else
ret = c.send(co_address, skynet.PTYPE_ERROR, co_session, "")
end
unresponse[response] = nil
ret = ret ~= nil
else
ret = false
end
pack = nil
return ret
end
unresponse[response] = co_address
return response
end
function skynet.retpack(...)
return skynet.ret(skynet.pack(...))
end
function skynet.wakeup(co)
if sleep_session[co] then
table.insert(wakeup_queue, co)
function skynet.wakeup(token)
if sleep_session[token] then
table.insert(wakeup_queue, token)
return true
end
end
@@ -463,14 +517,20 @@ function skynet.dispatch_unknown_response(unknown)
end
function skynet.fork(func,...)
local args = table.pack(...)
local co = co_create(function()
func(table.unpack(args,1,args.n))
end)
local n = select("#", ...)
local co
if n == 0 then
co = co_create(func)
else
local args = { ... }
co = co_create(function() func(table.unpack(args,1,n)) end)
end
table.insert(fork_queue, co)
return co
end
local trace_source = {}
local function raw_dispatch_message(prototype, msg, sz, session, source)
-- skynet.PTYPE_RESPONSE = 1, read skynet.h
if prototype == 1 then
@@ -480,35 +540,55 @@ local function raw_dispatch_message(prototype, msg, sz, session, source)
elseif co == nil then
unknown_response(session, source, msg, sz)
else
local tag = session_coroutine_tracetag[co]
if tag then c.trace(tag, "resume") end
session_id_coroutine[session] = nil
suspend(co, coroutine_resume(co, true, msg, sz))
end
else
local p = proto[prototype]
if p == nil then
if session ~= 0 then
if prototype == skynet.PTYPE_TRACE then
-- trace next request
trace_source[source] = c.tostring(msg,sz)
elseif session ~= 0 then
c.send(source, skynet.PTYPE_ERROR, session, "")
else
unknown_request(session, source, msg, sz, prototype)
end
return
end
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
local traceflag = p.trace
if traceflag == false then
-- force off
trace_source[source] = nil
session_coroutine_tracetag[co] = false
else
local tag = trace_source[source]
if tag then
trace_source[source] = nil
c.trace(tag, "request")
session_coroutine_tracetag[co] = tag
elseif traceflag then
-- set running_thread for trace
running_thread = co
skynet.trace()
end
end
suspend(co, coroutine_resume(co, session,source, p.unpack(msg,sz)))
elseif session ~= 0 then
c.send(source, skynet.PTYPE_ERROR, session, "")
else
unknown_request(session, source, msg, sz, proto[prototype].name)
trace_source[source] = nil
if session ~= 0 then
c.send(source, skynet.PTYPE_ERROR, session, "")
else
unknown_request(session, source, msg, sz, proto[prototype].name)
end
end
end
end
@@ -567,6 +647,15 @@ function skynet.harbor(addr)
end
skynet.error = c.error
skynet.tracelog = c.trace
-- true: force on
-- false: force off
-- nil: optional (use skynet.trace() to trace one message)
function skynet.traceproto(prototype, flag)
local p = assert(proto[prototype])
p.trace = flag
end
----- register protocol
do
@@ -646,8 +735,9 @@ end
function skynet.start(start_func)
c.callback(skynet.dispatch_message)
skynet.timeout(0, function()
init_thread = skynet.timeout(0, function()
skynet.init_service(start_func)
init_thread = nil
end)
end
@@ -664,14 +754,41 @@ function skynet.stat(what)
end
function skynet.task(ret)
local t = 0
for session,co in pairs(session_id_coroutine) do
if ret then
if ret == nil then
local t = 0
for session,co in pairs(session_id_coroutine) do
t = t + 1
end
return t
end
if ret == "init" then
if init_thread then
return debug.traceback(init_thread)
else
return
end
end
local tt = type(ret)
if tt == "table" then
for session,co in pairs(session_id_coroutine) do
ret[session] = debug.traceback(co)
end
t = t + 1
return
elseif tt == "number" then
local co = session_id_coroutine[ret]
if co then
return debug.traceback(co)
else
return "No session"
end
elseif tt == "thread" then
for session, co in pairs(session_id_coroutine) do
if co == ret then
return session
end
end
return
end
return t
end
function skynet.term(service)

View File

@@ -70,7 +70,7 @@ local datasheet = {}
local handles = {} -- handle:{ ref:count , name:name , collect:resp }
local dataset = {} -- name:{ handle:handle, monitor:{monitors queue} }
local function releasehandle(handle)
local function releasehandle(source, handle)
local h = handles[handle]
h.ref = h.ref - 1
if h.ref == 0 and h.collect then
@@ -78,10 +78,12 @@ local function releasehandle(handle)
h.collect = nil
handles[handle] = nil
end
local t=dataset[h.name]
t.monitor[source]=nil
end
-- from builder, create or update handle
function datasheet.update(name, handle)
function datasheet.update(source, name, handle)
local t = dataset[name]
if not t then
-- new datasheet
@@ -89,11 +91,11 @@ function datasheet.update(name, handle)
dataset[name] = t
handles[handle] = { ref = 1, name = name }
else
t.handle = handle
-- report update to customers
handles[handle] = { ref = 1 + #t.monitor, name = name }
handles[handle] = { ref = handles[t.handle].ref, name = name }
t.handle = handle
for k,v in ipairs(t.monitor) do
for k,v in pairs(t.monitor) do
v(true, handle)
t.monitor[k] = nil
end
@@ -102,7 +104,7 @@ function datasheet.update(name, handle)
end
-- from customers
function datasheet.query(name)
function datasheet.query(source, name)
local t = assert(dataset[name], "create data first")
local handle = t.handle
local h = handles[handle]
@@ -111,25 +113,25 @@ function datasheet.query(name)
end
-- from customers, monitor handle change
function datasheet.monitor(handle)
function datasheet.monitor(source, handle)
local h = assert(handles[handle], "Invalid data handle")
local t = dataset[h.name]
if t.handle ~= handle then -- already changes
skynet.ret(skynet.pack(t.handle))
else
h.ref = h.ref + 1
table.insert(t.monitor, skynet.response())
assert(not t.monitor[source])
t.monitor[source]=skynet.response()
end
end
-- from customers, release handle , ref count - 1
function datasheet.release(handle)
function datasheet.release(source, handle)
-- send message, don't ret
releasehandle(handle)
releasehandle(source, handle)
end
-- from builder, monitor handle release
function datasheet.collect(handle)
function datasheet.collect(source, handle)
local h = assert(handles[handle], "Invalid data handle")
if h.ref == 0 then
handles[handle] = nil
@@ -140,8 +142,8 @@ function datasheet.collect(handle)
end
end
skynet.dispatch("lua", function(_,_,cmd,...)
datasheet[cmd](...)
skynet.dispatch("lua", function(_,source,cmd,...)
datasheet[cmd](source,...)
end)
skynet.info_func(function()
@@ -155,7 +157,7 @@ skynet.info_func(function()
tmp[v.handle] = nil
info[k] = {
handle = v.handle,
monitors = #v.monitor,
monitors = h.ref,
}
end
for k,v in pairs(tmp) do

View File

@@ -59,7 +59,7 @@ function ctd.dump(root)
local index = dump_table(v)
return '\4', string.pack("<i4", index-1)
elseif t == "number" then
if math.tointeger(v) then
if math.tointeger(v) and v <= 0x7FFFFFFF and v >= -(0x7FFFFFFF+1) then
return '\1', string.pack("<i4", v)
else
return '\2', string.pack("<f",v)

View File

@@ -32,7 +32,7 @@ local function updateobject(name)
core.update(t.object, newhandle)
t.handle = newhandle
skynet.send(datasheet_svr, "lua", "release", handle)
skynet.fork(monitor)
return monitor()
end
skynet.fork(monitor)
end

View File

@@ -82,18 +82,24 @@ local function __parse_addr(addr)
return host, tonumber(port)
end
local auth_method = {}
local function mongo_auth(mongoc)
local user = rawget(mongoc, "username")
local pass = rawget(mongoc, "password")
local authmod = rawget(mongoc, "authmod") or "scram_sha1"
authmod = "auth_" .. authmod
local authdb = rawget(mongoc, "authdb")
if authdb then
authdb = mongo_client.getDB(mongoc, authdb) -- mongoc has not set metatable yet
end
return function()
if user ~= nil and pass ~= nil then
-- autmod can be "mongodb_cr" or "scram_sha1"
local auth_func = mongoc[authmod]
local auth_func = auth_method[authmod]
assert(auth_func , "Invalid authmod")
assert(auth_func(mongoc,user, pass))
assert(auth_func(authdb or mongoc, user, pass))
end
local rs_data = mongoc:runCommand("ismaster")
if rs_data.ok == 1 then
@@ -156,6 +162,7 @@ function mongo.client( conf )
username = first.username,
password = first.password,
authmod = first.authmod,
authdb = first.authdb,
}
obj.__id = 0
@@ -166,6 +173,7 @@ function mongo.client( conf )
auth = mongo_auth(obj),
backup = backup,
nodelay = true,
overload = conf.overload,
}
setmetatable(obj, client_meta)
obj.__sock:connect(true) -- try connect only once
@@ -206,7 +214,7 @@ function mongo_client:runCommand(...)
return self.admin:runCommand(...)
end
function mongo_client:auth_mongodb_cr(user,password)
function auth_method:auth_mongodb_cr(user,password)
local password = md5.sumhexa(string.format("%s:mongo:%s",user,password))
local result= self:runCommand "getnonce"
if result.ok ~=1 then
@@ -229,7 +237,7 @@ local function salt_password(password, salt, iter)
return output
end
function mongo_client:auth_scram_sha1(username,password)
function auth_method:auth_scram_sha1(username,password)
local user = string.gsub(string.gsub(username, '=', '=3D'), ',' , '=2C')
local nonce = crypt.base64encode(crypt.randomkey())
local first_bare = "n=" .. user .. ",r=" .. nonce
@@ -300,6 +308,13 @@ function mongo_client:logout()
return result.ok == 1
end
function mongo_db:auth(user, pass)
local authmod = rawget(self.connection, "authmod") or "scram_sha1"
local auth_func = auth_method["auth_" .. authmod]
assert(auth_func , "Invalid authmod")
return auth_func(self, user, pass)
end
function mongo_db:runCommand(cmd,cmd_v,...)
local conn = self.connection
local request_id = conn:genId()
@@ -340,8 +355,23 @@ function mongo_collection:insert(doc)
sock:request(pack)
end
local function werror(r)
local ok = (r.ok == 1 and not r.writeErrors and not r.writeConcernError)
local err
if not ok then
if r.writeErrors then
err = r.writeErrors[1].errmsg
else
err = r.writeConcernError.errmsg
end
end
return ok, err, r
end
function mongo_collection:safe_insert(doc)
return self.database:runCommand("insert", self.name, "documents", {bson_encode(doc)})
local r = self.database:runCommand("insert", self.name, "documents", {bson_encode(doc)})
return werror(r)
end
function mongo_collection:batch_insert(docs)
@@ -363,12 +393,30 @@ function mongo_collection:update(selector,update,upsert,multi)
sock:request(pack)
end
function mongo_collection:safe_update(selector, update, upsert, multi)
local r = self.database:runCommand("update", self.name, "updates", {bson_encode({
q = selector,
u = update,
upsert = upsert,
multi = multi,
})})
return werror(r)
end
function mongo_collection:delete(selector, single)
local sock = self.connection.__sock
local pack = driver.delete(self.full_name, single, bson_encode(selector))
sock:request(pack)
end
function mongo_collection:safe_delete(selector, single)
local r = self.database:runCommand("delete", self.name, "deletes", {bson_encode({
q = selector,
limit = single and 1 or 0,
})})
return werror(r)
end
function mongo_collection:findOne(query, selector)
local conn = self.connection
local request_id = conn:genId()
@@ -574,10 +622,20 @@ function mongo_cursor:hasNext()
if ok then
if doc then
self.__document = result.result
local doc = result.result
self.__document = doc
self.__data = result.data
self.__ptr = 1
self.__cursor = cursor
local limit = self.__limit
if cursor and limit > 0 then
limit = limit - #doc
if limit <= 0 then
-- reach limit
self:close()
end
self.__limit = limit
end
return true
else
self.__document = nil
@@ -615,11 +673,11 @@ function mongo_cursor:next()
end
function mongo_cursor:close()
-- todo: warning hasNext after close
if self.__cursor then
local sock = self.__collection.connection.__sock
local pack = driver.kill(self.__cursor)
sock:request(pack)
self.__cursor = nil
end
end

View File

@@ -21,26 +21,17 @@ local sha1= crypt.sha1
local setmetatable = setmetatable
local error = error
local tonumber = tonumber
local new_tab = function (narr, nrec) return {} end
local new_tab = function (narr, nrec) return {} end
local _M = { _VERSION = '0.13' }
local _M = { _VERSION = '0.14' }
-- constants
local STATE_CONNECTED = 1
local STATE_COMMAND_SENT = 2
local COM_QUERY = 0x03
local SERVER_MORE_RESULTS_EXISTS = 8
-- 16MB - 1, the default max allowed packet size used by libmysqlclient
local FULL_PACKET_SIZE = 16777215
local mt = { __index = _M }
-- mysql field value type converters
local converters = new_tab(0, 8)
@@ -601,22 +592,22 @@ local function _query_resp(self)
if err ~= "again" then
return true, res
end
local mulitresultset = {res}
mulitresultset.mulitresultset = true
local multiresultset = {res}
multiresultset.multiresultset = true
local i =2
while err =="again" do
res, err, errno, sqlstate = read_result(self,sock)
if not res then
mulitresultset.badresult = true
mulitresultset.err = err
mulitresultset.errno = errno
mulitresultset.sqlstate = sqlstate
return true, mulitresultset
multiresultset.badresult = true
multiresultset.err = err
multiresultset.errno = errno
multiresultset.sqlstate = sqlstate
return true, multiresultset
end
mulitresultset[i]=res
multiresultset[i]=res
i=i+1
end
return true, mulitresultset
return true, multiresultset
end
end
@@ -640,6 +631,7 @@ function _M.connect(opts)
host = opts.host,
port = opts.port or 3306,
auth = _mysql_login(self,user,password,database,opts.on_connect),
overload = opts.overload,
}
self.sockchannel = channel
-- try connect first only once

View File

@@ -143,6 +143,7 @@ function redis.connect(db_conf)
port = db_conf.port or 6379,
auth = redis_login(db_conf.auth, db_conf.db),
nodelay = true,
overload = db_conf.overload,
}
-- try connect first only once
channel:connect(true)

View File

@@ -8,24 +8,60 @@ local crc16 = require "skynet.db.redis.crc16"
local RedisClusterHashSlots = 16384
local RedisClusterRequestTTL = 16
local sync = {
once = {
tasks = {},
}
}
function sync.once.Do(id,func,...)
local tasks = sync.once.tasks
local task = tasks[id]
if not task then
task = {
waiting = {},
result = nil,
}
tasks[id] = task
local rettbl = table.pack(xpcall(func,debug.traceback,...))
task.result = rettbl
local waiting = task.waiting
tasks[id] = nil
if next(waiting) then
for i,co in ipairs(waiting) do
skynet.wakeup(co)
end
end
return table.unpack(task.result,1,task.result.n)
else
local co = coroutine.running()
table.insert(task.waiting,co)
skynet.wait(co)
return table.unpack(task.result,1,task.result.n)
end
end
local _M = {}
local rediscluster = {}
rediscluster.__index = rediscluster
_M.rediscluster = rediscluster
function _M.new(startup_nodes,opt)
function _M.new(startup_nodes,opt,onmessage)
if #startup_nodes == 0 then
startup_nodes = {startup_nodes,}
end
opt = opt or {}
local self = {
startup_nodes = startup_nodes,
max_connections = opt.max_connections or 16,
connections = setmetatable({},{__mode = "kv"}),
max_connections = opt.max_connections or 256,
connections = {},
opt = opt,
refresh_table_asap = false,
-- for subscribe/publish
__onmessage = onmessage,
__watching = {},
}
setmetatable(self,rediscluster)
self:initialize_slots_cache()
@@ -53,13 +89,6 @@ function rediscluster:set_node_name(node)
if not node.name then
node.name = nodename(node)
end
if not node.slaves then
local oldnode = self.name_node[node.name]
if oldnode then
node.slaves = oldnode.slaves
end
end
self.name_node[node.name] = node
end
-- Contact the startup nodes and try to fetch the hash slots -> instances
@@ -67,11 +96,9 @@ end
function rediscluster:initialize_slots_cache()
self.slots = {}
self.nodes = {}
self.name_node = {}
for _,startup_node in ipairs(self.startup_nodes) do
local ok = pcall(function ()
local name = nodename(startup_node)
local conn = self.connections[name] or self:get_redis_link(startup_node)
local conn = self:get_connection(startup_node)
local list = conn:cluster("slots")
for _,result in ipairs(list) do
local ip,port = table.unpack(result[3])
@@ -100,9 +127,6 @@ function rediscluster:initialize_slots_cache()
end
end
self.refresh_table_asap = false
if not self.connections[name] then
self.connections[name] = conn
end
end)
-- Exit the loop as long as the first node replies
if ok then
@@ -154,6 +178,21 @@ function rediscluster:get_key_from_command(argv)
cmd == "shutdown" then
return nil
end
if cmd == "eval" or cmd == "evalsha" then
-- eval script numkeys key [key...] arg [arg...]
local numkeys = argv[3]
local firstkey_slot = nil
for i=4,4+numkeys-1 do
local slot = self:keyslot(argv[i])
if not firstkey_slot then
firstkey_slot = slot
elseif firstkey_slot ~= slot then
error(string.format("%s - all keys must map to the same key slot",cmd))
end
end
-- numkeys <=0 will return nil
return argv[4]
end
-- Unknown commands, and all the commands having the key
-- as first argument are handled here:
-- set, get, ...
@@ -179,21 +218,28 @@ end
function rediscluster:close_all_connection()
local connections = self.connections
self.connections = setmetatable({},{__mode = "kv"})
self.connections = {}
for name,conn in pairs(connections) do
pcall(conn.disconnect,conn)
end
end
function rediscluster:get_connection(node)
node.port = assert(tonumber(node.port))
local name = node.name or nodename(node)
local conn = self.connections[name]
if not conn then
conn = self:get_redis_link(node)
self.connections[name] = conn
if not node.name then
node.name = nodename(node)
end
return self.connections[name]
local name = node.name
local ok,conn = sync.once.Do(name,function ()
local conn = self.connections[name]
if not conn then
self:close_existing_connection()
conn = self:get_redis_link(node)
self.connections[name] = conn
end
return conn
end)
assert(ok,conn)
return conn
end
-- Return a link to a random node, or raise an error if no node can be
@@ -219,25 +265,15 @@ function rediscluster:get_random_connection()
for i,idx in ipairs(shuffle_idx) do
local ok,conn = pcall(function ()
local node = self.nodes[idx]
local conn = self.connections[node.name]
if not conn then
-- Connect the node if it is not connected
conn = self:get_redis_link(node)
if conn:ping() == "PONG" then
self:close_existing_connection()
self.connections[node.name] = conn
return conn
else
-- If the connection is not good close it ASAP in order
-- to avoid waiting for the GC finalizer. File
-- descriptors are a rare resource.
conn:disconnect()
end
local conn = self:get_connection(node)
if conn:ping() == "PONG" then
return conn
else
-- The node was already connected, test the connection.
if conn:ping() == "PONG" then
return conn
end
-- If the connection is not good close it ASAP in order
-- to avoid waiting for the GC finalizer. File
-- descriptors are a rare resource.
self.connetions[node.name] = nil
conn:disconnect()
end
end)
if ok and conn then
@@ -256,33 +292,40 @@ function rediscluster:get_connection_by_slot(slot)
if not node then
return self:get_random_connection()
end
if not self.connections[node.name] then
local ok = pcall(function ()
self:close_existing_connection()
self.connections[node.name] = self:get_redis_link(node)
end)
if not ok then
if self.opt.read_slave and node.slaves and #node.slaves > 0 then
local slave_node = node.slaves[math.random(1,#node.slaves)]
local ok2,conn = pcall(self.get_connection,self,slave_node)
if ok2 then
conn:readonly() -- allow this connection read-slave
return conn
end
local ok,conn = pcall(self.get_connection,self,node)
if not ok then
if self.opt.read_slave and node.slaves and #node.slaves > 0 then
local slave_node = node.slaves[math.random(1,#node.slaves)]
local ok2,conn = pcall(self.get_connection,self,slave_node)
if ok2 then
conn:readonly() -- allow this connection read-slave
return conn
end
-- This will probably never happen with recent redis-rb
-- versions because the connection is enstablished in a lazy
-- way only when a command is called. However it is wise to
-- handle an instance creation error of some kind.
return self:get_random_connection()
end
-- This will probably never happen with recent redis-rb
-- versions because the connection is enstablished in a lazy
-- way only when a command is called. However it is wise to
-- handle an instance creation error of some kind.
return self:get_random_connection()
end
return self.connections[node.name]
return conn
end
-- Dispatch commands.
function rediscluster:call(...)
local argv = table.pack(...)
local cmd = argv[1]
local key = self:get_key_from_command(argv)
if not key then
error("No way to dispatch this command to Redis Cluster: " .. tostring(argv[1]))
end
if cmd == "subscribe" or cmd == "unsubscribe" or cmd == "psubscribe" or cmd == "punsubscribe" then
local slot = self:keyslot(key)
local conn = self:get_watch_connection_by_slot(slot)
local func = conn[cmd]
return func(conn,table.unpack(argv,2))
end
if self.refresh_table_asap then
self:initialize_slots_cache()
end
@@ -292,10 +335,6 @@ function rediscluster:call(...)
local try_random_node = false
while ttl > 0 do
ttl = ttl - 1
local key = self:get_key_from_command(argv)
if not key then
error("No way to dispatch this command to Redis Cluster: " .. tostring(argv[1]))
end
local conn
local slot = self:keyslot(key)
if asking then
@@ -312,7 +351,6 @@ function rediscluster:call(...)
conn:asking()
end
asking = false
local cmd = argv[1]
local func = conn[cmd]
return func(conn,table.unpack(argv,2))
end)}
@@ -321,7 +359,7 @@ function rediscluster:call(...)
err = table.unpack(result,2)
err = tostring(err)
if err == "[Error: socket]" then
-- may be nerver come here?
-- may be never come here?
try_random_node = true
if ttl < RedisClusterRequestTTL/2 then
skynet.sleep(10)
@@ -351,6 +389,10 @@ function rediscluster:call(...)
port = node_port,
}
if not asking then
local oldnode = self.slots[newslot]
if oldnode then
node.slaves = oldnode.slaves
end
self:set_node_name(node)
self.slots[newslot] = node
else
@@ -365,6 +407,44 @@ function rediscluster:call(...)
error(string.format("Too many Cluster redirections?,maybe node is disconnected (last error: %q)",err))
end
function rediscluster:get_watch_connection_by_slot(slot)
if not self.slots[slot] then
self:initialize_slots_cache()
end
local node = assert(self.slots[slot])
return self:get_watch_connection(node)
end
function rediscluster:get_watch_connection(node)
local name = node.name
local ok,conn = sync.once.Do(name,function ()
if not self.__watching[name] then
local conf = {
host = node.host,
port = node.port,
auth = self.opt.auth,
db = self.opt.db or 0,
}
local db = redis.watch(conf)
self.__watching[name] = db
local onmessage = rawget(self,"__onmessage")
skynet.fork(function ()
while true do
local ok,data,channel,pchannel = pcall(db.message,db)
if ok then
if data and onmessage then
pcall(onmessage,data,channel,pchannel)
end
end
end
end)
end
return self.__watching[name]
end)
assert(ok,conn)
return conn
end
-- Currently we handle all the commands using method_missing for
-- simplicity. For a Cluster client actually it will be better to have
-- every single command as a method with the right arity and possibly
@@ -379,5 +459,4 @@ setmetatable(rediscluster,{
end,
})
return _M

View File

@@ -32,10 +32,14 @@ local function init(skynet, export)
skynet.ret(skynet.pack(stat))
end
function dbgcmd.TASK()
local task = {}
skynet.task(task)
skynet.ret(skynet.pack(task))
function dbgcmd.TASK(session)
if session then
skynet.ret(skynet.pack(skynet.task(session)))
else
local task = {}
skynet.task(task)
skynet.ret(skynet.pack(task))
end
end
function dbgcmd.INFO(...)
@@ -50,9 +54,10 @@ local function init(skynet, export)
skynet.exit()
end
function dbgcmd.RUN(source, filename)
function dbgcmd.RUN(source, filename, ...)
local inject = require "skynet.inject"
local ok, output = inject(skynet, source, filename , export.dispatch, skynet.register_protocol)
local args = table.pack(...)
local ok, output = inject(skynet, source, filename, args, export.dispatch, skynet.register_protocol)
collectgarbage "collect"
skynet.ret(skynet.pack(ok, table.concat(output, "\n")))
end
@@ -78,6 +83,16 @@ local function init(skynet, export)
skynet.response() -- get response , but not return. raise error when exit
end
function dbgcmd.TRACELOG(proto, flag)
if type(proto) ~= "string" then
flag = proto
proto = "lua"
end
skynet.error(string.format("Turn trace log %s for %s", flag, proto))
skynet.traceproto(proto, flag)
skynet.ret()
end
return dbgcmd
end -- function init_dbgcmd

View File

@@ -219,7 +219,12 @@ local function unpack_rdata(qtype, chunk)
end
end
local dns_server
local dns_server = {
fd = nil,
address = nil,
port = nil,
retire = nil,
}
local function resolve(content)
if #content < DNS_HEADER_LEN then
@@ -281,6 +286,45 @@ local function resolve(content)
skynet.wakeup(resp.co)
end
local function connect_server()
local fd = socket.udp(function(str, from)
resolve(str)
end)
local ok, err = pcall(socket.udp_connect,fd, dns_server.address, dns_server.port)
if not ok then
socket.close(fd)
error(err)
end
dns_server.fd = fd
skynet.error(string.format("Udp server open %s:%s (%d)", dns_server.address, dns_server.port, fd))
end
local DNS_SERVER_RETIRE = 60 * 100
local function touch_server()
dns_server.retire = skynet.now()
if dns_server.fd then
return
end
connect_server()
local function check_alive()
if skynet.now() > dns_server.retire + DNS_SERVER_RETIRE then
local fd = dns_server.fd
if fd then
dns_server.fd = nil
socket.close(fd)
skynet.error(string.format("Udp server close %s:%s (%d)", dns_server.address, dns_server.port, fd))
end
else
skynet.timeout( 2 * DNS_SERVER_RETIRE, check_alive)
end
end
skynet.timeout( 2 * DNS_SERVER_RETIRE, check_alive)
end
function dns.server(server, port)
if not server then
local f = assert(io.open "/etc/resolv.conf")
@@ -293,11 +337,11 @@ function dns.server(server, port)
f:close()
assert(server, "Can't get nameserver")
end
dns_server = socket.udp(function(str, from)
resolve(str)
end)
socket.udp_connect(dns_server, server, port or 53)
return server
assert(dns_server.fd == nil) -- only set dns.server once
dns_server.address = server
dns_server.port = port or 53
touch_server()
return dns_server.address
end
local function lookup_cache(name, qtype, ignorettl)
@@ -328,7 +372,6 @@ local function suspend(tid, name, qtype)
end
end)
skynet.wait(req.co)
local answers = req.answers
request_pool[tid] = nil
if not req.answers then
local answers = lookup_cache(name, qtype, true)
@@ -354,8 +397,9 @@ function dns.resolve(name, ipv6)
qdcount = 1,
}
local req = pack_header(question_header) .. pack_question(name, qtype, QCLASS.IN)
assert(dns_server, "Call dns.server first")
socket.write(dns_server, req)
assert(dns_server.address, "Call dns.server first")
touch_server()
socket.write(dns_server.fd, req)
return suspend(question_header.tid, name, qtype)
end

View File

@@ -18,7 +18,7 @@ local function getupvaluetable(u, func, unique)
end
end
return function(skynet, source, filename , ...)
return function(skynet, source, filename, args, ...)
if filename then
filename = "@" .. filename
else
@@ -56,7 +56,7 @@ return function(skynet, source, filename , ...)
if not func then
return false, { err }
end
local ok, err = skynet.pcall(func)
local ok, err = skynet.pcall(func, table.unpack(args, 1, args.n))
if not ok then
table.insert(output, err)
return false, output

View File

@@ -3,7 +3,7 @@ local mc = require "skynet.multicast.core"
local multicastd
local multicast = {}
local dispatch = setmetatable({} , {__mode = "kv" })
local dispatch = {}
local chan = {}
local chan_meta = {
@@ -16,14 +16,6 @@ local chan_meta = {
end,
}
local function default_conf(conf)
conf = conf or {}
conf.pack = conf.pack or skynet.pack
conf.unpack = conf.unpack or skynet.unpack
return conf
end
function multicast.new(conf)
assert(multicastd, "Init first")
local self = {}
@@ -70,9 +62,12 @@ function chan:unsubscribe()
local c = assert(self.channel)
skynet.send(multicastd, "lua", "USUB", c)
self.__subscribe = nil
dispatch[c] = nil
end
local function dispatch_subscribe(channel, source, pack, msg, sz)
-- channel as session, do need response
skynet.ignoreret()
local self = dispatch[channel]
if not self then
mc.close(pack)

View File

@@ -1,9 +0,0 @@
local core = require "skynet.reload.core"
local skynet = require "skynet"
local function reload(...)
local args = SERVICE_NAME .. " " .. table.concat({...}, " ")
print(args)
end
return reload

View File

@@ -438,6 +438,7 @@ end
socket.sendto = assert(driver.udp_send)
socket.udp_address = assert(driver.udp_address)
socket.netstat = assert(driver.info)
function socket.warning(id, callback)
local obj = socket_pool[id]

View File

@@ -39,6 +39,8 @@ function socket_channel.channel(desc)
__closed = false,
__authcoroutine = false,
__nodelay = desc.nodelay,
__overload_notify = desc.overload,
__overload = false,
}
return setmetatable(c, channel_meta)
@@ -77,17 +79,6 @@ local function wakeup_all(self, errmsg)
end
end
local function exit_thread(self)
local co = coroutine.running()
if self.__dispatch_thread == co then
self.__dispatch_thread = nil
local connecting = self.__connecting_thread
if connecting then
skynet.wakeup(connecting)
end
end
end
local function dispatch_by_session(self)
local response = self.__response
-- response() return session
@@ -124,7 +115,6 @@ local function dispatch_by_session(self)
wakeup_all(self, errormsg)
end
end
exit_thread(self)
end
local function pop_response(self)
@@ -153,6 +143,25 @@ local function push_response(self, response, co)
end
end
local function get_response(func, sock)
local result_ok, result_data, padding = func(sock)
if result_ok and padding then
local result = { result_data }
local index = 2
repeat
result_ok, result_data, padding = func(sock)
if not result_ok then
return result_ok, result_data
end
result[index] = result_data
index = index + 1
until not padding
return true, result
else
return result_ok, result_data
end
end
local function dispatch_by_order(self)
while self.__sock do
local func, co = pop_response(self)
@@ -161,23 +170,15 @@ local function dispatch_by_order(self)
wakeup_all(self, "channel_closed")
break
end
local ok, result_ok, result_data, padding = pcall(func, self.__sock)
local ok, result_ok, result_data = pcall(get_response, func, self.__sock)
if ok then
if padding and result_ok then
-- if padding is true, wait for next result_data
-- self.__result_data[co] is a table
local result = self.__result_data[co] or {}
self.__result_data[co] = result
table.insert(result, result_data)
self.__result[co] = result_ok
if result_ok and self.__result_data[co] then
table.insert(self.__result_data[co], result_data)
else
self.__result[co] = result_ok
if result_ok and self.__result_data[co] then
table.insert(self.__result_data[co], result_data)
else
self.__result_data[co] = result_data
end
skynet.wakeup(co)
self.__result_data[co] = result_data
end
skynet.wakeup(co)
else
close_channel_socket(self)
local errmsg
@@ -190,7 +191,6 @@ local function dispatch_by_order(self)
wakeup_all(self, errmsg)
end
end
exit_thread(self)
end
local function dispatch_function(self)
@@ -222,11 +222,21 @@ local function connect_backup(self)
end
end
local function term_dispatch_thread(self)
if not self.__response and self.__dispatch_thread then
-- dispatch by order, send close signal to dispatch thread
push_response(self, true, false) -- (true, false) is close signal
end
end
local function connect_once(self)
if self.__closed then
return false
end
assert(not self.__sock and not self.__authcoroutine)
-- term current dispatch thread (send a signal)
term_dispatch_thread(self)
local fd,err = socket.open(self.__host, self.__port)
if not fd then
fd = connect_backup(self)
@@ -238,8 +248,43 @@ local function connect_once(self)
socketdriver.nodelay(fd)
end
-- register overload warning
local overload = self.__overload_notify
if overload then
local function overload_trigger(id, size)
if id == self.__sock[1] then
if size == 0 then
if self.__overload then
self.__overload = false
overload(false)
end
else
if not self.__overload then
self.__overload = true
overload(true)
else
skynet.error(string.format("WARNING: %d K bytes need to send out (fd = %d %s:%s)", size, id, self.__host, self.__port))
end
end
end
end
skynet.fork(overload_trigger, fd, 0)
socket.warning(fd, overload_trigger)
end
while self.__dispatch_thread do
-- wait for dispatch thread exit
skynet.yield()
end
self.__sock = setmetatable( {fd} , channel_socket_meta )
self.__dispatch_thread = skynet.fork(dispatch_function(self), self)
self.__dispatch_thread = skynet.fork(function()
pcall(dispatch_function(self), self)
-- clear dispatch_thread
self.__dispatch_thread = nil
end)
if self.__auth then
self.__authcoroutine = coroutine.running()
@@ -342,18 +387,7 @@ local function block_connect(self, once)
end
function channel:connect(once)
if self.__closed then
if self.__dispatch_thread then
-- closing, wait
assert(self.__connecting_thread == nil, "already connecting")
local co = coroutine.running()
self.__connecting_thread = co
skynet.wait(co)
self.__connecting_thread = nil
end
self.__closed = false
end
self.__closed = false
return block_connect(self, once)
end
@@ -426,13 +460,9 @@ end
function channel:close()
if not self.__closed then
local thread = self.__dispatch_thread
term_dispatch_thread(self)
self.__closed = true
close_channel_socket(self)
if not self.__response and self.__dispatch_thread == thread and thread then
-- dispatch by order, send close signal to dispatch thread
push_response(self, true, false) -- (true, false) is close signal
end
end
end

View File

@@ -71,7 +71,7 @@ local dummy_env = {}
for k,v in pairs(_ENV) do dummy_env[k] = v end
local function _patch(global, f)
local i = 1
local i = 1
while true do
local name, value = debug.getupvalue(f, i)
if name == nil then
@@ -81,10 +81,10 @@ local function _patch(global, f)
if old_uv then
debug.upvaluejoin(f, i, old_uv.func, old_uv.index)
end
else
if type(value) == "function" then
_patch(global, value)
end
else
if type(value) == "function" then
_patch(global, value)
end
end
i = i + 1
end
@@ -92,7 +92,7 @@ end
local function patch_func(funcs, global, group, name, f)
local desc = assert(find_func(funcs, group, name) , string.format("Patch mismatch %s.%s", group, name))
_patch(global, f)
_patch(global, f)
desc[4] = f
end

View File

@@ -1,24 +1,23 @@
local skynet = require "skynet"
local function dft_loader(path, name, G)
local errlist = {}
local errlist = {}
for pat in string.gmatch(path,"[^;]+") do
local filename = string.gsub(pat, "?", name)
local f , err = loadfile(filename, "bt", G)
if f then
return f, pat
else
table.insert(errlist, err)
end
end
for pat in string.gmatch(path,"[^;]+") do
local filename = string.gsub(pat, "?", name)
local f , err = loadfile(filename, "bt", G)
if f then
return f, pat
else
table.insert(errlist, err)
end
end
error(table.concat(errlist, "\n"))
error(table.concat(errlist, "\n"))
end
return function (name , G, loader)
loader = loader or dft_loader
local mainfunc
loader = loader or dft_loader
local function func_id(id, group)
local tmp = {}
@@ -42,7 +41,6 @@ return function (name , G, loader)
assert(getmetatable(G) == nil)
assert(G.init == nil)
assert(G.exit == nil)
assert(G.accept == nil)
assert(G.response == nil)
end
@@ -75,10 +73,8 @@ return function (name , G, loader)
end
end
local pattern
local path = assert(skynet.getenv "snax" , "please set snax in config file")
mainfunc, pattern = loader(path, name, G)
local path = assert(skynet.getenv "snax" , "please set snax in config file")
local mainfunc, pattern = loader(path, name, G)
setmetatable(G, { __index = env , __newindex = init_system })
local ok, err = xpcall(mainfunc, debug.traceback)

View File

@@ -10,7 +10,7 @@
#include <stdio.h>
#include <stdarg.h>
#define BACKLOG 32
#define BACKLOG 128
struct connection {
int id; // skynet_socket id
@@ -168,21 +168,26 @@ _report(struct gate * g, const char * data, ...) {
static void
_forward(struct gate *g, struct connection * c, int size) {
struct skynet_context * ctx = g->ctx;
int fd = c->id;
if (fd <= 0) {
// socket error
return;
}
if (g->broker) {
void * temp = skynet_malloc(size);
databuffer_read(&c->buffer,&g->mp,temp, size);
skynet_send(ctx, 0, g->broker, g->client_tag | PTYPE_TAG_DONTCOPY, 1, temp, size);
skynet_send(ctx, 0, g->broker, g->client_tag | PTYPE_TAG_DONTCOPY, fd, temp, size);
return;
}
if (c->agent) {
void * temp = skynet_malloc(size);
databuffer_read(&c->buffer,&g->mp,temp, size);
skynet_send(ctx, c->client, c->agent, g->client_tag | PTYPE_TAG_DONTCOPY, 1 , temp, size);
skynet_send(ctx, c->client, c->agent, g->client_tag | PTYPE_TAG_DONTCOPY, fd , temp, size);
} else if (g->watchdog) {
char * tmp = skynet_malloc(size + 32);
int n = snprintf(tmp,32,"%d data ",c->id);
databuffer_read(&c->buffer,&g->mp,tmp+n,size);
skynet_send(ctx, 0, g->watchdog, PTYPE_TEXT | PTYPE_TAG_DONTCOPY, 1, tmp, size + n);
skynet_send(ctx, 0, g->watchdog, PTYPE_TEXT | PTYPE_TAG_DONTCOPY, fd, tmp, size + n);
}
}
@@ -308,7 +313,7 @@ _cb(struct skynet_context * ctx, void * ud, int type, int session, uint32_t sour
static int
start_listen(struct gate *g, char * listen_addr) {
struct skynet_context * ctx = g->ctx;
char * portstr = strchr(listen_addr,':');
char * portstr = strrchr(listen_addr,':');
const char * host = "";
int port;
if (portstr == NULL) {

View File

@@ -690,21 +690,27 @@ mainloop(struct skynet_context * context, void * ud, int type, int session, uint
harbor_command(h, msg,sz,session,source);
return 0;
}
default: {
case PTYPE_SYSTEM : {
// remote message out
const struct remote_message *rmsg = msg;
if (rmsg->destination.handle == 0) {
if (remote_send_name(h, source , rmsg->destination.name, type, session, rmsg->message, rmsg->sz)) {
if (remote_send_name(h, source , rmsg->destination.name, rmsg->type, session, rmsg->message, rmsg->sz)) {
return 0;
}
} else {
if (remote_send_handle(h, source , rmsg->destination.handle, type, session, rmsg->message, rmsg->sz)) {
if (remote_send_handle(h, source , rmsg->destination.handle, rmsg->type, session, rmsg->message, rmsg->sz)) {
return 0;
}
}
skynet_free((void *)rmsg->message);
return 0;
}
default:
skynet_error(context, "recv invalid message from %x, type = %d", source, type);
if (session != 0 && type != PTYPE_ERROR) {
skynet_send(context,0,source,PTYPE_ERROR, session, NULL, 0);
}
return 0;
}
}

View File

@@ -65,7 +65,6 @@ logger_init(struct logger * inst, struct skynet_context *ctx, const char * parm)
}
if (inst->handle) {
skynet_callback(ctx, inst, logger_cb);
skynet_command(ctx, "REG", ".logger");
return 0;
}
return 1;

145
service/clusteragent.lua Normal file
View File

@@ -0,0 +1,145 @@
local skynet = require "skynet"
local sc = require "skynet.socketchannel"
local socket = require "skynet.socket"
local cluster = require "skynet.cluster.core"
local ignoreret = skynet.ignoreret
local clusterd, gate, fd = ...
clusterd = tonumber(clusterd)
gate = tonumber(gate)
fd = tonumber(fd)
local large_request = {}
local inquery_name = {}
local register_name_mt = { __index =
function(self, name)
local waitco = inquery_name[name]
if waitco then
local co=coroutine.running()
table.insert(waitco, co)
skynet.wait(co)
return rawget(self, name)
else
waitco = {}
inquery_name[name] = waitco
local addr = skynet.call(clusterd, "lua", "queryname", name:sub(2)) -- name must be '@xxxx'
if addr then
self[name] = addr
end
inquery_name[name] = nil
for _, co in ipairs(waitco) do
skynet.wakeup(co)
end
return addr
end
end
}
local function new_register_name()
return setmetatable({}, register_name_mt)
end
local register_name = new_register_name()
local tracetag
local function dispatch_request(_,_,addr, session, msg, sz, padding, is_push)
ignoreret() -- session is fd, don't call skynet.ret
if session == nil then
-- trace
tracetag = addr
return
end
if padding then
local req = large_request[session] or { addr = addr , is_push = is_push, tracetag = tracetag }
tracetag = nil
large_request[session] = req
cluster.append(req, msg, sz)
return
else
local req = large_request[session]
if req then
tracetag = req.tracetag
large_request[session] = nil
cluster.append(req, msg, sz)
msg,sz = cluster.concat(req)
addr = req.addr
is_push = req.is_push
end
if not msg then
tracetag = nil
local response = cluster.packresponse(session, false, "Invalid large req")
socket.write(fd, response)
return
end
end
local ok, response
if addr == 0 then
local name = skynet.unpack(msg, sz)
skynet.trash(msg, sz)
local addr = register_name["@" .. name]
if addr then
ok = true
msg, sz = skynet.pack(addr)
else
ok = false
msg = "name not found"
end
else
if cluster.isname(addr) then
addr = register_name[addr]
end
if addr then
if is_push then
skynet.rawsend(addr, "lua", msg, sz)
return -- no response
else
if tracetag then
ok , msg, sz = pcall(skynet.tracecall, tracetag, addr, "lua", msg, sz)
tracetag = nil
else
ok , msg, sz = pcall(skynet.rawcall, addr, "lua", msg, sz)
end
end
else
ok = false
msg = "Invalid name"
end
end
if ok then
response = cluster.packresponse(session, true, msg, sz)
if type(response) == "table" then
for _, v in ipairs(response) do
socket.lwrite(fd, v)
end
else
socket.write(fd, response)
end
else
response = cluster.packresponse(session, false, msg)
socket.write(fd, response)
end
end
skynet.start(function()
skynet.register_protocol {
name = "client",
id = skynet.PTYPE_CLIENT,
unpack = cluster.unpackrequest,
dispatch = dispatch_request,
}
-- fd can write, but don't read fd, the data package will forward from gate though client protocol.
skynet.call(gate, "lua", "forward", fd)
skynet.dispatch("lua", function(_,source, cmd, ...)
if cmd == "exit" then
socket.close(fd)
skynet.exit()
elseif cmd == "namechange" then
register_name = new_register_name()
else
skynet.error(string.format("Invalid command %s from %s", cmd, skynet.address(source)))
end
end)
end)

View File

@@ -7,6 +7,8 @@ local config_name = skynet.getenv "cluster"
local node_address = {}
local node_session = {}
local command = {}
local config = {}
local nodename = cluster.nodename()
local function read_response(sock)
local sz = socket.header(sock:read(2))
@@ -14,16 +16,49 @@ local function read_response(sock)
return cluster.unpackresponse(msg) -- session, ok, data, padding
end
local connecting = {}
local function open_channel(t, key)
local host, port = string.match(node_address[key], "([^:]+):(.*)$")
local c = sc.channel {
host = host,
port = tonumber(port),
response = read_response,
nodelay = true,
}
assert(c:connect(true))
t[key] = c
local ct = connecting[key]
if ct then
local co = coroutine.running()
table.insert(ct, co)
skynet.wait(co)
return assert(ct.channel)
end
ct = {}
connecting[key] = ct
local address = node_address[key]
if address == nil and not config.nowaiting then
local co = coroutine.running()
assert(ct.namequery == nil)
ct.namequery = co
skynet.error("Waiting for cluster node [".. key.."]")
skynet.wait(co)
address = node_address[key]
end
local succ, err, c
if address then
local host, port = string.match(address, "([^:]+):(.*)$")
c = sc.channel {
host = host,
port = tonumber(port),
response = read_response,
nodelay = true,
}
succ, err = pcall(c.connect, c, true)
if succ then
t[key] = c
ct.channel = c
end
else
err = string.format("cluster node [%s] is %s.", key, address == false and "down" or "absent")
end
connecting[key] = nil
for _, co in ipairs(ct) do
skynet.wakeup(co)
end
assert(succ, err)
return c
end
@@ -40,13 +75,32 @@ local function loadconfig(tmp)
end
end
for name,address in pairs(tmp) do
assert(type(address) == "string")
if node_address[name] ~= address then
-- address changed
if rawget(node_channel, name) then
node_channel[name] = nil -- reset connection
if name:sub(1,2) == "__" then
name = name:sub(3)
config[name] = address
skynet.error(string.format("Config %s = %s", name, address))
else
assert(address == false or type(address) == "string")
if node_address[name] ~= address then
-- address changed
if rawget(node_channel, name) then
node_channel[name] = nil -- reset connection
end
node_address[name] = address
end
local ct = connecting[name]
if ct and ct.namequery and not config.nowaiting then
skynet.error(string.format("Cluster node [%s] resloved : %s", name, address))
skynet.wakeup(ct.namequery)
end
end
end
if config.nowaiting then
-- wakeup all connecting request
for name, ct in pairs(connecting) do
if ct.namequery then
skynet.wakeup(ct.namequery)
end
node_address[name] = address
end
end
end
@@ -59,7 +113,8 @@ end
function command.listen(source, addr, port)
local gate = skynet.newservice("gate")
if port == nil then
addr, port = string.match(node_address[addr], "([^:]+):(.*)$")
local address = assert(node_address[addr], addr .. " is down")
addr, port = string.match(address, "([^:]+):(.*)$")
end
skynet.call(gate, "lua", "open", { address = addr, port = port })
skynet.ret(skynet.pack(nil))
@@ -74,11 +129,22 @@ local function send_request(source, node, addr, msg, sz)
-- node_channel[node] may yield or throw error
local c = node_channel[node]
local tracetag = skynet.tracetag()
if tracetag then
if tracetag:sub(1,1) ~= "(" then
-- add nodename
local newtag = string.format("(%s-%s-%d)%s", nodename, node, session, tracetag)
skynet.tracelog(tracetag, string.format("session %s", newtag))
tracetag = newtag
end
skynet.tracelog(tracetag, string.format("cluster %s", node))
c:request(cluster.packtrace(tracetag))
end
return c:request(request, session, padding)
end
function command.req(...)
local ok, msg, sz = pcall(send_request, ...)
local ok, msg = pcall(send_request, ...)
if ok then
if type(msg) == "table" then
skynet.ret(cluster.concat(msg))
@@ -109,6 +175,12 @@ end
local proxy = {}
function command.proxy(source, node, name)
if name == nil then
node, name = node:match "^([^@.]+)([@.].+)"
if name == nil then
error ("Invalid name " .. tostring(node))
end
end
local fullname = node .. "." .. name
if proxy[fullname] == nil then
proxy[fullname] = skynet.newservice("clusterproxy", node, name)
@@ -116,14 +188,24 @@ function command.proxy(source, node, name)
skynet.ret(skynet.pack(proxy[fullname]))
end
local cluster_agent = {} -- fd:service
local register_name = {}
local function clearnamecache()
for fd, service in pairs(cluster_agent) do
if type(service) == "number" then
skynet.send(service, "lua", "namechange")
end
end
end
function command.register(source, name, addr)
assert(register_name[name] == nil)
addr = addr or source
local old_name = register_name[addr]
if old_name then
register_name[old_name] = nil
clearnamecache()
end
register_name[addr] = name
register_name[name] = addr
@@ -131,76 +213,35 @@ function command.register(source, name, addr)
skynet.error(string.format("Register [%s] :%08x", name, addr))
end
local large_request = {}
function command.queryname(source, name)
skynet.ret(skynet.pack(register_name[name]))
end
function command.socket(source, subcmd, fd, msg)
if subcmd == "data" then
local sz
local addr, session, msg, padding, is_push = cluster.unpackrequest(msg)
if padding then
local requests = large_request[fd]
if requests == nil then
requests = {}
large_request[fd] = requests
end
local req = requests[session] or { addr = addr , is_push = is_push }
requests[session] = req
table.insert(req, msg)
return
else
local requests = large_request[fd]
if requests then
local req = requests[session]
if req then
requests[session] = nil
table.insert(req, msg)
msg,sz = cluster.concat(req)
addr = req.addr
is_push = req.is_push
end
end
if not msg then
local response = cluster.packresponse(session, false, "Invalid large req")
socket.write(fd, response)
return
end
end
local ok, response
if addr == 0 then
local name = skynet.unpack(msg, sz)
local addr = register_name[name]
if addr then
ok = true
msg, sz = skynet.pack(addr)
else
ok = false
msg = "name not found"
end
elseif is_push then
skynet.rawsend(addr, "lua", msg, sz)
return -- no response
else
ok , msg, sz = pcall(skynet.rawcall, addr, "lua", msg, sz)
end
if ok then
response = cluster.packresponse(session, true, msg, sz)
if type(response) == "table" then
for _, v in ipairs(response) do
socket.lwrite(fd, v)
end
else
socket.write(fd, response)
end
else
response = cluster.packresponse(session, false, msg)
socket.write(fd, response)
end
elseif subcmd == "open" then
if subcmd == "open" then
skynet.error(string.format("socket accept from %s", msg))
skynet.call(source, "lua", "accept", fd)
-- new cluster agent
cluster_agent[fd] = false
local agent = skynet.newservice("clusteragent", skynet.self(), source, fd)
local closed = cluster_agent[fd]
cluster_agent[fd] = agent
if closed then
skynet.send(agent, "lua", "exit")
cluster_agent[fd] = nil
end
else
large_request[fd] = nil
skynet.error(string.format("socket %s %d %s", subcmd, fd, msg or ""))
if subcmd == "close" or subcmd == "error" then
-- close cluster agent
local agent = cluster_agent[fd]
if type(agent) == "boolean" then
cluster_agent[fd] = true
else
skynet.send(agent, "lua", "exit")
cluster_agent[fd] = nil
end
else
skynet.error(string.format("socket %s %d %s", subcmd, fd, msg or ""))
end
end
end

View File

@@ -1,5 +1,6 @@
local skynet = require "skynet"
local socket = require "skynet.socket"
local socketdriver = require "skynet.socketdriver"
require "skynet.manager" -- import skynet.launch, ...
local table = table
@@ -41,6 +42,7 @@ local function connect_slave(slave_id, address)
local ok, err = pcall(function()
if slaves[slave_id] == nil then
local fd = assert(socket.open(address), "Can't connect to "..address)
socketdriver.nodelay(fd)
skynet.error(string.format("Connect to harbor %d (fd=%d), %s", slave_id, fd, address))
slaves[slave_id] = fd
monitor_clear(slave_id)
@@ -249,6 +251,7 @@ skynet.start(function()
local co = coroutine.running()
socket.start(slave_fd, function(fd, addr)
skynet.error(string.format("New connection (fd = %d, %s)",fd, addr))
socketdriver.nodelay(fd)
if pcall(accept_slave,fd) then
local s = 0
for k,v in pairs(slaves) do

View File

@@ -158,6 +158,8 @@ function COMMAND.help()
shrtbl = "Show shared short string table info",
ping = "ping address",
call = "call address ...",
trace = "trace address [proto] [on|off]",
netstat = "netstat : show netstat",
}
end
@@ -206,7 +208,10 @@ function COMMAND.service()
end
local function adjust_address(address)
if address:sub(1,1) ~= ":" then
local prefix = address:sub(1,1)
if prefix == '.' then
return assert(skynet.localname(address), "Not a valid name")
elseif prefix ~= ':' then
address = assert(tonumber("0x" .. address), "Need an address") | (skynet.harbor(skynet.self()) << 24)
end
return address
@@ -236,7 +241,7 @@ function COMMAND.exit(address)
skynet.send(adjust_address(address), "debug", "EXIT")
end
function COMMAND.inject(address, filename)
function COMMAND.inject(address, filename, ...)
address = adjust_address(address)
local f = io.open(filename, "rb")
if not f then
@@ -244,7 +249,7 @@ function COMMAND.inject(address, filename)
end
local source = f:read "*a"
f:close()
local ok, output = skynet.call(address, "debug", "RUN", source, filename)
local ok, output = skynet.call(address, "debug", "RUN", source, filename, ...)
if ok == false then
error(output)
end
@@ -343,6 +348,22 @@ function COMMAND.ping(address)
return tostring(ti)
end
local function toboolean(x)
return x and (x == "true" or x == "on")
end
function COMMAND.trace(address, proto, flag)
address = adjust_address(address)
if flag == nil then
if proto == "on" or proto == "off" then
proto = toboolean(proto)
end
else
flag = toboolean(flag)
end
skynet.call(address, "debug", "TRACELOG", proto, flag)
end
function COMMANDX.call(cmd)
local address = adjust_address(cmd[2])
local cmdline = assert(cmd[1]:match("%S+%s+%S+%s(.+)") , "need arguments")
@@ -354,3 +375,50 @@ function COMMANDX.call(cmd)
local rets = table.pack(skynet.call(address, "lua", table.unpack(args, 2, args.n)))
return rets
end
local function bytes(size)
if size == nil or size == 0 then
return
end
if size < 1024 then
return size
end
if size < 1024 * 1024 then
return tostring(size/1024) .. "K"
end
return tostring(size/(1024*1024)) .. "M"
end
local function convert_stat(info)
local now = skynet.now()
local function time(t)
if t == nil then
return
end
t = now - t
if t < 6000 then
return tostring(t/100) .. "s"
end
local hour = t // (100*60*60)
t = t - hour * 100 * 60 * 60
local min = t // (100*60)
t = t - min * 100 * 60
local sec = t / 100
return string.format("%s%d:%.2gs",hour == 0 and "" or (hour .. ":"),min,sec)
end
info.address = skynet.address(info.address)
info.read = bytes(info.read)
info.write = bytes(info.write)
info.wbuffer = bytes(info.wbuffer)
info.rtime = time(info.rtime)
info.wtime = time(info.wtime)
end
function COMMAND.netstat()
local stat = socket.netstat()
for _, info in ipairs(stat) do
convert_stat(info)
end
return stat
end

Some files were not shown because too many files have changed in this diff Show More