ready for 0.9.0

Ready for 0.9.0

HISTORY.md

@@ -1,3 +1,21 @@
+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

lualib-src/lua-netpack.c

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

@@ -179,7 +179,7 @@ wb_string(struct write_block *wb, const char *str, int len) {
 			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);

lualib-src/lua_mysqlaux.c

@@ -24,7 +24,7 @@ static unsigned int num_escape_sql_str(unsigned char *dst, unsigned char *src, s
                 case '\n':
                 case '\r':
                 case '\t':
-                case 26:  /* \z */
+                case 26:  /* \Z */
                 case '\\':
                 case '\'':
                 case '"':
@@ -73,7 +73,7 @@ escape_sql_str(unsigned char *dst, unsigned char *src, size_t size)
                     
                 case 26:
                     *dst++ = '\\';
-                    *dst++ = 'z';
+                    *dst++ = 'Z';
                     break;
                     
                 case '\\':

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()
+	MSG.data = dispatch_msg
-		for fd, msg, sz in netpack.pop, queue do
+
-			if connection[fd] then
+	local function dispatch_queue()
-				handler.message(fd, msg, sz)
+		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)

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(...)
+				initfunc(...)
-				if mode == "queue" then
-					skynet.fork(message_dispatch)
-				end
 				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
+			timing(method, ...)
-				queue(session, source, method , ...)
-			else
-				timing(method, ...)
-			end
 		end
 	end)
 end)

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);
+		write_buffer_free(ss, tmp);
-		FREE(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(ss,&s->high);
-	free_wb_list(&s->low);
+	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);
+		write_buffer_free(ss,tmp);
-		FREE(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;
+	struct send_object so;
-	buf->sz = request->sz - n;
+	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) {
+append_sendbuffer(struct socket_server *ss, struct socket *s, struct request_send * request, int n) {
-	s->wb_size += append_sendbuffer_(&s->high, request, n);
+	s->wb_size += append_sendbuffer_(ss, &s->high, request, n);
 }
 
 static inline void
-append_sendbuffer_low(struct socket *s, struct request_send * request) {
+append_sendbuffer_low(struct socket_server *ss,struct socket *s, struct request_send * request) {
-	s->wb_size += append_sendbuffer_(&s->low, request, 0);
+	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) {
+		if (n == so.sz) {
-			FREE(request->buffer);
+			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) {
+			if (e->read) {
-				int type = send_buffer(ss, s, result);
+				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) {
+			if (e->write) {
-				int type = forward_message(ss, s, result);
+				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/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)