Merge pull request #188 from cloudwu/dev

release v0.8.0

3rd/lua-mysqlaux/lua_mysqlaux.c

@@ -0,0 +1,372 @@
+//
+//  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>
+
+#define SHA1SIZE       20
+#define ROTL(bits,word) (((word) << (bits)) | ((word) >> (32-(bits))))
+typedef unsigned int uint32_t;
+
+struct sha
+{
+    uint32_t digest[5];
+    uint32_t w[80];
+    uint32_t a,b,c,d,e,f;
+    int err;
+};
+
+
+static uint32_t padded_length_in_bits(uint32_t len)
+{
+    if(len%64 == 56)
+    {
+        len++;
+    }
+    while((len%64)!=56)
+    {
+        len++;
+    }
+    return len*8;
+}
+
+
+static int calculate_sha1(struct sha *sha1, const unsigned char *text, uint32_t length)
+{
+    unsigned int i,j;
+    unsigned char *buffer=NULL, *pbuffer=NULL;
+    uint32_t bits=0;
+    uint32_t temp=0,k=0;
+    uint32_t lb = length*8;
+    
+    if (!sha1)
+    {
+        return 0;
+    }
+    // initialize the default digest values
+    sha1->digest[0] = 0x67452301;
+    sha1->digest[1] = 0xEFCDAB89;
+    sha1->digest[2] = 0x98BADCFE;
+    sha1->digest[3] = 0x10325476;
+    sha1->digest[4] = 0xC3D2E1F0;
+    sha1->a=sha1->b=sha1->c=sha1->d=sha1->e=sha1->f=0;
+    if (!text || !length)
+    {
+        return 0;
+    }
+    
+    bits = padded_length_in_bits(length);
+    buffer = (unsigned char *) malloc((bits/8)+8);
+    memset(buffer,0,(bits/8)+8);
+    if(buffer == NULL)
+    {
+    	return 1;
+    }
+    pbuffer = buffer;
+    memcpy(buffer, text, length);
+    
+    
+    //add 1 on the last of the message..
+    *(buffer+length) = 0x80;
+    for(i=length+1; i<(bits/8); i++)
+    {
+        *(buffer+i) = 0x00;
+    }
+    
+    *(buffer +(bits/8)+4+0) = (lb>>24) & 0xFF;
+    *(buffer +(bits/8)+4+1) = (lb>>16) & 0xFF;
+    *(buffer +(bits/8)+4+2) = (lb>>8) & 0xFF;
+    *(buffer +(bits/8)+4+3) = (lb>>0) & 0xFF;
+    
+    
+    //main loop
+    for(i=0; i<((bits+64)/512); i++)
+    {
+        //first empty the block for each pass..
+        for(j=0; j<80; j++)
+        {
+            sha1->w[j] = 0x00;
+        }
+        
+        
+        //fill the first 16 words with the characters read directly from the buffer.
+        for(j=0; j<16; j++)
+        {
+            sha1->w[j] =buffer[j*4+0];
+            sha1->w[j] = sha1->w[j]<<8;
+            sha1->w[j] |= buffer[j*4+1];
+            sha1->w[j] = sha1->w[j]<<8;
+            sha1->w[j] |= buffer[j*4+2];
+            sha1->w[j] = sha1->w[j]<<8;
+            sha1->w[j] |= buffer[j*4+3];
+        }
+        
+        //fill the rest 64 words using the formula
+        for(j=16; j<80; j++)
+        {
+            sha1->w[j] = (ROTL(1,(sha1->w[j-3] ^ sha1->w[j-8] ^ sha1->w[j-14] ^ sha1->w[j-16])));
+        }
+        
+        
+        //initialize hash for this chunck reading that has been stored in the structure digest
+        sha1->a = sha1->digest[0];
+        sha1->b = sha1->digest[1];
+        sha1->c = sha1->digest[2];
+        sha1->d = sha1->digest[3];
+        sha1->e = sha1->digest[4];
+        
+		//for all the 80 32bit blocks calculate f and use k accordingly per specification.
+        for(j=0; j<80; j++)
+        {
+            if((j>=0) && (j<20))
+            {
+                sha1->f = ((sha1->b)&(sha1->c)) | ((~(sha1->b))&(sha1->d));
+                k = 0x5A827999;
+                
+            }
+            else if((j>=20) && (j<40))
+            {
+                sha1->f = (sha1->b)^(sha1->c)^(sha1->d);
+                k = 0x6ED9EBA1;
+            }
+            else if((j>=40) && (j<60))
+            {
+                sha1->f = ((sha1->b)&(sha1->c)) | ((sha1->b)&(sha1->d)) | ((sha1->c)&(sha1->d));
+                k = 0x8F1BBCDC;
+            }
+            else if((j>=60) && (j<80))
+            {
+                sha1->f = (sha1->b)^(sha1->c)^(sha1->d);
+                k = 0xCA62C1D6;
+            }
+            
+            temp = ROTL(5,(sha1->a)) + (sha1->f) + (sha1->e) + k + sha1->w[j];
+            sha1->e = (sha1->d);
+            sha1->d = (sha1->c);
+            sha1->c = ROTL(30,(sha1->b));
+            sha1->b = (sha1->a);
+            sha1->a = temp;
+            
+            //reset temp to 0 to be in safe side only, not mandatory.
+            temp =0x00;
+            
+            
+        }
+        
+        // append to total hash.
+        sha1->digest[0] += sha1->a;
+        sha1->digest[1] += sha1->b;
+        sha1->digest[2] += sha1->c;
+        sha1->digest[3] += sha1->d;
+        sha1->digest[4] += sha1->e;
+        
+        
+		//since we used 512bit size block per each pass, let us update the buffer pointer accordingly.
+        buffer = buffer+64;
+        
+    }
+    free(pbuffer);
+    return 0;
+}
+
+static void int2ch4(int intVal,unsigned char *result)
+{
+    result[0]= (unsigned char)((intVal>>24) & 0x000000ff);
+    result[1]= (unsigned char)((intVal>>16) & 0x000000ff);
+    result[2]= (unsigned char)((intVal>> 8) & 0x000000ff);
+    result[3]= (unsigned char)((intVal>> 0) & 0x000000ff);
+}
+
+
+static int sha1_bin (lua_State *L) {
+    const void * msg = NULL;
+    size_t len =0;
+
+    if( lua_gettop(L) != 1 ){
+        return 0;
+    }	
+    if( lua_isnil(L,1) ) {
+        msg = NULL;
+        len =0;
+    }else{
+        msg=luaL_checklstring(L,1,&len);
+    }
+    struct sha tmpsha;
+    calculate_sha1( &tmpsha, msg, (uint32_t)len);
+    unsigned char tmpret[SHA1SIZE+8];
+    memset(tmpret,0,SHA1SIZE+8);
+    int i=0;
+    for ( i=0; i<5; i++)
+    {
+        int2ch4(tmpsha.digest[i], tmpret+i*4);
+    }
+    
+    lua_pushlstring(L, (char *)tmpret, SHA1SIZE);
+    return 1;
+}
+
+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[] = {
+    {"sha1_bin", sha1_bin},
+    {"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;
+}
+

HISTORY.md

@@ -1,3 +1,32 @@
+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
+* Bugfix: crash on exit
+
+v0.7.2 (2014-9-29)
+-----------
+* Bugfix : datacenter.wait
+* Bugfix : error in forker coroutine
+* Add skynet.term
+* Accept socket report port
+* sharedata can be update more than once
+
+v0.7.1 (2014-9-22)
+-----------
+* bugfix: wakeup sleep should return BREAK
+* bugfix: sharedatad load string
+* bugfix: dataserver forward error msg
+
 v0.7.0 (2014-9-8)
 -----------
 * Use sproto instead of cjson

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 : 3rd/lua-mysqlaux/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_log.lua

@@ -1,9 +1,16 @@
 local skynet = require "skynet"
+local harbor = require "skynet.harbor"
+
+local function monitor_master()
+	harbor.linkmaster()
+	print("master is down")
+	skynet.exit()
+end
 
 skynet.start(function()
 	print("Log server start")
 	skynet.monitor "simplemonitor"
 	local log = skynet.newservice("globallog")
-	skynet.exit()
+	skynet.fork(monitor_master)
 end)
 

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/watchdog.lua

@@ -10,6 +10,7 @@ local gate
 local agent = {}
 
 function SOCKET.open(fd, addr)
+	skynet.error("New client from : " .. addr)
 	agent[fd] = skynet.newservice("agent")
 	skynet.call(agent[fd], "lua", "start", gate, fd, proto)
 end

lualib-src/lua-netpack.c

@@ -240,12 +240,19 @@ filter_data_(lua_State *L, int fd, uint8_t * buffer, int size) {
 		buffer += need;
 		size -= need;
 		if (size == 0) {
-			lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
-			lua_pushinteger(L, fd);
-			lua_pushlightuserdata(L, uc->pack.buffer);
-			lua_pushinteger(L, uc->pack.size);
-			skynet_free(uc);
-			return 5;
+			if (q == NULL || q->head == q->tail ) {
+				lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
+				lua_pushinteger(L, fd);
+				lua_pushlightuserdata(L, uc->pack.buffer);
+				lua_pushinteger(L, uc->pack.size);
+				skynet_free(uc);
+				return 5;
+			}
+			else{
+				push_data(L, fd, uc->pack.buffer, uc->pack.size, 0);
+				skynet_free(uc);
+				return 1;
+			}
 		}
 		// more data
 		push_data(L, fd, uc->pack.buffer, uc->pack.size, 0);
@@ -274,13 +281,21 @@ filter_data_(lua_State *L, int fd, uint8_t * buffer, int size) {
 		}
 		if (size == pack_size) {
 			// just one package
-			lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
-			lua_pushinteger(L, fd);
-			void * result = skynet_malloc(pack_size);
-			memcpy(result, buffer, size);
-			lua_pushlightuserdata(L, result);
-			lua_pushinteger(L, size);
-			return 5;
+			if ( q == NULL || q->head == q->tail) {
+				lua_pushvalue(L, lua_upvalueindex(TYPE_DATA));
+				lua_pushinteger(L, fd);
+				void * result = skynet_malloc(pack_size);
+				memcpy(result, buffer, size);
+				lua_pushlightuserdata(L, result);
+				lua_pushinteger(L, size);
+				return 5;
+			}
+			else{
+				push_data(L, fd, buffer, pack_size, 1);
+				buffer += pack_size;
+				size -= pack_size;
+				return 1;
+			}
 		}
 		// more data
 		push_data(L, fd, buffer, pack_size, 1);

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,7 +173,7 @@ 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);
@@ -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-sharedata.c

@@ -747,9 +747,6 @@ lupdate(lua_State *L) {
 	luaL_checktype(L, 3, LUA_TTABLE);
 	struct ctrl * c= lua_touserdata(L, 1);
 	struct table *n = lua_touserdata(L, 2);
-	if (c->update) {
-		return luaL_error(L, "can't update more than once");
-	}
 	if (c->root == n) {
 		return luaL_error(L, "You should update a new object");
 	}

lualib-src/sproto/sproto.c

@@ -1076,9 +1076,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 +1098,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 +1108,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 +1119,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 +1135,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 +1154,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/internal.lua

@@ -4,14 +4,15 @@ local LIMIT = 8192
 
 local function chunksize(readbytes, body)
 	while true do
-		if #body > 128 then
-			return
-		end
-		body = body .. readbytes()
 		local f,e = body:find("\r\n",1,true)
 		if f then
 			return tonumber(body:sub(1,f-1),16), body:sub(e+1)
 		end
+		if #body > 128 then
+			-- pervent the attacker send very long stream without \r\n
+			return
+		end
+		body = body .. readbytes()
 	end
 end
 

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 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= mysqlaux.sha1_bin
+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/sharedata/corelib.lua

@@ -59,15 +59,15 @@ function meta:__index(key)
 		local newobj, newtbl = needupdate(self.__gcobj)
 		if newobj then
 			local newgcobj = newtbl.__gcobj
-			-- todo: update
 			local root = findroot(self)
 			update(root, newobj, newgcobj)
 			if obj == self.__obj then
 				error ("The key [" .. genkey(self) .. "] doesn't exist after update")
 			end
+			obj = self.__obj
 		end
 	end
-	local v = index(self.__obj, key)
+	local v = index(obj, key)
 	if type(v) == "userdata" then
 		local r = setmetatable({
 			__obj = v,
@@ -127,7 +127,7 @@ end
 
 function conf.update(self, pointer)
 	local cobj = self.__obj
-	assert(isdirty(cobj), "Obly dirty object can be update")
+	assert(isdirty(cobj), "Only dirty object can be update")
 	core.update(self.__gcobj, pointer, { __gcobj = core.box(pointer) })
 end
 

lualib/skynet.lua

@@ -95,6 +95,7 @@ end
 
 local coroutine_pool = {}
 local coroutine_yield = coroutine.yield
+local coroutine_count = 0
 
 local function co_create(f)
 	local co = table.remove(coroutine_pool)
@@ -109,6 +110,11 @@ local function co_create(f)
 				f(coroutine_yield())
 			end
 		end)
+		coroutine_count = coroutine_count + 1
+		if coroutine_count > 1024 then
+			skynet.error("May overload, create 1024 task")
+			coroutine_count = 0
+		end
 	else
 		coroutine.resume(co, f)
 	end
@@ -122,7 +128,7 @@ local function dispatch_wakeup()
 		local session = sleep_session[co]
 		if session then
 			session_id_coroutine[session] = "BREAK"
-			return suspend(co, coroutine.resume(co, true))
+			return suspend(co, coroutine.resume(co, true, "BREAK"))
 		end
 	end
 end
@@ -143,13 +149,15 @@ end
 function suspend(co, result, command, param, size)
 	if not result then
 		local session = session_coroutine_id[co]
-		local addr = session_coroutine_address[co]
-		if session ~= 0 then
-			-- only call response error
-			c.send(addr, skynet.PTYPE_ERROR, session, "")
+		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
+				c.send(addr, skynet.PTYPE_ERROR, session, "")
+			end
+			session_coroutine_id[co] = nil
+			session_coroutine_address[co] = nil
 		end
-		session_coroutine_id[co] = nil
-		session_coroutine_address[co] = nil
 		error(debug.traceback(co,tostring(command)))
 	end
 	if command == "CALL" then
@@ -166,7 +174,7 @@ function suspend(co, result, command, param, size)
 		session_response[co] = true
 		local ret
 		if not dead_service[co_address] then
-			ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size) >= 0
+			ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size) ~= nil
 		elseif size == nil then
 			c.trash(param, size)
 			ret = false
@@ -200,9 +208,9 @@ function suspend(co, result, command, param, size)
 			local ret
 			if not dead_service[co_address] then
 				if ok then
-					ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, f(...)) >=0
+					ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, f(...)) ~= nil
 				else
-					ret = c.send(co_address, skynet.PTYPE_ERROR, co_session, "") >=0
+					ret = c.send(co_address, skynet.PTYPE_ERROR, co_session, "") ~= nil
 				end
 			else
 				ret = false
@@ -246,7 +254,8 @@ function skynet.sleep(ti)
 	session = tonumber(session)
 	local succ, ret = coroutine_yield("SLEEP", session)
 	sleep_session[coroutine.running()] = nil
-	if ret == true then
+	assert(succ, ret)
+	if ret == "BREAK" then
 		return "BREAK"
 	end
 end
@@ -328,6 +337,7 @@ function skynet.time()
 end
 
 function skynet.exit()
+	fork_queue = {}	-- no fork coroutine can be execute after skynet.exit
 	skynet.send(".launcher","lua","REMOVE",skynet.self())
 	-- report the sources that call me
 	for co, session in pairs(session_coroutine_id) do
@@ -695,8 +705,19 @@ function skynet.task(ret)
 	return t
 end
 
+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,11 +49,15 @@ 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
 
+function dbgcmd.TERM(service)
+	skynet.term(service)
+end
+
 local function _debug_dispatch(session, address, cmd, ...)
 	local f = dbgcmd[cmd]
 	assert(f, cmd)

lualib/skynet/harbor.lua

@@ -19,4 +19,8 @@ function harbor.connect(id)
 	skynet.call(".cslave", "lua", "CONNECT", id)
 end
 
+function harbor.linkmaster()
+	skynet.call(".cslave", "lua", "LINKMASTER")
+end
+
 return harbor

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/gateserver.lua

@@ -37,6 +37,7 @@ function gateserver.start(handler)
 		local port = assert(conf.port)
 		maxclient = conf.maxclient or 1024
 		nodelay = conf.nodelay
+		skynet.error(string.format("Listen on %s:%d", address, port))
 		socket = socketdriver.listen(address, port)
 		socketdriver.start(socket)
 		if handler.open then
@@ -96,7 +97,7 @@ function gateserver.start(handler)
 
 	function MSG.error(fd, msg)
 		if handler.error then
-			handler.error(fd)
+			handler.error(fd, msg)
 		end
 		close_fd(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_snlua.c

@@ -3,7 +3,6 @@
 #include <lua.h>
 #include <lualib.h>
 #include <lauxlib.h>
-#include <assert.h>
 
 #include <assert.h>
 #include <string.h>

service/cslave.lua

@@ -9,6 +9,7 @@ local queryname = {}
 local harbor = {}
 local harbor_service
 local monitor = {}
+local monitor_master_set = {}
 
 local function read_package(fd)
 	local sz = socket.read(fd, 1)
@@ -58,7 +59,7 @@ local function ready()
 		connect_slave(k,v)
 	end
 	for name,address in pairs(globalname) do
-		skynet.redirect(harbor_service, address, "harbor", "N " .. name)
+		skynet.redirect(harbor_service, address, "harbor", 0, "N " .. name)
 	end
 end
 
@@ -99,6 +100,9 @@ local function monitor_master(master_fd)
 			end
 		else
 			skynet.error("Master disconnect")
+			for _, v in ipairs(monitor_master_set) do
+				v(true)
+			end
 			socket.close(master_fd)
 			break
 		end
@@ -183,6 +187,10 @@ function harbor.LINK(fd, id)
 	end
 end
 
+function harbor.LINKMASTER()
+	table.insert(monitor_master_set, skynet.response())
+end
+
 function harbor.CONNECT(fd, id)
 	if not slaves[id] then
 		if monitor[id] == nil then

service/datacenterd.lua

@@ -33,7 +33,7 @@ local function update(db, key, value, ...)
 	end
 end
 
-local function wakeup(db, key1, key2, ...)
+local function wakeup(db, key1, ...)
 	if key1 == nil then
 		return
 	end
@@ -43,7 +43,7 @@ local function wakeup(db, key1, key2, ...)
 	end
 	if q[mode] == "queue" then
 		db[key1] = nil
-		if key2 then
+		if select("#", ...) ~= 1 then
 			-- throw error because can't wake up a branch
 			for _,response in ipairs(q) do
 				response(false)
@@ -53,7 +53,7 @@ local function wakeup(db, key1, key2, ...)
 		end
 	else
 		-- it's branch
-		return wakeup(q , key2, ...)
+		return wakeup(q , ...)
 	end
 end
 

service/sharedatad.lua

@@ -4,6 +4,7 @@ local table = table
 
 local NORET = {}
 local pool = {}
+local pool_count = {}
 local objmap = {}
 
 local function newobj(name, tbl)
@@ -13,6 +14,7 @@ local function newobj(name, tbl)
 	local v = { value = tbl , obj = cobj, watch = {} }
 	objmap[cobj] = v
 	pool[name] = v
+	pool_count[name] = { n = 0, threshold = 16 }
 end
 
 local function collectobj()
@@ -32,15 +34,18 @@ end
 
 local CMD = {}
 
+local env_mt = { __index = _ENV }
+
 function CMD.new(name, t)
 	local dt = type(t)
 	local value
 	if dt == "table" then
 		value = t
 	elseif dt == "string" then
-		value = {}
-		local f = load(t, "=" .. name, "t", env)
+		value = setmetatable({}, env_mt)
+		local f = load(t, "=" .. name, "t", value)
 		f()
+		setmetatable(value, nil)
 	elseif dt == "nil" then
 		value = {}
 	else
@@ -52,10 +57,11 @@ end
 function CMD.delete(name)
 	local v = assert(pool[name])
 	pool[name] = nil
+	pool_count[name] = nil
 	assert(objmap[v.obj])
 	objmap[v.obj] = true
 	sharedata.host.decref(v.obj)
-	for _,response in ipairs(v.watch) do
+	for _,response in pairs(v.watch) do
 		response(true)
 	end
 end
@@ -83,23 +89,42 @@ function CMD.update(name, t)
 		objmap[oldcobj] = true
 		sharedata.host.decref(oldcobj)
 		pool[name] = nil
+		pool_count[name] = nil
 	end
 	CMD.new(name, t)
 	local newobj = pool[name].obj
 	if watch then
 		sharedata.host.markdirty(oldcobj)
-		for _,response in ipairs(watch) do
+		for _,response in pairs(watch) do
 			response(true, newobj)
 		end
 	end
 end
 
+local function check_watch(queue)
+	local n = 0
+	for k,response in pairs(queue) do
+		if not response "TEST" then
+			queue[k] = nil
+			n = n + 1
+		end
+	end
+	return n
+end
+
 function CMD.monitor(name, obj)
 	local v = assert(pool[name])
 	if obj ~= v.obj then
 		return v.obj
 	end
 
+	local n = pool_count[name].n
+	pool_count[name].n = n + 1
+	if n > pool_count[name].threshold then
+		n = n - check_watch(v.watch)
+		pool_count[name].threshold = n * 2
+	end
+
 	table.insert(v.watch, skynet.response())
 
 	return NORET

skynet-src/skynet_harbor.c

@@ -31,5 +31,17 @@ skynet_harbor_init(int harbor) {
 
 void
 skynet_harbor_start(void *ctx) {
+	// the HARBOR must be reserved to ensure the pointer is valid.
+	// It will be released at last by calling skynet_harbor_exit
+	skynet_context_reserve(ctx);
 	REMOTE = ctx;
 }
+
+void
+skynet_harbor_exit() {
+	struct skynet_context * ctx = REMOTE;
+	REMOTE= NULL;
+	if (ctx) {
+		skynet_context_release(ctx);
+	}
+}

skynet-src/skynet_harbor.h

@@ -26,5 +26,6 @@ void skynet_harbor_send(struct remote_message *rmsg, uint32_t source, int sessio
 int skynet_harbor_message_isremote(uint32_t handle);
 void skynet_harbor_init(int harbor);
 void skynet_harbor_start(void * ctx);
+void skynet_harbor_exit();
 
 #endif

skynet-src/skynet_mq.c

@@ -15,6 +15,7 @@
 // 1 means mq is in global mq , or the message is dispatching.
 
 #define MQ_IN_GLOBAL 1
+#define MQ_OVERLOAD 1024
 
 struct message_queue {
 	uint32_t handle;
@@ -24,6 +25,8 @@ struct message_queue {
 	int lock;
 	int release;
 	int in_global;
+	int overload;
+	int overload_threshold;
 	struct skynet_message *queue;
 	struct message_queue *next;
 };
@@ -116,6 +119,8 @@ skynet_mq_create(uint32_t handle) {
 	// If the service init success, skynet_context_new will call skynet_mq_force_push to push it to global queue.
 	q->in_global = MQ_IN_GLOBAL;
 	q->release = 0;
+	q->overload = 0;
+	q->overload_threshold = MQ_OVERLOAD;
 	q->queue = skynet_malloc(sizeof(struct skynet_message) * q->cap);
 	q->next = NULL;
 
@@ -150,17 +155,42 @@ skynet_mq_length(struct message_queue *q) {
 	return tail + cap - head;
 }
 
+int
+skynet_mq_overload(struct message_queue *q) {
+	if (q->overload) {
+		int overload = q->overload;
+		q->overload = 0;
+		return overload;
+	} 
+	return 0;
+}
+
 int
 skynet_mq_pop(struct message_queue *q, struct skynet_message *message) {
 	int ret = 1;
 	LOCK(q)
 
 	if (q->head != q->tail) {
-		*message = q->queue[q->head];
+		*message = q->queue[q->head++];
 		ret = 0;
-		if ( ++ q->head >= q->cap) {
-			q->head = 0;
+		int head = q->head;
+		int tail = q->tail;
+		int cap = q->cap;
+
+		if (head >= cap) {
+			q->head = head = 0;
 		}
+		int length = tail - head;
+		if (length < 0) {
+			length += cap;
+		}
+		while (length > q->overload_threshold) {
+			q->overload = length;
+			q->overload_threshold *= 2;
+		}
+	} else {
+		// reset overload_threshold when queue is empty
+		q->overload_threshold = MQ_OVERLOAD;
 	}
 
 	if (ret) {

skynet-src/skynet_mq.h

@@ -30,6 +30,7 @@ void skynet_mq_push(struct message_queue *q, struct skynet_message *message);
 
 // return the length of message queue, for debug
 int skynet_mq_length(struct message_queue *q);
+int skynet_mq_overload(struct message_queue *q);
 
 void skynet_mq_init();
 

skynet-src/skynet_server.c

@@ -178,6 +178,14 @@ skynet_context_grab(struct skynet_context *ctx) {
 	__sync_add_and_fetch(&ctx->ref,1);
 }
 
+void
+skynet_context_reserve(struct skynet_context *ctx) {
+	skynet_context_grab(ctx);
+	// don't count the context reserved, because skynet abort (the worker threads terminate) only when the total context is 0 .
+	// the reserved context will be release at last.
+	context_dec();
+}
+
 static void 
 delete_context(struct skynet_context *ctx) {
 	if (ctx->logfile) {
@@ -283,6 +291,10 @@ skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue
 			n = skynet_mq_length(q);
 			n >>= weight;
 		}
+		int overload = skynet_mq_overload(q);
+		if (overload) {
+			skynet_error(ctx, "May overload, message queue length = %d", overload);
+		}
 
 		skynet_monitor_trigger(sm, msg.source , handle);
 

skynet-src/skynet_server.h

@@ -10,6 +10,7 @@ struct skynet_monitor;
 
 struct skynet_context * skynet_context_new(const char * name, const char * parm);
 void skynet_context_grab(struct skynet_context *);
+void skynet_context_reserve(struct skynet_context *ctx);
 struct skynet_context * skynet_context_release(struct skynet_context *);
 uint32_t skynet_context_handle(struct skynet_context *);
 int skynet_context_push(uint32_t handle, struct skynet_message *message);

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

skynet-src/skynet_start.c

@@ -127,7 +127,6 @@ _worker(void *p) {
 	for (;;) {
 		q = skynet_context_message_dispatch(sm, q, weight);
 		if (q == NULL) {
-			CHECK_ABORT
 			if (pthread_mutex_lock(&m->mutex) == 0) {
 				++ m->sleep;
 				// "spurious wakeup" is harmless,
@@ -140,6 +139,7 @@ _worker(void *p) {
 				}
 			}
 		} 
+		CHECK_ABORT
 	}
 	return NULL;
 }
@@ -232,6 +232,9 @@ skynet_start(struct skynet_config * config) {
 	bootstrap(ctx, config->bootstrap);
 
 	_start(config->thread);
+
+	// harbor_exit may call socket send, so it should exit before socket_free
+	skynet_harbor_exit();
 	skynet_socket_free();
 	if (config->daemon) {
 		daemon_exit(config->daemon);

skynet-src/socket_server.c

@@ -335,7 +335,6 @@ open_socket(struct socket_server *ss, struct request_open * request, struct sock
 			sock = -1;
 			continue;
 		}
-		sp_nonblocking(sock);
 		break;
 	}
 
@@ -832,7 +831,10 @@ report_accept(struct socket_server *ss, struct socket *s, struct socket_message
 	result->data = NULL;
 
 	void * sin_addr = (u.s.sa_family == AF_INET) ? (void*)&u.v4.sin_addr : (void *)&u.v6.sin6_addr;
-	if (inet_ntop(u.s.sa_family, sin_addr, ss->buffer, sizeof(ss->buffer))) {
+	int sin_port = ntohs((u.s.sa_family == AF_INET) ? u.v4.sin_port : u.v6.sin6_port);
+	char tmp[INET6_ADDRSTRLEN];
+	if (inet_ntop(u.s.sa_family, sin_addr, tmp, sizeof(tmp))) {
+		snprintf(ss->buffer, sizeof(ss->buffer), "%s:%d", tmp, sin_port);
 		result->data = ss->buffer;
 	}
 

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/testoverload.lua

@@ -0,0 +1,44 @@
+local skynet = require "skynet"
+
+local mode = ...
+
+if mode == "slave" then
+
+local CMD = {}
+
+function CMD.sum(n)
+	skynet.error("for loop begin")
+	local s = 0
+	for i = 1, n do
+		s = s + i
+	end
+	skynet.error("for loop end")
+end
+
+function CMD.blackhole()
+end
+
+skynet.start(function()
+	skynet.dispatch("lua", function(_,_, cmd, ...)
+		local f = CMD[cmd]
+		f(...)
+	end)
+end)
+
+else
+
+skynet.start(function()
+	local slave = skynet.newservice(SERVICE_NAME, "slave")
+	for step = 1, 20 do
+		skynet.error("overload test ".. step)
+		for i = 1, 512 * step do
+			skynet.send(slave, "lua", "blackhole")
+		end
+		skynet.sleep(step)
+	end
+	local n = 1000000000
+	skynet.error(string.format("endless test n=%d", n))
+	skynet.send(slave, "lua", "sum", n)
+end)
+
+end

test/testterm.lua

@@ -0,0 +1,15 @@
+local skynet = require "skynet"
+
+local function term()
+	skynet.error("Sleep one second, and term the call to UNEXIST")
+	skynet.sleep(100)
+	local self = skynet.self()
+	skynet.send(skynet.self(), "debug", "TERM", "UNEXIST")
+end
+
+skynet.start(function()
+	skynet.fork(term)
+	skynet.error("call an unexist named service UNEXIST, may block")
+	pcall(skynet.call, "UNEXIST", "lua", "test")
+	skynet.error("unblock the unexisted service call")
+end)