ready for 0.9.0

Ready for 0.9.0

3rd/lua/lgc.c

@@ -403,7 +403,7 @@ static int traverseephemeron (global_State *g, Table *h) {
       reallymarkobject(g, gcvalue(gval(n)));  /* mark it now */
     }
   }
-  if (prop)
+  if (g->gcstate != GCSatomic || prop)
     linktable(h, &g->ephemeron);  /* have to propagate again */
   else if (hasclears)  /* does table have white keys? */
     linktable(h, &g->allweak);  /* may have to clean white keys */

HISTORY.md

@@ -1,3 +1,31 @@
+v0.9.0 (2014-11-17)
+-----------
+* Add UDP support
+* Add IPv6 support
+* socket send package can define a release method
+* dispatch read before write in epoll
+* remove snax queue mode
+* Fix a bug in big-endian architecture
+
+v0.8.1 (2014-11-3)
+-----------
+* Send to an invalid remote service will raise an error
+* Bugifx: socket open address string
+* Remove sha1 from mysqlaux
+* merge lua and sproto bugfix , use crypt lib instead
+* Fix a memory leak in socket
+* minor bugfix in http module
+
+v0.8.0 (2014-10-27)
+-----------
+* Add mysql client driver
+* Bugfix : skynet.queue
+
+v0.7.4 (2014-10-13)
+-----------
+* Bugfix : clear coroutine pool when GC
+* hotfix : A bug introduce by 0.7.3 
+
 v0.7.3 (2014-10-13)
 -----------
 * Add some logs (warning) when overload

Makefile

@@ -43,7 +43,8 @@ jemalloc : $(MALLOC_STATICLIB)
 CSERVICE = snlua logger gate harbor
 LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack \
   clientsocket memory profile multicast \
-  cluster crypt sharedata stm sproto lpeg
+  cluster crypt sharedata stm sproto lpeg \
+  mysqlaux
 
 SKYNET_SRC = skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c \
   skynet_server.c skynet_start.c skynet_timer.c skynet_error.c \
@@ -122,6 +123,9 @@ $(LUA_CLIB_PATH)/sproto.so : lualib-src/sproto/sproto.c lualib-src/sproto/lsprot
 $(LUA_CLIB_PATH)/lpeg.so : 3rd/lpeg/lpcap.c 3rd/lpeg/lpcode.c 3rd/lpeg/lpprint.c 3rd/lpeg/lptree.c 3rd/lpeg/lpvm.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) -I3rd/lpeg $^ -o $@ 
 
+$(LUA_CLIB_PATH)/mysqlaux.so : lualib-src/lua_mysqlaux.c | $(LUA_CLIB_PATH)
+	$(CC) $(CFLAGS) $(SHARED) $^ -o $@	
+
 clean :
 	rm -f $(SKYNET_BUILD_PATH)/skynet $(CSERVICE_PATH)/*.so $(LUA_CLIB_PATH)/*.so
 

examples/config.mysql

@@ -0,0 +1,11 @@
+root = "./"
+thread = 8
+logger = nil
+harbor = 0
+start = "main_mysql"	-- 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"

examples/main_mysql.lua

@@ -0,0 +1,10 @@
+local skynet = require "skynet"
+
+
+skynet.start(function()
+	print("Main Server start")
+	local console = skynet.newservice("testmysql")
+	
+	print("Main Server exit")
+	skynet.exit()
+end)

examples/simpleweb.lua

@@ -66,6 +66,7 @@ skynet.start(function()
 	end
 	local balance = 1
 	local id = socket.listen("0.0.0.0", 8001)
+	skynet.error("Listen web port 8001")
 	socket.start(id , function(id, addr)
 		skynet.error(string.format("%s connected, pass it to agent :%08x", addr, agent[balance]))
 		skynet.send(agent[balance], "lua", id)

lualib-src/lua-netpack.c

@@ -302,9 +302,9 @@ filter_data(lua_State *L, int fd, uint8_t * buffer, int size) {
 }
 
 static void
-pushstring(lua_State *L, const char * msg) {
+pushstring(lua_State *L, const char * msg, int size) {
 	if (msg) {
-		lua_pushstring(L, msg);
+		lua_pushlstring(L, msg, size);
 	} else {
 		lua_pushliteral(L, "");
 	}
@@ -350,12 +350,12 @@ lfilter(lua_State *L) {
 		lua_pushvalue(L, lua_upvalueindex(TYPE_OPEN));
 		// ignore listen id (message->id);
 		lua_pushinteger(L, message->ud);
-		pushstring(L, buffer);
+		pushstring(L, buffer, size);
 		return 4;
 	case SKYNET_SOCKET_TYPE_ERROR:
 		lua_pushvalue(L, lua_upvalueindex(TYPE_ERROR));
 		lua_pushinteger(L, message->id);
-		pushstring(L, buffer);
+		pushstring(L, buffer, size);
 		return 4;
 	default:
 		// never get here

lualib-src/lua-seri.c

@@ -120,37 +120,37 @@ rb_read(struct read_block *rb, void *buffer, int sz) {
 
 static inline void
 wb_nil(struct write_block *wb) {
-	int n = TYPE_NIL;
+	uint8_t n = TYPE_NIL;
 	wb_push(wb, &n, 1);
 }
 
 static inline void
 wb_boolean(struct write_block *wb, int boolean) {
-	int n = COMBINE_TYPE(TYPE_BOOLEAN , boolean ? 1 : 0);
+	uint8_t n = COMBINE_TYPE(TYPE_BOOLEAN , boolean ? 1 : 0);
 	wb_push(wb, &n, 1);
 }
 
 static inline void
 wb_integer(struct write_block *wb, int v, int type) {
 	if (v == 0) {
-		int n = COMBINE_TYPE(type , 0);
+		uint8_t n = COMBINE_TYPE(type , 0);
 		wb_push(wb, &n, 1);
 	} else if (v<0) {
-		int n = COMBINE_TYPE(type , 4);
+		uint8_t n = COMBINE_TYPE(type , 4);
 		wb_push(wb, &n, 1);
 		wb_push(wb, &v, 4);
 	} else if (v<0x100) {
-		int n = COMBINE_TYPE(type , 1);
+		uint8_t n = COMBINE_TYPE(type , 1);
 		wb_push(wb, &n, 1);
 		uint8_t byte = (uint8_t)v;
 		wb_push(wb, &byte, 1);
 	} else if (v<0x10000) {
-		int n = COMBINE_TYPE(type , 2);
+		uint8_t n = COMBINE_TYPE(type , 2);
 		wb_push(wb, &n, 1);
 		uint16_t word = (uint16_t)v;
 		wb_push(wb, &word, 2);
 	} else {
-		int n = COMBINE_TYPE(type , 4);
+		uint8_t n = COMBINE_TYPE(type , 4);
 		wb_push(wb, &n, 1);
 		wb_push(wb, &v, 4);
 	}
@@ -158,14 +158,14 @@ wb_integer(struct write_block *wb, int v, int type) {
 
 static inline void
 wb_number(struct write_block *wb, double v) {
-	int n = COMBINE_TYPE(TYPE_NUMBER , 8);
+	uint8_t n = COMBINE_TYPE(TYPE_NUMBER , 8);
 	wb_push(wb, &n, 1);
 	wb_push(wb, &v, 8);
 }
 
 static inline void
 wb_pointer(struct write_block *wb, void *v) {
-	int n = TYPE_USERDATA;
+	uint8_t n = TYPE_USERDATA;
 	wb_push(wb, &n, 1);
 	wb_push(wb, &v, sizeof(v));
 }
@@ -173,13 +173,13 @@ wb_pointer(struct write_block *wb, void *v) {
 static inline void
 wb_string(struct write_block *wb, const char *str, int len) {
 	if (len < MAX_COOKIE) {
-		int n = COMBINE_TYPE(TYPE_SHORT_STRING, len);
+		uint8_t n = COMBINE_TYPE(TYPE_SHORT_STRING, len);
 		wb_push(wb, &n, 1);
 		if (len > 0) {
 			wb_push(wb, str, len);
 		}
 	} else {
-		int n;
+		uint8_t n;
 		if (len < 0x10000) {
 			n = COMBINE_TYPE(TYPE_LONG_STRING, 2);
 			wb_push(wb, &n, 1);
@@ -201,11 +201,11 @@ static int
 wb_table_array(lua_State *L, struct write_block * wb, int index, int depth) {
 	int array_size = lua_rawlen(L,index);
 	if (array_size >= MAX_COOKIE-1) {
-		int n = COMBINE_TYPE(TYPE_TABLE, MAX_COOKIE-1);
+		uint8_t n = COMBINE_TYPE(TYPE_TABLE, MAX_COOKIE-1);
 		wb_push(wb, &n, 1);
 		wb_integer(wb, array_size,TYPE_NUMBER);
 	} else {
-		int n = COMBINE_TYPE(TYPE_TABLE, array_size);
+		uint8_t n = COMBINE_TYPE(TYPE_TABLE, array_size);
 		wb_push(wb, &n, 1);
 	}
 

lualib-src/lua-socket.c

@@ -368,7 +368,7 @@ lunpack(lua_State *L) {
 	lua_pushinteger(L, message->id);
 	lua_pushinteger(L, message->ud);
 	if (message->buffer == NULL) {
-		lua_pushlstring(L, (char *)(message+1),size - sizeof(*message) - 1);
+		lua_pushlstring(L, (char *)(message+1),size - sizeof(*message));
 	} else {
 		lua_pushlightuserdata(L, message->buffer);
 	}

lualib-src/lua_mysqlaux.c

@@ -0,0 +1,170 @@
+//
+//  lua_mysqlaux.c
+//
+//  Created by changfeng on 6/17/14.
+//  Copyright (c) 2014 changfeng. All rights reserved.
+//
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#include <lua.h>
+#include <lauxlib.h>
+
+static unsigned int num_escape_sql_str(unsigned char *dst, unsigned char *src, size_t size)
+{
+    unsigned int n =0;
+    while (size) {
+        /* the highest bit of all the UTF-8 chars
+         * is always 1 */
+        if ((*src & 0x80) == 0) {
+            switch (*src) {
+                case '\0':
+                case '\b':
+                case '\n':
+                case '\r':
+                case '\t':
+                case 26:  /* \Z */
+                case '\\':
+                case '\'':
+                case '"':
+                    n++;
+                    break;
+                default:
+                    break;
+            }
+        }
+        src++;
+        size--;
+    }
+    return n;
+}
+static unsigned char*
+escape_sql_str(unsigned char *dst, unsigned char *src, size_t size)
+{
+    
+      while (size) {
+        if ((*src & 0x80) == 0) {
+            switch (*src) {
+                case '\0':
+                    *dst++ = '\\';
+                    *dst++ = '0';
+                    break;
+                    
+                case '\b':
+                    *dst++ = '\\';
+                    *dst++ = 'b';
+                    break;
+                    
+                case '\n':
+                    *dst++ = '\\';
+                    *dst++ = 'n';
+                    break;
+                    
+                case '\r':
+                    *dst++ = '\\';
+                    *dst++ = 'r';
+                    break;
+                    
+                case '\t':
+                    *dst++ = '\\';
+                    *dst++ = 't';
+                    break;
+                    
+                case 26:
+                    *dst++ = '\\';
+                    *dst++ = 'Z';
+                    break;
+                    
+                case '\\':
+                    *dst++ = '\\';
+                    *dst++ = '\\';
+                    break;
+                    
+                case '\'':
+                    *dst++ = '\\';
+                    *dst++ = '\'';
+                    break;
+                    
+                case '"':
+                    *dst++ = '\\';
+                    *dst++ = '"';
+                    break;
+                    
+                default:
+                    *dst++ = *src;
+                    break;
+            }
+        } else {
+            *dst++ = *src;
+        }
+        src++;
+        size--;
+    } /* while (size) */
+    
+    return  dst;
+}
+
+
+
+
+static int
+quote_sql_str(lua_State *L)
+{
+    size_t                   len, dlen, escape;
+    unsigned char                  *p;
+    unsigned char                  *src, *dst;
+    
+    if (lua_gettop(L) != 1) {
+        return luaL_error(L, "expecting one argument");
+    }
+    
+    src = (unsigned char *) luaL_checklstring(L, 1, &len);
+    
+    if (len == 0) {
+        dst = (unsigned char *) "''";
+        dlen = sizeof("''") - 1;
+        lua_pushlstring(L, (char *) dst, dlen);
+        return 1;
+    }
+    
+    escape = num_escape_sql_str(NULL, src, len);
+    
+    dlen = sizeof("''") - 1 + len + escape;
+    p = lua_newuserdata(L, dlen);
+    
+    dst = p;
+    
+    *p++ = '\'';
+    
+    if (escape == 0) {
+        memcpy(p, src, len);
+        p+=len;
+    } else {
+        p = (unsigned char *) escape_sql_str(p, src, len);
+    }
+    
+    *p++ = '\'';
+    
+    if (p != dst + dlen) {
+        return luaL_error(L, "quote sql string error");
+    }
+    
+    lua_pushlstring(L, (char *) dst, p - dst);
+    
+    return 1;
+}
+
+
+static struct luaL_Reg mysqlauxlib[] = {
+    {"quote_sql_str",quote_sql_str},
+    {NULL, NULL}
+};
+
+
+int luaopen_mysqlaux_c (lua_State *L) {
+    lua_newtable(L);
+    luaL_setfuncs(L, mysqlauxlib, 0);
+    return 1;
+}
+

lualib-src/sproto/lsproto.c

@@ -19,17 +19,17 @@
 */
 LUALIB_API void luaL_setfuncs (lua_State *L, const luaL_Reg *l, int nup) {
 #ifdef luaL_checkversion
-  luaL_checkversion(L);
+	luaL_checkversion(L);
 #endif
-  luaL_checkstack(L, nup, "too many upvalues");
-  for (; l->name != NULL; l++) {  /* fill the table with given functions */
-    int i;
-    for (i = 0; i < nup; i++)  /* copy upvalues to the top */
-      lua_pushvalue(L, -nup);
-    lua_pushcclosure(L, l->func, nup);  /* closure with those upvalues */
-    lua_setfield(L, -(nup + 2), l->name);
-  }
-  lua_pop(L, nup);  /* remove upvalues */
+	luaL_checkstack(L, nup, "too many upvalues");
+	for (; l->name != NULL; l++) {  /* fill the table with given functions */
+		int i;
+		for (i = 0; i < nup; i++)  /* copy upvalues to the top */
+			lua_pushvalue(L, -nup);
+		lua_pushcclosure(L, l->func, nup);  /* closure with those upvalues */
+		lua_setfield(L, -(nup + 2), l->name);
+	}
+	lua_pop(L, nup);  /* remove upvalues */
 }
 
 #define luaL_newlibtable(L,l) \
@@ -62,18 +62,20 @@ ldeleteproto(lua_State *L) {
 
 static int
 lquerytype(lua_State *L) {
+	const char * type_name;
 	struct sproto *sp = lua_touserdata(L,1);
+	struct sproto_type *st;
 	if (sp == NULL) {
 		return luaL_argerror(L, 1, "Need a sproto object");
 	}
-	const char * typename = luaL_checkstring(L,2);
-	struct sproto_type *st = sproto_type(sp, typename);
+	type_name = luaL_checkstring(L,2);
+	st = sproto_type(sp, type_name);
 	if (st) {
 		lua_pushlightuserdata(L, st);
 		return 1;
 	}
 
-	return luaL_error(L, "type %s not found", typename);
+	return luaL_error(L, "type %s not found", type_name);
 }
 
 struct encode_ud {
@@ -119,9 +121,10 @@ encode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
 	switch (type) {
 	case SPROTO_TINTEGER: {
 		lua_Integer v = luaL_checkinteger(L, -1);
+		lua_Integer vh;
 		lua_pop(L,1);
 		// notice: in lua 5.2, lua_Integer maybe 52bit
-		lua_Integer vh = v >> 31;
+		vh = v >> 31;
 		if (vh == 0 || vh == -1) {
 			*(uint32_t *)value = (uint32_t)v;
 			return 4;
@@ -148,13 +151,14 @@ encode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
 	}
 	case SPROTO_TSTRUCT: {
 		struct encode_ud sub;
+		int r;
 		sub.L = L;
 		sub.st = st;
 		sub.tbl_index = lua_gettop(L);
 		sub.array_tag = NULL;
 		sub.array_index = 0;
 		sub.deep = self->deep + 1;
-		int r = sproto_encode(st, value, length, encode, &sub);
+		r = sproto_encode(st, value, length, encode, &sub);
 		lua_pop(L,1);
 		return r;
 	}
@@ -165,6 +169,7 @@ encode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
 
 static void *
 expand_buffer(lua_State *L, int osz, int nsz) {
+	void *output;
 	do {
 		osz *= 2;
 	} while (osz < nsz);
@@ -172,7 +177,7 @@ expand_buffer(lua_State *L, int osz, int nsz) {
 		luaL_error(L, "object is too large (>%d)", ENCODE_MAXSIZE);
 		return NULL;
 	}
-	void *output = lua_newuserdata(L, osz);
+	output = lua_newuserdata(L, osz);
 	lua_replace(L, lua_upvalueindex(1));
 	lua_pushinteger(L, osz);
 	lua_replace(L, lua_upvalueindex(2));
@@ -188,6 +193,7 @@ expand_buffer(lua_State *L, int osz, int nsz) {
  */
 static int
 lencode(lua_State *L) {
+	struct encode_ud self;
 	void * buffer = lua_touserdata(L, lua_upvalueindex(1));
 	int sz = lua_tointeger(L, lua_upvalueindex(2));
 
@@ -197,7 +203,6 @@ lencode(lua_State *L) {
 	}
 	luaL_checktype(L, 2, LUA_TTABLE);
 	luaL_checkstack(L, ENCODE_DEEPLEVEL + 8, NULL);
-	struct encode_ud self;
 	self.L = L;
 	self.st = st;
 	self.tbl_index = 2;
@@ -261,15 +266,16 @@ decode(void *ud, const char *tagname, int type, int index, struct sproto_type *s
 		break;
 	}
 	case SPROTO_TSTRUCT: {
-		lua_newtable(L);
 		struct decode_ud sub;
+		int r;
+		lua_newtable(L);
 		sub.L = L;
 		sub.result_index = lua_gettop(L);
 		sub.deep = self->deep + 1;
 		sub.array_index = 0;
 		sub.array_tag = NULL;
 
-		int r = sproto_decode(st, value, length, decode, &sub);
+		r = sproto_decode(st, value, length, decode, &sub);
 		if (r < 0 || r != length)
 			return r;
 		lua_settop(L, sub.result_index);
@@ -312,22 +318,25 @@ getbuffer(lua_State *L, int index, size_t *sz) {
 static int
 ldecode(lua_State *L) {
 	struct sproto_type * st = lua_touserdata(L, 1);
+	const void * buffer;
+	struct decode_ud self;
+	size_t sz;
+	int r;
 	if (st == NULL) {
 		return luaL_argerror(L, 1, "Need a sproto_type object");
 	}
-	size_t sz=0;
-	const void * buffer = getbuffer(L, 2, &sz);
+	sz = 0;
+	buffer = getbuffer(L, 2, &sz);
 	if (!lua_istable(L, -1)) {
 		lua_newtable(L);
 	}
 	luaL_checkstack(L, ENCODE_DEEPLEVEL*2 + 8, NULL);
-	struct decode_ud self;
 	self.L = L;
 	self.result_index = lua_gettop(L);
 	self.array_index = 0;
 	self.array_tag = NULL;
 	self.deep = 0;
-	int r = sproto_decode(st, buffer, (int)sz, decode, &self);
+	r = sproto_decode(st, buffer, (int)sz, decode, &self);
 	if (r < 0) {
 		return luaL_error(L, "decode error");
 	}
@@ -359,11 +368,12 @@ lpack(lua_State *L) {
 	// the worst-case space overhead of packing is 2 bytes per 2 KiB of input (256 words = 2KiB).
 	size_t maxsz = (sz + 2047) / 2048 * 2 + sz;
 	void * output = lua_touserdata(L, lua_upvalueindex(1));
+	int bytes;
 	int osz = lua_tointeger(L, lua_upvalueindex(2));
 	if (osz < maxsz) {
 		output = expand_buffer(L, osz, maxsz);
 	}
-	int bytes = sproto_pack(buffer, sz, output, maxsz);
+	bytes = sproto_pack(buffer, sz, output, maxsz);
 	if (bytes > maxsz) {
 		return luaL_error(L, "packing error, return size = %d", bytes);
 	}
@@ -402,14 +412,18 @@ pushfunction_withbuffer(lua_State *L, const char * name, lua_CFunction func) {
 static int
 lprotocol(lua_State *L) {
 	struct sproto * sp = lua_touserdata(L, 1);
+	struct sproto_type * request;
+	struct sproto_type * response;
+	int t;
+	int tag;
 	if (sp == NULL) {
 		return luaL_argerror(L, 1, "Need a sproto_type object");
 	}
-	int t = lua_type(L,2);
-	int tag;
+	t = lua_type(L,2);
 	if (t == LUA_TNUMBER) {
+		const char * name;
 		tag = lua_tointeger(L, 2);
-		const char * name = sproto_protoname(sp, tag);
+		name = sproto_protoname(sp, tag);
 		if (name == NULL)
 			return 0;
 		lua_pushstring(L, name);
@@ -420,13 +434,13 @@ lprotocol(lua_State *L) {
 			return 0;
 		lua_pushinteger(L, tag);
 	}
-	struct sproto_type * request = sproto_protoquery(sp, tag, SPROTO_REQUEST);
+	request = sproto_protoquery(sp, tag, SPROTO_REQUEST);
 	if (request == NULL) {
 		lua_pushnil(L);
 	} else {
 		lua_pushlightuserdata(L, request);
 	}
-	struct sproto_type * response = sproto_protoquery(sp, tag, SPROTO_RESPONSE);
+	response = sproto_protoquery(sp, tag, SPROTO_RESPONSE);
 	if (response == NULL) {
 		lua_pushnil(L);
 	} else {

lualib-src/sproto/sproto.c

@@ -119,8 +119,8 @@ todword(const uint8_t *p) {
 static int
 count_array(const uint8_t * stream) {
 	uint32_t length = todword(stream);
-	stream += SIZEOF_LENGTH;
 	int n = 0;
+	stream += SIZEOF_LENGTH;
 	while (length > 0) {
 		uint32_t nsz;
 		if (length < SIZEOF_LENGTH)
@@ -180,6 +180,10 @@ static const uint8_t *
 import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
 	uint32_t sz;
 	const uint8_t * result;
+	int fn;
+	int i;
+	int array = 0;
+	int tag = -1;
 	f->tag = -1;
 	f->type = -1;
 	f->name = NULL;
@@ -188,13 +192,10 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
 	sz = todword(stream);
 	stream += SIZEOF_LENGTH;
 	result = stream + sz;
-	int fn = struct_field(stream, sz);
+	fn = struct_field(stream, sz);
 	if (fn < 0)
 		return NULL;
 	stream += SIZEOF_HEADER;
-	int i;
-	int array = 0;
-	int tag = -1;
 	for (i=0;i<fn;i++) {
 		int value;
 		++tag;
@@ -203,7 +204,7 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
 			tag+= value/2;
 			continue;
 		}
-		if (tag == 0) {	// name
+		if (tag == 0) { // name
 			if (value != 0)
 				return NULL;
 			f->name = import_string(s, stream + fn * SIZEOF_FIELD);
@@ -213,12 +214,12 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
 			return NULL;
 		value = value/2 - 1;
 		switch(tag) {
-		case 1:	// buildin
+		case 1: // buildin
 			if (value >= SPROTO_TSTRUCT)
 				return NULL;	// invalid buildin type
 			f->type = value;
 			break;
-		case 2:	// type index
+		case 2: // type index
 			if (value >= s->type_n)
 				return NULL;	// invalid type index
 			if (f->type >= 0)
@@ -226,10 +227,10 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
 			f->type = SPROTO_TSTRUCT;
 			f->st = &s->type[value];
 			break;
-		case 3:	// tag
+		case 3: // tag
 			f->tag = value;
 			break;
-		case 4:	// array
+		case 4: // array
 			if (value)
 				array = SPROTO_TARRAY;
 			break;
@@ -248,10 +249,10 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
 .type {
 	.field {
 		name 0 : string
-		buildin	1 :	integer
+		buildin 1 : integer
 		type 2 : integer
-		tag	3 :	integer
-		array 4	: boolean
+		tag 3 : integer
+		array 4 : boolean
 	}
 	name 0 : string
 	fields 1 : *field
@@ -259,11 +260,16 @@ import_field(struct sproto *s, struct field *f, const uint8_t * stream) {
 */
 static const uint8_t *
 import_type(struct sproto *s, struct sproto_type *t, const uint8_t * stream) {
+	const uint8_t * result;
 	uint32_t sz = todword(stream);
 	int i;
+	int fn;
+	int n;
+	int maxn;
+	int last;
 	stream += SIZEOF_LENGTH;
-	const uint8_t * result = stream + sz;
-	int fn = struct_field(stream, sz);
+	result = stream + sz;
+	fn = struct_field(stream, sz);
 	if (fn <= 0 || fn > 2)
 		return NULL;
 	for (i=0;i<fn*SIZEOF_FIELD;i+=SIZEOF_FIELD) {
@@ -279,20 +285,21 @@ import_type(struct sproto *s, struct sproto_type *t, const uint8_t * stream) {
 		return result;
 	}
 	stream += todword(stream)+SIZEOF_LENGTH;	// second data
-	int n = count_array(stream);
+	n = count_array(stream);
 	if (n<0)
 		return NULL;
 	stream += SIZEOF_LENGTH;
-	int maxn = n;
-	int last = -1;
+	maxn = n;
+	last = -1;
 	t->n = n;
 	t->f = pool_alloc(&s->memory, sizeof(struct field) * n);
 	for (i=0;i<n;i++) {
+		int tag;
 		struct field *f = &t->f[i];
 		stream = import_field(s, f, stream);
 		if (stream == NULL)
 			return NULL;
-		int tag = f->tag;
+		tag = f->tag;
 		if (tag < last)
 			return NULL;	// tag must in ascending order
 		if (tag > last+1) {
@@ -312,24 +319,27 @@ import_type(struct sproto *s, struct sproto_type *t, const uint8_t * stream) {
 /*
 .protocol {
 	name 0 : string
-	tag	1 :	integer
-	request	2 :	integer
+	tag 1 : integer
+	request 2 : integer
 	response 3 : integer
 }
 */
 static const uint8_t *
 import_protocol(struct sproto *s, struct protocol *p, const uint8_t * stream) {
+	const uint8_t * result;
 	uint32_t sz = todword(stream);
+	int fn;
+	int i;
+	int tag;
 	stream += SIZEOF_LENGTH;
-	const uint8_t * result = stream + sz;
-	int fn = struct_field(stream, sz);
+	result = stream + sz;
+	fn = struct_field(stream, sz);
 	stream += SIZEOF_HEADER;
 	p->name = NULL;
 	p->tag = -1;
 	p->p[SPROTO_REQUEST] = NULL;
 	p->p[SPROTO_RESPONSE] = NULL;
-	int i;
-	int tag = 0;
+	tag = 0;
 	for (i=0;i<fn;i++,tag++) {
 		int value = toword(stream + SIZEOF_FIELD * i);
 		if (value & 1) {
@@ -338,24 +348,24 @@ import_protocol(struct sproto *s, struct protocol *p, const uint8_t * stream) {
 		}
 		value = value/2 - 1;
 		switch (i) {
-		case 0:	// name
+		case 0: // name
 			if (value != -1) {
 				return NULL;
 			}
 			p->name = import_string(s, stream + SIZEOF_FIELD *fn);
 			break;
-		case 1:	// tag
+		case 1: // tag
 			if (value < 0) {
 				return NULL;
 			}
 			p->tag = value;
 			break;
-		case 2:	// request
+		case 2: // request
 			if (value < 0 || value>=s->type_n)
 				return NULL;
 			p->p[SPROTO_REQUEST] = &s->type[value];
 			break;
-		case 3:	// response
+		case 3: // response
 			if (value < 0 || value>=s->type_n)
 				return NULL;
 			p->p[SPROTO_RESPONSE] = &s->type[value];
@@ -374,22 +384,23 @@ import_protocol(struct sproto *s, struct protocol *p, const uint8_t * stream) {
 
 static struct sproto *
 create_from_bundle(struct sproto *s, const uint8_t * stream, size_t sz) {
+	const uint8_t * content;
+	const uint8_t * typedata = NULL;
+	const uint8_t * protocoldata = NULL;
 	int fn = struct_field(stream, sz);
+	int i;
 	if (fn < 0)
 		return NULL;
 
 	stream += SIZEOF_HEADER;
+	content = stream + fn*SIZEOF_FIELD;
 
-	const uint8_t * content = stream + fn*SIZEOF_FIELD;
-	const uint8_t * typedata = NULL;
-	const uint8_t * protocoldata = NULL;
-
-	int i;
 	for (i=0;i<fn;i++) {
 		int value = toword(stream + i*SIZEOF_FIELD);
+		int n;
 		if (value != 0)
 			return NULL;
-		int n = count_array(content);
+		n = count_array(content);
 		if (n<0)
 			return NULL;
 		if (i == 0) {
@@ -423,8 +434,9 @@ create_from_bundle(struct sproto *s, const uint8_t * stream, size_t sz) {
 struct sproto *
 sproto_create(const void * proto, size_t sz) {
 	struct pool mem;
+	struct sproto * s;
 	pool_init(&mem);
-	struct sproto * s = pool_alloc(&mem, sizeof(*s));
+	s = pool_alloc(&mem, sizeof(*s));
 	if (s == NULL)
 		return NULL;
 	memset(s, 0, sizeof(*s));
@@ -445,34 +457,35 @@ sproto_release(struct sproto * s) {
 
 void
 sproto_dump(struct sproto *s) {
+	int i,j;
 	static const char * buildin[] = {
 		"integer",
 		"boolean",
 		"string",
 	};
 	printf("=== %d types ===\n", s->type_n);
-	int i,j;
 	for (i=0;i<s->type_n;i++) {
 		struct sproto_type *t = &s->type[i];
 		printf("%s\n", t->name);
 		for (j=0;j<t->n;j++) {
 			char array[2] = { 0, 0 };
-			const char * typename = NULL;
+			const char * type_name = NULL;
 			struct field *f = &t->f[j];
 			if (f->type & SPROTO_TARRAY) {
 				array[0] = '*';
 			} else {
 				array[0] = 0;
 			}
-			int t = f->type & ~SPROTO_TARRAY;
-			if (t == SPROTO_TSTRUCT) {
-				typename = f->st->name;
-			} else {
-				assert(t<SPROTO_TSTRUCT);
-				typename = buildin[t];
+			{
+				int t = f->type & ~SPROTO_TARRAY;
+				if (t == SPROTO_TSTRUCT) {
+					type_name = f->st->name;
+				} else {
+					assert(t<SPROTO_TSTRUCT);
+					type_name = buildin[t];
+				}
 			}
-
-			printf("\t%s (%d) %s%s\n", f->name, f->tag, array, typename);
+			printf("\t%s (%d) %s%s\n", f->name, f->tag, array, type_name);
 		}
 	}
 	printf("=== %d protocol ===\n", s->protocol_n);
@@ -518,10 +531,11 @@ query_proto(struct sproto *sp, int tag) {
 
 struct sproto_type * 
 sproto_protoquery(struct sproto *sp, int proto, int what) {
+	struct protocol * p;
 	if (what <0 || what >1) {
 		return NULL;
 	}
-	struct protocol * p = query_proto(sp, proto);
+	p = query_proto(sp, proto);
 	if (p) {
 		return p->p[what];
 	}
@@ -555,13 +569,15 @@ sproto_name(struct sproto_type * st) {
 
 static struct field *
 findtag(struct sproto_type *st, int tag) {
+	int begin, end;
 	if (st->base >=0 ) {
 		tag -= st->base;
 		if (tag < 0 || tag >= st->n)
 			return NULL;
 		return &st->f[tag];
 	}
-	int begin = 0, end = st->n;
+	begin = 0;
+	end = st->n;
 	while (begin < end) {
 		int mid = (begin+end)/2;
 		struct field *f = &st->f[mid];
@@ -580,7 +596,7 @@ findtag(struct sproto_type *st, int tag) {
 
 // encode & decode
 // sproto_callback(void *ud, int tag, int type, struct sproto_type *, void *value, int length)
-//		return size, -1 means error
+//	  return size, -1 means error
 
 static inline int
 fill_size(uint8_t * data, int sz) {
@@ -623,18 +639,20 @@ encode_uint64(uint64_t v, uint8_t * data, int size) {
 
 static int
 encode_string(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) {
+	int sz;
 	if (size < SIZEOF_LENGTH)
 		return -1;
-	int sz = cb(ud, f->name, SPROTO_TSTRING, 0, NULL, data+SIZEOF_LENGTH, size-SIZEOF_LENGTH);
+	sz = cb(ud, f->name, SPROTO_TSTRING, 0, NULL, data+SIZEOF_LENGTH, size-SIZEOF_LENGTH);
 	return fill_size(data, sz);
 }
 
 static int
 encode_struct(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) {
+	int sz;
 	if (size < SIZEOF_LENGTH) {
 		return -1;
 	}
-	int sz = cb(ud, f->name, SPROTO_TSTRUCT, 0, f->st, data+SIZEOF_LENGTH, size-SIZEOF_LENGTH);
+	sz = cb(ud, f->name, SPROTO_TSTRUCT, 0, f->st, data+SIZEOF_LENGTH, size-SIZEOF_LENGTH);
 	return fill_size(data, sz);
 }
 
@@ -656,18 +674,21 @@ uint32_to_uint64(int negative, uint8_t *buffer) {
 static uint8_t *
 encode_integer_array(sproto_callback cb, void *ud, struct field *f, uint8_t *buffer, int size) {
 	uint8_t * header = buffer;
+	int intlen;
+	int index;
 	if (size < 1)
 		return NULL;
 	buffer++;
 	size--;
-	int intlen = sizeof(uint32_t);
-	int index = 1;
+	intlen = sizeof(uint32_t);
+	index = 1;
 	for (;;) {
+		int sz;
 		union {
 			uint64_t u64;
 			uint32_t u32;
 		} u;
-		int sz = cb(ud, f->name, SPROTO_TINTEGER, index, f->st, &u, sizeof(u));
+		sz = cb(ud, f->name, SPROTO_TINTEGER, index, f->st, &u, sizeof(u));
 		if (sz < 0)
 			return NULL;
 		if (sz == 0)
@@ -685,24 +706,26 @@ encode_integer_array(sproto_callback cb, void *ud, struct field *f, uint8_t *buf
 				uint32_to_uint64(v & 0x80000000, buffer);
 			}
 		} else {
+			uint64_t v;
 			if (sz != sizeof(uint64_t))
 				return NULL;
 			if (intlen == sizeof(uint32_t)) {
+				int i;
 				// rearrange
 				size -= (index-1) * sizeof(uint32_t);
 				if (size < sizeof(uint64_t))
 					return NULL;
 				buffer += (index-1) * sizeof(uint32_t);
-				int i;
 				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));
-					int negative = header[1+i*sizeof(uint64_t)+3] & 0x80;
+					negative = header[1+i*sizeof(uint64_t)+3] & 0x80;
 					uint32_to_uint64(negative, header+1+i*sizeof(uint64_t));
 				}
 				intlen = sizeof(uint64_t);
 			}
 
-			uint64_t v = u.u64;
+			v = u.u64;
 			buffer[0] = v & 0xff;
 			buffer[1] = (v >> 8) & 0xff;
 			buffer[2] = (v >> 16) & 0xff;
@@ -727,12 +750,16 @@ encode_integer_array(sproto_callback cb, void *ud, struct field *f, uint8_t *buf
 
 static int
 encode_array(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int size) {
+	uint8_t * buffer;
+	int index;
+	int type;
+	int sz;
 	if (size < SIZEOF_LENGTH)
 		return -1;
 	size -= SIZEOF_LENGTH;
-	int index = 1;
-	uint8_t * buffer = data + SIZEOF_LENGTH;
-	int type = f->type & ~SPROTO_TARRAY;
+	index = 1;
+	buffer = data + SIZEOF_LENGTH;
+	type = f->type & ~SPROTO_TARRAY;
 	switch (type) {
 	case SPROTO_TINTEGER:
 		buffer = encode_integer_array(cb,ud,f,buffer,size);
@@ -757,10 +784,11 @@ encode_array(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int s
 		break;
 	default:
 		for (;;) {
+			int sz;
 			if (size < SIZEOF_LENGTH)
 				return -1;
 			size -= SIZEOF_LENGTH;
-			int sz = cb(ud, f->name, type, index, f->st, buffer+SIZEOF_LENGTH, size);
+			sz = cb(ud, f->name, type, index, f->st, buffer+SIZEOF_LENGTH, size);
 			if (sz < 0)
 				return -1;
 			if (sz == 0)
@@ -772,7 +800,7 @@ encode_array(sproto_callback cb, void *ud, struct field *f, uint8_t *data, int s
 		}
 		break;
 	}
-	int sz = buffer - (data + SIZEOF_LENGTH);
+	sz = buffer - (data + SIZEOF_LENGTH);
 	if (sz == 0)
 		return 0;
 	return fill_size(data, sz);
@@ -783,13 +811,16 @@ sproto_encode(struct sproto_type *st, void * buffer, int size, sproto_callback c
 	uint8_t * header = buffer;
 	uint8_t * data;
 	int header_sz = SIZEOF_HEADER + st->maxn * SIZEOF_FIELD;
+	int i;
+	int index;
+	int lasttag;
+	int datasz;
 	if (size < header_sz)
 		return -1;
 	data = header + header_sz;
 	size -= header_sz;
-	int i;
-	int index = 0;
-	int lasttag = -1;
+	index = 0;
+	lasttag = -1;
 	for (i=0;i<st->n;i++) {
 		struct field *f = &st->f[i];
 		int type = f->type;
@@ -813,7 +844,7 @@ sproto_encode(struct sproto_type *st, void * buffer, int size, sproto_callback c
 				if (sz == sizeof(uint32_t)) {
 					if (u.u32 < 0x7fff) {
 						value = (u.u32+1) * 2;
-						sz = 2;	// sz can be any number > 0
+						sz = 2; // sz can be any number > 0
 					} else {
 						sz = encode_integer(u.u32, data, size);
 					}
@@ -837,12 +868,14 @@ sproto_encode(struct sproto_type *st, void * buffer, int size, sproto_callback c
 		if (sz < 0)
 			return -1;
 		if (sz > 0) {
+			uint8_t * record;
+			int tag;
 			if (value == 0) {
 				data += sz;
 				size -= sz;
 			}
-			uint8_t * record = header+SIZEOF_HEADER+SIZEOF_FIELD*index;
-			int tag = f->tag - lasttag - 1;
+			record = header+SIZEOF_HEADER+SIZEOF_FIELD*index;
+			tag = f->tag - lasttag - 1;
 			if (tag > 0) {
 				// skip tag
 				tag = (tag - 1) * 2 + 1;
@@ -862,7 +895,7 @@ sproto_encode(struct sproto_type *st, void * buffer, int size, sproto_callback c
 	header[0] = index & 0xff;
 	header[1] = (index >> 8) & 0xff;
 
-	int datasz = data - (header + header_sz);
+	datasz = data - (header + header_sz);
 	data = header + header_sz;
 	if (index != st->maxn) {
 		memmove(header + SIZEOF_HEADER + index * SIZEOF_FIELD, data, datasz);
@@ -909,9 +942,10 @@ decode_array(sproto_callback cb, void *ud, struct field *f, uint8_t * stream) {
 	stream += SIZEOF_LENGTH;
 	switch (type) {
 	case SPROTO_TINTEGER: {
+		int len;
 		if (sz < 1)
 			return -1;
-		int len = *stream;
+		len = *stream;
 		++stream;
 		--sz;
 		if (len == sizeof(uint32_t)) {
@@ -956,6 +990,8 @@ sproto_decode(struct sproto_type *st, const void * data, int size, sproto_callba
 	uint8_t * stream;
 	uint8_t * datastream;
 	int fn;
+	int i;
+	int tag;
 	if (size < SIZEOF_HEADER)
 		return -1;
 	stream = (void *)data;
@@ -967,17 +1003,18 @@ sproto_decode(struct sproto_type *st, const void * data, int size, sproto_callba
 	datastream = stream + fn * SIZEOF_FIELD;
 	size -= fn * SIZEOF_FIELD ;
 
-	int i;
-	int tag = -1;
+	tag = -1;
 	for (i=0;i<fn;i++) {
-		++ tag;
+		uint8_t * currentdata;
+		struct field * f;
 		int value = toword(stream + i * SIZEOF_FIELD);
+		++ tag;
 		if (value & 1) {
 			tag += value/2;
 			continue;
 		}
 		value = value/2 - 1;
-		uint8_t * currentdata = datastream;
+		currentdata = datastream;
 		if (value < 0) {
 			uint32_t sz;
 			if (size < SIZEOF_LENGTH)
@@ -988,7 +1025,7 @@ sproto_decode(struct sproto_type *st, const void * data, int size, sproto_callba
 			datastream += sz+SIZEOF_LENGTH;
 			size -= sz+SIZEOF_LENGTH;
 		}
-		struct field * f= findtag(st, tag);
+		f = findtag(st, tag);
 		if (f == NULL)
 			continue;
 		if (value < 0) {
@@ -1076,9 +1113,15 @@ pack_seg(const uint8_t *src, uint8_t * buffer, int sz, int n) {
 
 static inline void
 write_ff(const uint8_t * src, uint8_t * des, int n) {
+	int i;
+	int align8_n = (n+7)&(~7);
+
 	des[0] = 0xff;
-	des[1] = n-1;
-	memcpy(des+2, src, n * 8);
+	des[1] = align8_n/8 - 1;
+	memcpy(des+2, src, n);
+	for(i=0; i< align8_n-n; i++){
+		des[n+2+i] = 0;
+	}
 }
 
 int 
@@ -1092,6 +1135,7 @@ sproto_pack(const void * srcv, int srcsz, void * bufferv, int bufsz) {
 	const uint8_t * src = srcv;
 	uint8_t * buffer = bufferv;
 	for (i=0;i<srcsz;i+=8) {
+		int n;
 		int padding = i+8 - srcsz;
 		if (padding > 0) {
 			int j;
@@ -1101,7 +1145,7 @@ sproto_pack(const void * srcv, int srcsz, void * bufferv, int bufsz) {
 			}
 			src = tmp;
 		}
-		int n = pack_seg(src, buffer, bufsz, ff_n);
+		n = pack_seg(src, buffer, bufsz, ff_n);
 		bufsz -= n;
 		if (n == 10) {
 			// first FF
@@ -1112,14 +1156,14 @@ sproto_pack(const void * srcv, int srcsz, void * bufferv, int bufsz) {
 			++ff_n;
 			if (ff_n == 256) {
 				if (bufsz >= 0) {
-					write_ff(ff_srcstart, ff_desstart, 256);
+					write_ff(ff_srcstart, ff_desstart, 256*8);
 				}
 				ff_n = 0;
 			}
 		} else {
 			if (ff_n > 0) {
 				if (bufsz >= 0) {
-					write_ff(ff_srcstart, ff_desstart, ff_n);
+					write_ff(ff_srcstart, ff_desstart, ff_n*8);
 				}
 				ff_n = 0;
 			}
@@ -1128,8 +1172,11 @@ sproto_pack(const void * srcv, int srcsz, void * bufferv, int bufsz) {
 		buffer += n;
 		size += n;
 	}
-	if (ff_n > 0 && bufsz >= 0) {
-		write_ff(ff_srcstart, ff_desstart, ff_n);
+	if(bufsz >= 0){
+		if(ff_n == 1)
+			write_ff(ff_srcstart, ff_desstart, 8);
+		else if (ff_n > 1)
+			write_ff(ff_srcstart, ff_desstart, srcsz - (intptr_t)(ff_srcstart - (const uint8_t*)srcv));
 	}
 	return size;
 }
@@ -1144,10 +1191,11 @@ sproto_unpack(const void * srcv, int srcsz, void * bufferv, int bufsz) {
 		--srcsz;
 		++src;
 		if (header == 0xff) {
+			int n;
 			if (srcsz < 0) {
 				return -1;
 			}
-			int n = (src[0] + 1) * 8;
+			n = (src[0] + 1) * 8;
 			if (srcsz < n + 1)
 				return -1;
 			srcsz -= n + 1;

lualib/http/httpc.lua

@@ -81,10 +81,10 @@ function httpc.request(method, host, url, recvheader, header, content)
 	end
 	local fd = socket.connect(hostname, port)
 	local ok , statuscode, body = pcall(request, fd,method, host, url, recvheader, header, content)
+	socket.close(fd)
 	if ok then
 		return statuscode, body
 	else
-		socket.close(fd)
 		error(statuscode)
 	end
 end

lualib/http/httpd.lua

@@ -83,7 +83,7 @@ local function readall(readbytes, bodylimit)
 	else
 		-- identity mode
 		if length then
-			if length > bodylimit then
+			if bodylimit and length > bodylimit then
 				return 413
 			end
 			if #body >= length then
@@ -130,6 +130,7 @@ local function writeall(writefunc, statuscode, bodyfunc, header)
 				end
 			else
 				writefunc("\r\n0\r\n\r\n")
+				break
 			end
 		end
 	else

lualib/mongo.lua

@@ -129,6 +129,7 @@ function mongo.client( conf	)
 		response = dispatch_reply,
 		auth = mongo_auth(obj),
 		backup = backup,
+		nodelay = true,
 	}
 	setmetatable(obj, client_meta)
 	obj.__sock:connect(true)	-- try connect only	once

lualib/mysql.lua

@@ -0,0 +1,747 @@
+-- Copyright (C) 2012 Yichun Zhang (agentzh)
+-- Copyright (C) 2014 Chang Feng
+-- This file is modified version from https://github.com/openresty/lua-resty-mysql
+-- The license is under the BSD license.
+
+local socketchannel = require "socketchannel"
+local bit = require "bit32"
+local mysqlaux = require "mysqlaux.c"
+local crypt = require "crypt"
+
+
+local sub = string.sub
+local strbyte = string.byte
+local strchar = string.char
+local strfind = string.find
+local strrep = string.rep
+local null = nil
+local band = bit.band
+local bxor = bit.bxor
+local bor = bit.bor
+local lshift = bit.lshift
+local rshift = bit.rshift
+local sha1= crypt.sha1
+local concat = table.concat
+local unpack = unpack
+local setmetatable = setmetatable
+local error = error
+local tonumber = tonumber
+local    new_tab = function (narr, nrec) return {} end
+
+
+local _M = { _VERSION = '0.13' }
+-- 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)
+
+for i = 0x01, 0x05 do
+    -- tiny, short, long, float, double
+    converters[i] = tonumber
+end
+-- converters[0x08] = tonumber  -- long long
+converters[0x09] = tonumber  -- int24
+converters[0x0d] = tonumber  -- year
+converters[0xf6] = tonumber  -- newdecimal
+
+
+local function _get_byte2(data, i)
+    local a, b = strbyte(data, i, i + 1)
+    return bor(a, lshift(b, 8)), i + 2
+end
+
+
+local function _get_byte3(data, i)
+    local a, b, c = strbyte(data, i, i + 2)
+    return bor(a, lshift(b, 8), lshift(c, 16)), i + 3
+end
+
+
+local function _get_byte4(data, i)
+    local a, b, c, d = strbyte(data, i, i + 3)
+    return bor(a, lshift(b, 8), lshift(c, 16), lshift(d, 24)), i + 4
+end
+
+
+local function _get_byte8(data, i)
+    local a, b, c, d, e, f, g, h = strbyte(data, i, i + 7)
+
+    -- XXX workaround for the lack of 64-bit support in bitop:
+    local lo = bor(a, lshift(b, 8), lshift(c, 16), lshift(d, 24))
+    local hi = bor(e, lshift(f, 8), lshift(g, 16), lshift(h, 24))
+    return lo + hi * 4294967296, i + 8
+
+    -- return bor(a, lshift(b, 8), lshift(c, 16), lshift(d, 24), lshift(e, 32),
+               -- lshift(f, 40), lshift(g, 48), lshift(h, 56)), i + 8
+end
+
+
+local function _set_byte2(n)
+    return strchar(band(n, 0xff), band(rshift(n, 8), 0xff))
+end
+
+
+local function _set_byte3(n)
+    return strchar(band(n, 0xff),
+                   band(rshift(n, 8), 0xff),
+                   band(rshift(n, 16), 0xff))
+end
+
+
+local function _set_byte4(n)
+    return strchar(band(n, 0xff),
+                   band(rshift(n, 8), 0xff),
+                   band(rshift(n, 16), 0xff),
+                   band(rshift(n, 24), 0xff))
+end
+
+
+local function _from_cstring(data, i)
+    local last = strfind(data, "\0", i, true)
+    if not last then
+        return nil, nil
+    end
+
+    return sub(data, i, last), last + 1
+end
+
+
+local function _to_cstring(data)
+    return data .. "\0"
+end
+
+
+local function _to_binary_coded_string(data)
+    return strchar(#data) .. data
+end
+
+
+local function _dump(data)
+    local len = #data
+    local bytes = new_tab(len, 0)
+    for i = 1, len do
+        bytes[i] = strbyte(data, i)
+    end
+    return concat(bytes, " ")
+end
+
+
+
+local function _dumphex(bytes)
+    local result ={}
+
+    for i = 1, string.len(bytes) do
+        local charcode = tonumber(strbyte(bytes, i, i))
+        local hexstr = string.format("%02X", charcode)
+        result[i]=hexstr
+    end
+
+    local res=table.concat(result, " ")
+    return res
+end
+
+
+local function _compute_token(password, scramble)
+    if password == "" then
+        return ""
+    end
+    --_dump(scramble)
+
+    local stage1 = sha1(password)
+    --print("stage1:", _dumphex(stage1) )
+    local stage2 = sha1(stage1)
+    local stage3 = sha1(scramble .. stage2)
+    local n = #stage1
+    local bytes = new_tab(n, 0)
+    for i = 1, n do
+         bytes[i] = strchar(bxor(strbyte(stage3, i), strbyte(stage1, i)))
+    end
+
+    return concat(bytes)
+end
+
+local function _compose_packet(self, req, size)
+   self.packet_no = self.packet_no + 1
+
+    local packet = _set_byte3(size) .. strchar(self.packet_no) .. req
+    return packet
+end
+
+
+local function _send_packet(self, req, size)
+    local sock = self.sock
+
+    self.packet_no = self.packet_no + 1
+
+
+    local packet = _set_byte3(size) .. strchar(self.packet_no) .. req
+
+    return socket.write(self.sock,packet)
+end
+
+
+local function _recv_packet(self,sock)
+
+
+    local data = sock:read( 4)
+    if not data then
+        return nil, nil, "failed to receive packet header: "
+    end
+
+
+    local len, pos = _get_byte3(data, 1)
+
+
+    if len == 0 then
+        return nil, nil, "empty packet"
+    end
+
+    if len > self._max_packet_size then
+        return nil, nil, "packet size too big: " .. len
+    end
+
+    local num = strbyte(data, pos)
+
+    self.packet_no = num
+
+    data = sock:read(len)
+
+    if not data then
+        return nil, nil, "failed to read packet content: "
+    end
+
+
+    local field_count = strbyte(data, 1)
+    local typ
+    if field_count == 0x00 then
+        typ = "OK"
+    elseif field_count == 0xff then
+        typ = "ERR"
+    elseif field_count == 0xfe then
+        typ = "EOF"
+    elseif field_count <= 250 then
+        typ = "DATA"
+    end
+
+    return data, typ
+end
+
+
+local function _from_length_coded_bin(data, pos)
+    local first = strbyte(data, pos)
+
+    if not first then
+        return nil, pos
+    end
+
+    if first >= 0 and first <= 250 then
+        return first, pos + 1
+    end
+
+    if first == 251 then
+        return null, pos + 1
+    end
+
+    if first == 252 then
+        pos = pos + 1
+        return _get_byte2(data, pos)
+    end
+
+    if first == 253 then
+        pos = pos + 1
+        return _get_byte3(data, pos)
+    end
+
+    if first == 254 then
+        pos = pos + 1
+        return _get_byte8(data, pos)
+    end
+
+    return false, pos + 1
+end
+
+
+local function _from_length_coded_str(data, pos)
+    local len
+    len, pos = _from_length_coded_bin(data, pos)
+    if len == nil or len == null then
+        return null, pos
+    end
+
+    return sub(data, pos, pos + len - 1), pos + len
+end
+
+
+local function _parse_ok_packet(packet)
+    local res = new_tab(0, 5)
+    local pos
+
+    res.affected_rows, pos = _from_length_coded_bin(packet, 2)
+
+    res.insert_id, pos = _from_length_coded_bin(packet, pos)
+
+    res.server_status, pos = _get_byte2(packet, pos)
+
+    res.warning_count, pos = _get_byte2(packet, pos)
+
+
+    local message = sub(packet, pos)
+    if message and message ~= "" then
+        res.message = message
+    end
+
+
+    return res
+end
+
+
+local function _parse_eof_packet(packet)
+    local pos = 2
+
+    local warning_count, pos = _get_byte2(packet, pos)
+    local status_flags = _get_byte2(packet, pos)
+
+    return warning_count, status_flags
+end
+
+
+local function _parse_err_packet(packet)
+    local errno, pos = _get_byte2(packet, 2)
+    local marker = sub(packet, pos, pos)
+    local sqlstate
+    if marker == '#' then
+        -- with sqlstate
+        pos = pos + 1
+        sqlstate = sub(packet, pos, pos + 5 - 1)
+        pos = pos + 5
+    end
+
+    local message = sub(packet, pos)
+    return errno, message, sqlstate
+end
+
+
+local function _parse_result_set_header_packet(packet)
+    local field_count, pos = _from_length_coded_bin(packet, 1)
+
+    local extra
+    extra = _from_length_coded_bin(packet, pos)
+
+    return field_count, extra
+end
+
+
+local function _parse_field_packet(data)
+    local col = new_tab(0, 2)
+    local catalog, db, table, orig_table, orig_name, charsetnr, length
+    local pos
+    catalog, pos = _from_length_coded_str(data, 1)
+
+
+    db, pos = _from_length_coded_str(data, pos)
+    table, pos = _from_length_coded_str(data, pos)
+    orig_table, pos = _from_length_coded_str(data, pos)
+    col.name, pos = _from_length_coded_str(data, pos)
+
+    orig_name, pos = _from_length_coded_str(data, pos)
+
+    pos = pos + 1 -- ignore the filler
+
+    charsetnr, pos = _get_byte2(data, pos)
+
+    length, pos = _get_byte4(data, pos)
+
+    col.type = strbyte(data, pos)
+
+    --[[
+    pos = pos + 1
+
+    col.flags, pos = _get_byte2(data, pos)
+
+    col.decimals = strbyte(data, pos)
+    pos = pos + 1
+
+    local default = sub(data, pos + 2)
+    if default and default ~= "" then
+        col.default = default
+    end
+    --]]
+
+    return col
+end
+
+
+local function _parse_row_data_packet(data, cols, compact)
+    local pos = 1
+    local ncols = #cols
+    local row
+    if compact then
+        row = new_tab(ncols, 0)
+    else
+        row = new_tab(0, ncols)
+    end
+    for i = 1, ncols do
+        local value
+        value, pos = _from_length_coded_str(data, pos)
+        local col = cols[i]
+        local typ = col.type
+        local name = col.name
+
+        if value ~= null then
+            local conv = converters[typ]
+            if conv then
+                value = conv(value)
+            end
+        end
+
+        if compact then
+            row[i] = value
+
+        else
+            row[name] = value
+        end
+    end
+
+    return row
+end
+
+
+local function _recv_field_packet(self, sock)
+    local packet, typ, err = _recv_packet(self, sock)
+    if not packet then
+        return nil, err
+    end
+
+    if typ == "ERR" then
+        local errno, msg, sqlstate = _parse_err_packet(packet)
+        return nil, msg, errno, sqlstate
+    end
+
+    if typ ~= 'DATA' then
+        return nil, "bad field packet type: " .. typ
+    end
+
+    -- typ == 'DATA'
+
+    return _parse_field_packet(packet)
+end
+
+local function _recv_decode_packet_resp(self)
+     return function(sock)
+        return true, _recv_packet(self,sock)
+    end
+end
+
+local function _recv_auth_resp(self)
+     return function(sock)
+        local packet, typ, err = _recv_packet(self,sock)
+        if not packet then
+            --print("recv auth resp : failed to receive the result packet")
+            error ("failed to receive the result packet"..err)
+            --return nil,err
+        end
+
+        if typ == 'ERR' then
+            local errno, msg, sqlstate = _parse_err_packet(packet)
+            error( string.format("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
+            --return nil, errno,msg, sqlstate
+        end
+
+        if typ == 'EOF' then
+            error "old pre-4.1 authentication protocol not supported"
+        end
+
+        if typ ~= 'OK' then
+            error "bad packet type: "
+        end
+        return true, true
+    end
+end
+
+
+local function _mysql_login(self,user,password,database)
+
+    return function(sockchannel)
+          local packet, typ, err =   sockchannel:response( _recv_decode_packet_resp(self) )
+        --local aat={}
+        if not packet then
+            error(  err )
+        end
+
+        if typ == "ERR" then
+            local errno, msg, sqlstate = _parse_err_packet(packet)
+            error( string.format("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
+        end
+
+        self.protocol_ver = strbyte(packet)
+
+        local server_ver, pos = _from_cstring(packet, 2)
+        if not server_ver then
+            error "bad handshake initialization packet: bad server version"
+        end
+
+        self._server_ver = server_ver
+
+
+        local thread_id, pos = _get_byte4(packet, pos)
+
+        local scramble1 = sub(packet, pos, pos + 8 - 1)
+        if not scramble1 then
+            error "1st part of scramble not found"
+        end
+
+        pos = pos + 9 -- skip filler
+
+        -- two lower bytes
+        self._server_capabilities, pos = _get_byte2(packet, pos)
+
+        self._server_lang = strbyte(packet, pos)
+        pos = pos + 1
+
+        self._server_status, pos = _get_byte2(packet, pos)
+
+        local more_capabilities
+        more_capabilities, pos = _get_byte2(packet, pos)
+
+        self._server_capabilities = bor(self._server_capabilities,
+                                        lshift(more_capabilities, 16))
+
+
+        local len = 21 - 8 - 1
+
+        pos = pos + 1 + 10
+
+        local scramble_part2 = sub(packet, pos, pos + len - 1)
+        if not scramble_part2 then
+            error "2nd part of scramble not found"
+        end
+
+
+        local scramble = scramble1..scramble_part2
+        local token = _compute_token(password, scramble)
+
+        local client_flags = 260047;
+
+        local req = _set_byte4(client_flags)
+                    .. _set_byte4(self._max_packet_size)
+                    .. "\0" -- TODO: add support for charset encoding
+                    .. strrep("\0", 23)
+                    .. _to_cstring(user)
+                    .. _to_binary_coded_string(token)
+                    .. _to_cstring(database)
+
+        local packet_len = 4 + 4 + 1 + 23 + #user + 1
+            + #token + 1 + #database + 1
+
+        local authpacket=_compose_packet(self,req,packet_len)
+        return sockchannel:request(authpacket,_recv_auth_resp(self))
+    end
+end
+
+
+local function _compose_query(self, query)
+
+    self.packet_no = -1
+
+    local cmd_packet = strchar(COM_QUERY) .. query
+    local packet_len = 1 + #query
+
+    local querypacket = _compose_packet(self, cmd_packet, packet_len)
+    return querypacket
+end
+
+
+
+local function read_result(self, sock)
+    local packet, typ, err = _recv_packet(self, sock)
+    if not packet then
+        return nil, err
+        --error( err )
+    end
+
+    if typ == "ERR" then
+        local errno, msg, sqlstate = _parse_err_packet(packet)
+        return nil, msg, errno, sqlstate
+        --error( string.format("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
+    end
+
+    if typ == 'OK' then
+        local res = _parse_ok_packet(packet)
+        if res and band(res.server_status, SERVER_MORE_RESULTS_EXISTS) ~= 0 then
+            return res, "again"
+        end
+        return res
+    end
+
+    if typ ~= 'DATA' then
+        return nil, "packet type " .. typ .. " not supported"
+        --error( "packet type " .. typ .. " not supported" )
+    end
+
+    -- typ == 'DATA'
+
+    local field_count, extra = _parse_result_set_header_packet(packet)
+
+    local cols = new_tab(field_count, 0)
+    for i = 1, field_count do
+        local col, err, errno, sqlstate = _recv_field_packet(self, sock)
+        if not col then
+            return nil, err, errno, sqlstate
+            --error( string.format("errno:%d, msg:%s,sqlstate:%s",errno,msg,sqlstate))
+        end
+
+        cols[i] = col
+    end
+
+    local packet, typ, err = _recv_packet(self, sock)
+    if not packet then
+        --error( err)
+        return nil, err
+    end
+
+    if typ ~= 'EOF' then
+        --error ( "unexpected packet type " .. typ .. " while eof packet is ".. "expected" )
+        return nil, "unexpected packet type " .. typ .. " while eof packet is ".. "expected"
+    end
+
+    -- typ == 'EOF'
+
+    local compact = self.compact
+
+    local rows = new_tab( 4, 0)
+    local i = 0
+    while true do
+        packet, typ, err = _recv_packet(self, sock)
+        if not packet then
+            --error (err)
+            return nil, err
+        end
+
+        if typ == 'EOF' then
+            local warning_count, status_flags = _parse_eof_packet(packet)
+
+            if band(status_flags, SERVER_MORE_RESULTS_EXISTS) ~= 0 then
+                return rows, "again"
+            end
+
+            break
+        end
+
+        -- if typ ~= 'DATA' then
+            -- return nil, 'bad row packet type: ' .. typ
+        -- end
+
+        -- typ == 'DATA'
+
+        local row = _parse_row_data_packet(packet, cols, compact)
+        i = i + 1
+        rows[i] = row
+    end
+
+    return rows
+end
+
+local function _query_resp(self)
+     return function(sock)
+        local res, err, errno, sqlstate = read_result(self,sock)
+        if not res then
+            local badresult ={}
+            badresult.badresult = true
+            badresult.err = err
+            badresult.errno = errno
+            badresult.sqlstate = sqlstate
+            return true , badresult
+        end
+        if err ~= "again" then
+            return true, res
+        end
+        local mulitresultset = {res}
+        mulitresultset.mulitresultset = true
+        local i =2
+        while err =="again" do
+            res, err, errno, sqlstate = read_result(self,sock)
+            if not res then
+                return true, mulitresultset
+            end
+            mulitresultset[i]=res
+            i=i+1
+        end
+        return true, mulitresultset
+    end
+end
+
+function _M.connect( opts)
+
+    local self = setmetatable( {}, mt)
+
+    local max_packet_size = opts.max_packet_size
+    if not max_packet_size then
+        max_packet_size = 1024 * 1024 -- default 1 MB
+    end
+    self._max_packet_size = max_packet_size
+    self.compact = opts.compact_arrays
+
+
+    local database = opts.database or ""
+    local user = opts.user or ""
+    local password = opts.password or ""
+
+    local channel = socketchannel.channel {
+        host = opts.host,
+        port = opts.port or 3306,
+        auth = _mysql_login(self,user,password,database ),
+    }
+    -- try connect first only once
+    channel:connect(true)
+    self.sockchannel = channel
+
+
+    return self
+end
+
+
+
+function _M.disconnect(self)
+    self.sockchannel:close()
+    setmetatable(self, nil)
+end
+
+
+function _M.query(self, query)
+    local querypacket = _compose_query(self, query)
+    local sockchannel = self.sockchannel
+    if not self.query_resp then
+        self.query_resp = _query_resp(self)
+    end
+    return  sockchannel:request( querypacket, self.query_resp )
+end
+
+function _M.server_ver(self)
+    return self._server_ver
+end
+
+
+function _M.quote_sql_str( str)
+    return mysqlaux.quote_sql_str(str)
+end
+
+function _M.set_compact_arrays(self, value)
+    self.compact = value
+end
+
+
+return _M

lualib/redis.lua

@@ -84,6 +84,7 @@ function redis.connect(db_conf)
 		host = db_conf.host,
 		port = db_conf.port or 6379,
 		auth = redis_login(db_conf.auth, db_conf.db),
+		nodelay = true,
 	}
 	-- try connect first only once
 	channel:connect(true)
@@ -199,6 +200,7 @@ function redis.watch(db_conf)
 		host = db_conf.host,
 		port = db_conf.port or 6379,
 		auth = watch_login(obj, db_conf.auth),
+		nodelay = true,
 	}
 	obj.__sock = channel
 

lualib/skynet.lua

@@ -709,8 +709,15 @@ function skynet.term(service)
 	return _error_dispatch(0, service)
 end
 
+local function clear_pool()
+	coroutine_pool = {}
+end
+
 -- Inject internal debug framework
 local debug = require "skynet.debug"
-debug(skynet, dispatch_message)
+debug(skynet, {
+	dispatch = dispatch_message,
+	clear = clear_pool,
+})
 
 return skynet

lualib/skynet/debug.lua

@@ -2,7 +2,7 @@ local io = io
 local table = table
 local debug = debug
 
-return function (skynet, dispatch_func)
+return function (skynet, export)
 
 local internal_info_func
 
@@ -18,7 +18,7 @@ function dbgcmd.MEM()
 end
 
 function dbgcmd.GC()
-	coroutine_pool = {}
+	export.clear()
 	collectgarbage "collect"
 end
 
@@ -49,7 +49,7 @@ end
 
 function dbgcmd.RUN(source, filename)
 	local inject = require "skynet.inject"
-	local output = inject(source, filename , dispatch_func, skynet.register_protocol)
+	local output = inject(source, filename , export.dispatch, skynet.register_protocol)
 	collectgarbage "collect"
 	skynet.ret(skynet.pack(table.concat(output, "\n")))
 end

lualib/skynet/queue.lua

@@ -1,6 +1,7 @@
 local skynet = require "skynet"
 local coroutine = coroutine
-local pcall = pcall
+local xpcall = xpcall
+local traceback = debug.traceback
 local table = table
 
 function skynet.queue()
@@ -9,25 +10,24 @@ function skynet.queue()
 	local thread_queue = {}
 	return function(f, ...)
 		local thread = coroutine.running()
-		if ref == 0 then
-			current_thread = thread
-		elseif current_thread ~= thread then
+		if current_thread and current_thread ~= thread then
 			table.insert(thread_queue, thread)
 			skynet.wait()
-			assert(ref == 0)
+			assert(ref == 0)	-- current_thread == thread
 		end
+		current_thread = thread
+
 		ref = ref + 1
-		local ok, err = pcall(f, ...)
+		local ok, err = xpcall(f, traceback, ...)
 		ref = ref - 1
 		if ref == 0 then
-			current_thread = nil
-			local co = table.remove(thread_queue,1)
-			if co then
-				skynet.wakeup(co)
+			current_thread = table.remove(thread_queue,1)
+			if current_thread then
+				skynet.wakeup(current_thread)
 			end
 		end
 		assert(ok,err)
 	end
 end
 
-return skynet.queue
+return skynet.queue

lualib/snax.lua

@@ -132,6 +132,14 @@ function snax.kill(obj, ...)
 	skynet_call(obj.handle, "snax", t.system.exit, ...)
 end
 
+function snax.self()
+	return snax.bind(skynet.self(), SERVICE_NAME)
+end
+
+function snax.exit(...)
+	snax.kill(snax.self(), ...)
+end
+
 local function test_result(ok, ...)
 	if ok then
 		return ...

lualib/snax/gateserver.lua

@@ -53,20 +53,32 @@ function gateserver.start(handler)
 
 	local MSG = {}
 
-	function MSG.data(fd, msg, sz)
+	local function dispatch_msg(fd, msg, sz)
 		if connection[fd] then
 			handler.message(fd, msg, sz)
+		else
+			skynet.error(string.format("Drop message from fd (%d) : %s", fd, netpack.tostring(msg,sz)))
 		end
 	end
 
-	function MSG.more()
-		for fd, msg, sz in netpack.pop, queue do
-			if connection[fd] then
-				handler.message(fd, msg, sz)
+	MSG.data = dispatch_msg
+
+	local function dispatch_queue()
+		local fd, msg, sz = netpack.pop(queue)
+		if fd then
+			-- may dispatch even the handler.message blocked
+			-- If the handler.message never block, the queue should be empty, so only fork once and then exit.
+			skynet.fork(dispatch_queue)
+			dispatch_msg(fd, msg, sz)
+
+			for fd, msg, sz in netpack.pop, queue do
+				dispatch_msg(fd, msg, sz)
 			end
 		end
 	end
 
+	MSG.more = dispatch_queue
+
 	function MSG.open(fd, msg)
 		if client_number >= maxclient then
 			socketdriver.close(fd)
@@ -128,4 +140,4 @@ function gateserver.start(handler)
 	end)
 end
 
-return gateserver
+return gateserver

lualib/snax/loginserver.lua

@@ -90,7 +90,7 @@ local function launch_slave(auth_handler)
 	local function ret_pack(ok, err, ...)
 		if ok then
 			skynet.ret(skynet.pack(err, ...))
-		elseif err ~= socket_error then
+		else
 			error(err)
 		end
 	end
@@ -163,6 +163,7 @@ local function launch_master(conf)
 			if err ~= socket_error then
 				skynet.error(string.format("invalid client (fd = %d) error = %s", fd, err))
 			end
+			socket.start(fd)
 		end
 		socket.close(fd)
 	end)

lualib/socketchannel.lua

@@ -1,5 +1,6 @@
 local skynet = require "skynet"
 local socket = require "socket"
+local socketdriver = require "socketdriver"
 
 -- channel support auto reconnect , and capture socket error in request/response transaction
 -- { host = "", port = , auth = function(so) , response = function(so) session, data }
@@ -37,6 +38,7 @@ function socket_channel.channel(desc)
 		__sock = false,
 		__closed = false,
 		__authcoroutine = false,
+		__nodelay = desc.nodelay,
 	}
 
 	return setmetatable(c, channel_meta)
@@ -133,7 +135,7 @@ local function dispatch_by_order(self)
 			else
 				close_channel_socket(self)
 				local errmsg
-				if result ~= socket_error then
+				if result_ok ~= socket_error then
 					errmsg = result_ok
 				end
 				self.__result[co] = socket_error
@@ -186,6 +188,9 @@ local function connect_once(self)
 			return false
 		end
 	end
+	if self.__nodelay then
+		socketdriver.nodelay(fd)
+	end
 
 	self.__sock = setmetatable( {fd} , channel_socket_meta )
 	skynet.fork(dispatch_function(self), self)

service-src/service_harbor.c

@@ -316,6 +316,7 @@ forward_local_messsage(struct harbor *h, void *msg, int sz) {
 	destination = (destination & HANDLE_MASK) | ((uint32_t)h->id << HANDLE_REMOTE_SHIFT);
 
 	if (skynet_send(h->ctx, header.source, destination, type, (int)header.session, (void *)msg, sz-HEADER_COOKIE_LENGTH) < 0) {
+		skynet_send(h->ctx, destination, header.source , PTYPE_ERROR, (int)header.session, NULL, 0);
 		skynet_error(h->ctx, "Unknown destination :%x from :%x", destination, header.source);
 	}
 }

service/snaxd.lua

@@ -12,10 +12,6 @@ package.path = snax_path .. "?.lua;" .. package.path
 SERVICE_NAME = snax_name
 SERVICE_PATH = snax_path
 
-local mode
-local thread_id
-local message_queue = {}
-local init = false
 local profile_table = {}
 
 local function update_stat(name, ti)
@@ -38,57 +34,6 @@ local function dispatch(f, ...)
 	return skynet.pack(f(...))
 end
 
-local function message_dispatch()
-	while true do
-		if #message_queue==0 then
-			thread_id = coroutine.running()
-			skynet.wait()
-		else
-			local msg = table.remove(message_queue,1)
-			local method = msg.method
-			local f = method[4]
-			if f then
-				if method[2] == "accept" then
-					-- no return
-					profile.start()
-					local ok, data = xpcall(f, traceback, table.unpack(msg))
-					local ti = profile.stop()
-					update_stat(method[3], ti)
-					if not ok then
-						print(string.format("Error on [:%x] to [:%x] %s", msg.source, skynet.self(), tostring(data)))
-					end
-				else
-					profile.start()
-					local ok, data, size = xpcall(dispatch, traceback, f, table.unpack(msg))
-					local ti = profile.stop()
-					update_stat(method[3], ti)
-					if ok then
-						-- skynet.PTYPE_RESPONSE == 1
-						c.send(msg.source, 1, msg.session, data, size)
-						if method[2] == "system" then
-							init = false
-							skynet.exit()
-							break
-						end
-					else
-						-- Can't throw error, so print it directly
-						print(string.format("Error on [:%x] to [:%x] %s", msg.source, skynet.self(), tostring(data)))
-						c.send(msg.source, skynet.PTYPE_ERROR, msg.session, "")
-					end
-				end
-			end
-		end
-	end
-end
-
-local function queue( session, source, method, ...)
-	table.insert(message_queue, {session = session, source = source, method = method, ... })
-	if thread_id then
-		skynet.wakeup(thread_id)
-		thread_id = nil
-	end
-end
-
 local function return_f(f, ...)
 	return skynet.ret(skynet.pack(f(...)))
 end
@@ -108,6 +53,7 @@ local function timing( method, ... )
 end
 
 skynet.start(function()
+	local init = false
 	skynet.dispatch("snax", function ( session , source , id, ...)
 		local method = func[id]
 
@@ -119,17 +65,12 @@ skynet.start(function()
 			elseif command == "init" then
 				assert(not init, "Already init")
 				local initfunc = method[4] or function() end
-				mode = initfunc(...)
-				if mode == "queue" then
-					skynet.fork(message_dispatch)
-				end
+				initfunc(...)
 				skynet.ret()
 				skynet.info_func(function()
 					return profile_table
 				end)
 				init = true
-			elseif mode == "queue" then
-				queue( session, source, method , ...)
 			else
 				assert(init, "Never init")
 				assert(command == "exit")
@@ -141,11 +82,7 @@ skynet.start(function()
 			end
 		else
 			assert(init, "Init first")
-			if mode == "queue" then
-				queue(session, source, method , ...)
-			else
-				timing(method, ...)
-			end
+			timing(method, ...)
 		end
 	end)
 end)

skynet-src/skynet_server.c

@@ -169,7 +169,11 @@ skynet_context_new(const char * name, const char *param) {
 int
 skynet_context_newsession(struct skynet_context *ctx) {
 	// session always be a positive number
-	int session = (++ctx->session_id) & 0x7fffffff;
+	int session = ++ctx->session_id;
+	if (session <= 0) {
+		ctx->session_id = 1;
+		return 1;
+	}
 	return session;
 }
 
@@ -691,7 +695,7 @@ skynet_sendname(struct skynet_context * context, uint32_t source, const char * a
 			if (type & PTYPE_TAG_DONTCOPY) {
 				skynet_free(data);
 			}
-			return session;
+			return -1;
 		}
 	} else {
 		_filter_args(context, type, &session, (void **)&data, &sz);

skynet-src/skynet_socket.c

@@ -36,10 +36,9 @@ forward_message(int type, bool padding, struct socket_message * result) {
 	int sz = sizeof(*sm);
 	if (padding) {
 		if (result->data) {
-			sz += strlen(result->data) + 1;
+			sz += strlen(result->data);
 		} else {
 			result->data = "";
-			sz += 1;
 		}
 	}
 	sm = (struct skynet_socket_message *)skynet_malloc(sz);
@@ -48,7 +47,7 @@ forward_message(int type, bool padding, struct socket_message * result) {
 	sm->ud = result->ud;
 	if (padding) {
 		sm->buffer = NULL;
-		strcpy((char*)(sm+1), result->data);
+		memcpy(sm+1, result->data, sz - sizeof(*sm));
 	} else {
 		sm->buffer = result->data;
 	}
@@ -62,6 +61,7 @@ forward_message(int type, bool padding, struct socket_message * result) {
 	if (skynet_context_push((uint32_t)result->opaque, &message)) {
 		// todo: report somewhere to close socket
 		// don't call skynet_socket_close here (It will block mainloop)
+		skynet_free(sm->buffer);
 		skynet_free(sm);
 	}
 }

skynet-src/socket_server.c

@@ -40,8 +40,9 @@
 struct write_buffer {
 	struct write_buffer * next;
 	char *ptr;
-	int sz;
 	void *buffer;
+	int sz;
+	bool userobject;
 };
 
 struct wb_list {
@@ -68,6 +69,7 @@ struct socket_server {
 	int alloc_id;
 	int event_n;
 	int event_index;
+	struct socket_object_interface soi;
 	struct event ev[MAX_EVENT];
 	struct socket slot[MAX_SOCKET];
 	char buffer[MAX_INFO];
@@ -137,9 +139,40 @@ union sockaddr_all {
 	struct sockaddr_in6 v6;
 };
 
+struct send_object {
+	void * buffer;
+	int sz;
+	void (*free_func)(void *);
+};
+
 #define MALLOC skynet_malloc
 #define FREE skynet_free
 
+static inline bool
+send_object_init(struct socket_server *ss, struct send_object *so, void *object, int sz) {
+	if (sz < 0) {
+		so->buffer = ss->soi.buffer(object);
+		so->sz = ss->soi.size(object);
+		so->free_func = ss->soi.free;
+		return true;
+	} else {
+		so->buffer = object;
+		so->sz = sz;
+		so->free_func = FREE;
+		return false;
+	}
+}
+
+static inline void
+write_buffer_free(struct socket_server *ss, struct write_buffer *wb) {
+	if (wb->userobject) {
+		ss->soi.free(wb->buffer);
+	} else {
+		FREE(wb->buffer);
+	}
+	FREE(wb);
+}
+
 static void
 socket_keepalive(int fd) {
 	int keepalive = 1;
@@ -213,6 +246,7 @@ socket_server_create() {
 	ss->alloc_id = 0;
 	ss->event_n = 0;
 	ss->event_index = 0;
+	memset(&ss->soi, 0, sizeof(ss->soi));
 	FD_ZERO(&ss->rfds);
 	assert(ss->recvctrl_fd < FD_SETSIZE);
 
@@ -220,13 +254,12 @@ socket_server_create() {
 }
 
 static void
-free_wb_list(struct wb_list *list) {
+free_wb_list(struct socket_server *ss, struct wb_list *list) {
 	struct write_buffer *wb = list->head;
 	while (wb) {
 		struct write_buffer *tmp = wb;
 		wb = wb->next;
-		FREE(tmp->buffer);
-		FREE(tmp);
+		write_buffer_free(ss, tmp);
 	}
 	list->head = NULL;
 	list->tail = NULL;
@@ -242,8 +275,8 @@ force_close(struct socket_server *ss, struct socket *s, struct socket_message *r
 		return;
 	}
 	assert(s->type != SOCKET_TYPE_RESERVE);
-	free_wb_list(&s->high);
-	free_wb_list(&s->low);
+	free_wb_list(ss,&s->high);
+	free_wb_list(ss,&s->low);
 	if (s->type != SOCKET_TYPE_PACCEPT && s->type != SOCKET_TYPE_PLISTEN) {
 		sp_del(ss->event_fd, s->fd);
 	}
@@ -395,8 +428,7 @@ send_list(struct socket_server *ss, struct socket *s, struct wb_list *list, stru
 			break;
 		}
 		list->head = tmp->next;
-		FREE(tmp->buffer);
-		FREE(tmp);
+		write_buffer_free(ss,tmp);
 	}
 	list->tail = NULL;
 
@@ -469,10 +501,12 @@ send_buffer(struct socket_server *ss, struct socket *s, struct socket_message *r
 }
 
 static int
-append_sendbuffer_(struct wb_list *s, struct request_send * request, int n) {
+append_sendbuffer_(struct socket_server *ss, struct wb_list *s, struct request_send * request, int n) {
 	struct write_buffer * buf = MALLOC(sizeof(*buf));
-	buf->ptr = request->buffer+n;
-	buf->sz = request->sz - n;
+	struct send_object so;
+	buf->userobject = send_object_init(ss, &so, request->buffer, request->sz);
+	buf->ptr = so.buffer+n;
+	buf->sz = so.sz - n;
 	buf->buffer = request->buffer;
 	buf->next = NULL;
 	if (s->head == NULL) {
@@ -487,13 +521,13 @@ append_sendbuffer_(struct wb_list *s, struct request_send * request, int n) {
 }
 
 static inline void
-append_sendbuffer(struct socket *s, struct request_send * request, int n) {
-	s->wb_size += append_sendbuffer_(&s->high, request, n);
+append_sendbuffer(struct socket_server *ss, struct socket *s, struct request_send * request, int n) {
+	s->wb_size += append_sendbuffer_(ss, &s->high, request, n);
 }
 
 static inline void
-append_sendbuffer_low(struct socket *s, struct request_send * request) {
-	s->wb_size += append_sendbuffer_(&s->low, request, 0);
+append_sendbuffer_low(struct socket_server *ss,struct socket *s, struct request_send * request) {
+	s->wb_size += append_sendbuffer_(ss, &s->low, request, 0);
 }
 
 static inline int
@@ -512,15 +546,17 @@ static int
 send_socket(struct socket_server *ss, struct request_send * request, struct socket_message *result, int priority) {
 	int id = request->id;
 	struct socket * s = &ss->slot[HASH_ID(id)];
+	struct send_object so;
+	send_object_init(ss, &so, request->buffer, request->sz);
 	if (s->type == SOCKET_TYPE_INVALID || s->id != id 
 		|| s->type == SOCKET_TYPE_HALFCLOSE
 		|| s->type == SOCKET_TYPE_PACCEPT) {
-		FREE(request->buffer);
+		so.free_func(request->buffer);
 		return -1;
 	}
 	assert(s->type != SOCKET_TYPE_PLISTEN && s->type != SOCKET_TYPE_LISTEN);
 	if (send_buffer_empty(s) && s->type == SOCKET_TYPE_CONNECTED) {
-		int n = write(s->fd, request->buffer, request->sz);
+		int n = write(s->fd, so.buffer, so.sz);
 		if (n<0) {
 			switch(errno) {
 			case EINTR:
@@ -528,22 +564,22 @@ send_socket(struct socket_server *ss, struct request_send * request, struct sock
 				n = 0;
 				break;
 			default:
-				fprintf(stderr, "socket-server: write to %d (fd=%d) error.",id,s->fd);
+				fprintf(stderr, "socket-server: write to %d (fd=%d) error :%s.\n",id,s->fd,strerror(errno));
 				force_close(ss,s,result);
 				return SOCKET_CLOSE;
 			}
 		}
-		if (n == request->sz) {
-			FREE(request->buffer);
+		if (n == so.sz) {
+			so.free_func(request->buffer);
 			return -1;
 		}
-		append_sendbuffer(s, request, n);	// add to high priority list, even priority == PRIORITY_LOW
+		append_sendbuffer(ss, s, request, n);	// add to high priority list, even priority == PRIORITY_LOW
 		sp_write(ss->event_fd, s->fd, s, true);
 	} else {
 		if (priority == PRIORITY_LOW) {
-			append_sendbuffer_low(s, request);
+			append_sendbuffer_low(ss, s, request);
 		} else {
-			append_sendbuffer(s, request, 0);
+			append_sendbuffer(ss, s, request, 0);
 		}
 	}
 	return -1;
@@ -660,7 +696,7 @@ block_readpipe(int pipefd, void *buffer, int sz) {
 		if (n<0) {
 			if (errno == EINTR)
 				continue;
-			fprintf(stderr, "socket-server : read pipe error %s.",strerror(errno));
+			fprintf(stderr, "socket-server : read pipe error %s.\n",strerror(errno));
 			return;
 		}
 		// must atomic read from a pipe
@@ -904,15 +940,20 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int
 			fprintf(stderr, "socket-server: invalid socket\n");
 			break;
 		default:
-			if (e->write) {
-				int type = send_buffer(ss, s, result);
+			if (e->read) {
+				int type = forward_message(ss, s, result);
+				if (e->write) {
+					// Try to dispatch write message next step if write flag set.
+					e->read = false;
+					--ss->event_index;
+				}
 				if (type == -1)
 					break;
 				clear_closed_event(ss, result, type);
 				return type;
 			}
-			if (e->read) {
-				int type = forward_message(ss, s, result);
+			if (e->write) {
+				int type = send_buffer(ss, s, result);
 				if (type == -1)
 					break;
 				clear_closed_event(ss, result, type);
@@ -1087,3 +1128,9 @@ socket_server_nodelay(struct socket_server *ss, int id) {
 	request.u.setopt.value = 1;
 	send_request(ss, &request, 'T', sizeof(request.u.setopt));
 }
+
+void 
+socket_server_userobject(struct socket_server *ss, struct socket_object_interface *soi) {
+	ss->soi = *soi;
+}
+

skynet-src/socket_server.h

@@ -38,4 +38,13 @@ int socket_server_bind(struct socket_server *, uintptr_t opaque, int fd);
 
 void socket_server_nodelay(struct socket_server *, int id);
 
+struct socket_object_interface {
+	void * (*buffer)(void *);
+	int (*size)(void *);
+	void (*free)(void *);
+};
+
+// if you send package sz == -1, use soi.
+void socket_server_userobject(struct socket_server *, struct socket_object_interface *soi);
+
 #endif

test/pingserver.lua

@@ -1,5 +1,6 @@
 local skynet = require "skynet"
 local queue = require "skynet.queue"
+local snax = require "snax"
 
 local i = 0
 local hello = "hello"
@@ -32,6 +33,10 @@ function accept.hello()
 	end)
 end
 
+function accept.exit(...)
+	snax.exit(...)
+end
+
 function response.error()
 	error "throw an error"
 end
@@ -40,9 +45,6 @@ function init( ... )
 	print ("ping server start:", ...)
 	-- init queue
 	lock = queue()
-
--- You can return "queue" for queue service mode
---	return "queue"
 end
 
 function exit(...)

test/testmysql.lua

@@ -0,0 +1,127 @@
+local skynet = require "skynet"
+local mysql = require "mysql"
+
+local function dump(obj)
+    local getIndent, quoteStr, wrapKey, wrapVal, dumpObj
+    getIndent = function(level)
+        return string.rep("\t", level)
+    end
+    quoteStr = function(str)
+        return '"' .. string.gsub(str, '"', '\\"') .. '"'
+    end
+    wrapKey = function(val)
+        if type(val) == "number" then
+            return "[" .. val .. "]"
+        elseif type(val) == "string" then
+            return "[" .. quoteStr(val) .. "]"
+        else
+            return "[" .. tostring(val) .. "]"
+        end
+    end
+    wrapVal = function(val, level)
+        if type(val) == "table" then
+            return dumpObj(val, level)
+        elseif type(val) == "number" then
+            return val
+        elseif type(val) == "string" then
+            return quoteStr(val)
+        else
+            return tostring(val)
+        end
+    end
+    dumpObj = function(obj, level)
+        if type(obj) ~= "table" then
+            return wrapVal(obj)
+        end
+        level = level + 1
+        local tokens = {}
+        tokens[#tokens + 1] = "{"
+        for k, v in pairs(obj) do
+            tokens[#tokens + 1] = getIndent(level) .. wrapKey(k) .. " = " .. wrapVal(v, level) .. ","
+        end
+        tokens[#tokens + 1] = getIndent(level - 1) .. "}"
+        return table.concat(tokens, "\n")
+    end
+    return dumpObj(obj, 0)
+end
+
+local function test2( db)
+    local i=1
+    while true do
+        local    res = db:query("select * from cats order by id asc")
+        print ( "test2 loop times=" ,i,"\n","query result=",dump( res ) )
+        res = db:query("select * from cats order by id asc")
+        print ( "test2 loop times=" ,i,"\n","query result=",dump( res ) )
+
+        skynet.sleep(1000)
+        i=i+1
+    end
+end
+local function test3( db)
+    local i=1
+    while true do
+        local    res = db:query("select * from cats order by id asc")
+        print ( "test3 loop times=" ,i,"\n","query result=",dump( res ) )
+        res = db:query("select * from cats order by id asc")
+        print ( "test3 loop times=" ,i,"\n","query result=",dump( res ) )        
+        skynet.sleep(1000)
+        i=i+1
+    end
+end
+skynet.start(function()
+
+	local db=mysql.connect{	
+		host="127.0.0.1",
+		port=3306,
+		database="skynet",
+		user="root",
+		password="1",
+		max_packet_size = 1024 * 1024
+	}
+	if not db then
+		print("failed to connect")
+	end
+	print("testmysql success to connect to mysql server")
+
+	local res = db:query("drop table if exists cats")
+	res = db:query("create table cats " 
+		               .."(id serial primary key, ".. "name varchar(5))")
+	print( dump( res ) )
+
+	res = db:query("insert into cats (name) "
+                             .. "values (\'Bob\'),(\'\'),(null)")
+	print ( dump( res ) )
+
+	res = db:query("select * from cats order by id asc")
+	print ( dump( res ) )
+
+    -- test in another coroutine
+	skynet.fork( test2, db)
+    skynet.fork( test3, db)
+	-- multiresultset test
+	res = db:query("select * from cats order by id asc ; select * from cats")
+	print ("multiresultset test result=", dump( res ) )
+
+	print ("escape string test result=", mysql.quote_sql_str([[\mysql escape %string test'test"]]) )
+
+	-- bad sql statement
+	local res =  db:query("select * from notexisttable" )
+	print( "bad query test result=" ,dump(res) )
+
+    local i=1
+    while true do
+        local    res = db:query("select * from cats order by id asc")
+        print ( "test1 loop times=" ,i,"\n","query result=",dump( res ) )
+      
+        res = db:query("select * from cats order by id asc")
+        print ( "test1 loop times=" ,i,"\n","query result=",dump( res ) )
+
+
+        skynet.sleep(1000)
+        i=i+1
+    end
+
+	--db:disconnect()
+	--skynet.exit()
+end)
+

test/testping.lua

@@ -2,7 +2,7 @@ local skynet = require "skynet"
 local snax = require "snax"
 
 skynet.start(function()
-	local ps = snax.uniqueservice ("pingserver", "hello world")
+	local ps = snax.newservice ("pingserver", "hello world")
 	print(ps.req.ping("foobar"))
 	print(ps.post.hello())
 	print(pcall(ps.req.error))
@@ -31,6 +31,6 @@ end
 		print(string.format("%s\tcount:%d time:%f", name, v.count, v.time))
 	end
 
-	print(snax.kill(ps,"exit"))
+	print(ps.post.exit("exit")) -- == snax.kill(ps, "exit")
 	skynet.exit()
 end)