lua serialize support

Makefile

@@ -12,7 +12,7 @@ snlua.so : service_lua.c
 gate.so : gate/mread.c gate/ringbuffer.c gate/main.c
 	gcc -Wall -g -fPIC --shared -o $@ $^ -I. -Igate 
 
-skynet.so : lua-skynet.c
+skynet.so : lua-skynet.c lua-serialize/serialize.c
 	gcc -Wall -g -fPIC --shared $^ -o $@
 
 client.so : service_client.c

agent.lua

@@ -1,7 +1,8 @@
 local skynet = require "skynet"
 local client = ...
 
-skynet.dispatch(function(msg,session)
+skynet.dispatch(function(msg, sz , session, address)
+	local message = skynet.tostring(msg,sz)
 	if session == 0 then
 		print("client command",msg)
 		local result = skynet.call("SIMPLEDB",msg)

globallog.lua

@@ -1,6 +1,7 @@
 local skynet = require "skynet"
 
-skynet.dispatch(function(message, session , from)
+skynet.dispatch(function(msg, sz , session , from)
+	local message = skynet.tostring(msg,sz)
 	print("[GLOBALLOG]", from,message)
 end)
 

launcher.lua

@@ -3,7 +3,8 @@ local string = string
 
 local instance = {}
 
-skynet.dispatch(function(message, session, address)
+skynet.dispatch(function(msg, sz , session, address)
+	local message = skynet.tostring(msg,sz)
 	if session == 0 then
 		-- init notice
 		local reply = instance[address]

lua-serialize/Makefile

@@ -0,0 +1,8 @@
+all : luaseri.so
+
+luaseri.so : serialize.c
+	gcc -Wall -fPIC -O2 -D LUAMODULE -o $@ --shared $^ 
+
+clean :
+	rm luaseri.so
+

lua-serialize/README

@@ -0,0 +1 @@
+see https://github.com/cloudwu/lua-serialize

lua-serialize/luaseri.h

@@ -0,0 +1,9 @@
+#ifndef LUA_SERIALIZE_H
+#define LUA_SERIALIZE_H
+
+#include <lua.h>
+
+int _luaseri_pack(lua_State *L);
+int _luaseri_unpack(lua_State *L);
+
+#endif

lua-serialize/serialize.c

@@ -0,0 +1,721 @@
+#include <lua.h>
+#include <lauxlib.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <assert.h>
+#include <string.h>
+
+#define TYPE_NIL 0
+#define TYPE_BOOLEAN 1
+// hibits 0 false 1 true
+#define TYPE_NUMBER 2
+// hibits 0 : 0 , 1: byte, 2:word, 4: dword, 8 : double
+#define TYPE_USERDATA 3
+#define TYPE_SHORT_STRING 4
+// hibits 0~31 : len
+#define TYPE_LONG_STRING 5
+#define TYPE_TABLE 6
+
+#define MAX_COOKIE 32
+#define COMBINE_TYPE(t,v) ((t) | (v) << 3)
+
+#define BLOCK_SIZE 128
+#define MAX_DEPTH 32
+
+struct block {
+	struct block * next;
+	char buffer[BLOCK_SIZE];
+};
+
+struct write_block {
+	struct block * head;
+	int len;
+	struct block * current;
+	int ptr;
+};
+
+struct read_block {
+	char * buffer;
+	struct block * current;
+	int len;
+	int ptr;
+};
+
+inline static struct block *
+blk_alloc(void) {
+	struct block *b = malloc(sizeof(struct block));
+	b->next = NULL;
+	return b;
+}
+
+inline static void
+wb_push(struct write_block *b, const void *buf, int sz) {
+	const char * buffer = buf;
+	if (b->ptr == BLOCK_SIZE) {
+_again:
+		b->current = b->current->next = blk_alloc();
+		b->ptr = 0;
+	}
+	if (b->ptr <= BLOCK_SIZE - sz) {
+		memcpy(b->current->buffer + b->ptr, buffer, sz);
+		b->ptr+=sz;
+		b->len+=sz;
+	} else {
+		int copy = BLOCK_SIZE - b->ptr;
+		memcpy(b->current->buffer + b->ptr, buffer, copy);
+		buffer += copy;
+		b->len += copy;
+		sz -= copy;
+		goto _again;
+	}
+}
+
+static void
+wb_init(struct write_block *wb , struct block *b) {
+	if (b==NULL) {
+		wb->head = blk_alloc();
+		wb->len = 0;
+		wb->current = wb->head;
+		wb->ptr = 0;
+		wb_push(wb, &wb->len, sizeof(wb->len));
+	} else {
+		wb->head = b;
+		int * plen = (int *)b->buffer;
+		int sz = *plen;
+		wb->len = sz;
+		while (b->next) {
+			sz -= BLOCK_SIZE;		
+			b = b->next;
+		}
+		wb->current = b;
+		wb->ptr = sz;
+	}
+}
+
+static struct block *
+wb_close(struct write_block *b) {
+	b->current = b->head;
+	b->ptr = 0;
+	wb_push(b, &b->len, sizeof(b->len));
+	b->current = NULL;
+	return b->head;
+}
+
+static void
+wb_free(struct write_block *wb) {
+	struct block *blk = wb->head;
+	while (blk) {
+		struct block * next = blk->next;
+		free(blk);
+		blk = next;
+	}
+	wb->head = NULL;
+	wb->current = NULL;
+	wb->ptr = 0;
+	wb->len = 0;
+}
+
+static void
+rball_init(struct read_block * rb, char * buffer, int size) {
+	rb->buffer = buffer;
+	rb->current = NULL;
+	rb->len = size;
+	rb->ptr = 0;
+}
+
+#ifdef LUAMODULE
+
+static int
+rb_init(struct read_block *rb, struct block *b) {
+	rb->buffer = NULL;
+	rb->current = b;
+	memcpy(&(rb->len),b->buffer,sizeof(rb->len));
+	rb->ptr = sizeof(rb->len);
+	rb->len -= rb->ptr;
+	return rb->len;
+}
+
+#endif
+
+static void *
+rb_read(struct read_block *rb, void *buffer, int sz) {
+	if (rb->len < sz) {
+		return NULL;
+	}
+
+	if (rb->buffer) {
+		int ptr = rb->ptr;
+		rb->ptr += sz;
+		rb->len -= sz;
+		return rb->buffer + ptr;
+	}
+
+	if (rb->ptr == BLOCK_SIZE) {
+		struct block * next = rb->current->next;
+		free(rb->current);
+		rb->current = next;
+		rb->ptr = 0;
+	}
+
+	int copy = BLOCK_SIZE - rb->ptr;
+
+	if (sz <= copy) {
+		void * ret = rb->current->buffer + rb->ptr;
+		rb->ptr += sz;
+		rb->len -= sz;
+		return ret;
+	}
+
+	char * tmp = buffer;
+
+	memcpy(tmp, rb->current->buffer + rb->ptr,  copy);
+	sz -= copy;
+	tmp += copy;
+	rb->len -= copy;
+
+	for (;;) {
+		struct block * next = rb->current->next;
+		free(rb->current);
+		rb->current = next;
+
+		if (sz < BLOCK_SIZE) {
+			memcpy(tmp, rb->current->buffer, sz);
+			rb->ptr = sz;
+			rb->len -= sz;
+			return buffer;
+		}
+		memcpy(tmp, rb->current->buffer, BLOCK_SIZE);
+		sz -= BLOCK_SIZE;
+		tmp += BLOCK_SIZE;
+		rb->len -= BLOCK_SIZE;
+	}
+}
+
+static void
+rb_close(struct read_block *rb) {
+	while (rb->current) {
+		struct block * next = rb->current->next;
+		free(rb->current);
+		rb->current = next;
+	}
+	rb->len = 0;
+	rb->ptr = 0;
+}
+
+static inline void
+wb_nil(struct write_block *wb) {
+	int 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);
+	wb_push(wb, &n, 1);
+}
+
+static inline void
+wb_integer(struct write_block *wb, int v) {
+	if (v == 0) {
+		int n = COMBINE_TYPE(TYPE_NUMBER , 0);
+		wb_push(wb, &n, 1);
+	} else if (v<0) {
+		int n = COMBINE_TYPE(TYPE_NUMBER , 4);
+		wb_push(wb, &n, 1);
+		wb_push(wb, &v, 4);
+	} else if (v<0x100) {
+		int n = COMBINE_TYPE(TYPE_NUMBER , 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_NUMBER , 2);
+		wb_push(wb, &n, 1);
+		uint16_t word = (uint16_t)v;
+		wb_push(wb, &word, 2);
+	} else {
+		int n = COMBINE_TYPE(TYPE_NUMBER , 4);
+		wb_push(wb, &n, 1);
+		wb_push(wb, &v, 4);
+	}
+}
+
+static inline void
+wb_number(struct write_block *wb, double v) {
+	int 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;
+	wb_push(wb, &n, 1);
+	wb_push(wb, &v, sizeof(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);
+		wb_push(wb, &n, 1);
+		if (len > 0) {
+			wb_push(wb, str, len);
+		}
+	} else {
+		assert(len < 0x10000);
+		int n = TYPE_LONG_STRING;
+		wb_push(wb, &n, 1);
+		uint16_t x = (uint16_t) len;
+		wb_push(wb, &x, 2);
+		wb_push(wb, str, len);
+	}
+}
+
+static void _pack_one(lua_State *L, struct write_block *b, int index, int depth);
+
+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);
+		wb_push(wb, &n, 1);
+		wb_integer(wb, array_size);
+	} else {
+		int n = COMBINE_TYPE(TYPE_TABLE, array_size);
+		wb_push(wb, &n, 1);
+	}
+
+	int i;
+	for (i=1;i<=array_size;i++) {
+		lua_rawgeti(L,index,i);
+		_pack_one(L, wb, -1, depth);
+		lua_pop(L,1);
+	}
+
+	return array_size;
+}
+
+static void
+wb_table_hash(lua_State *L, struct write_block * wb, int index, int depth, int array_size) {
+	lua_pushnil(L);
+	while (lua_next(L, index) != 0) {
+		if (lua_type(L,-2) == LUA_TNUMBER) {
+			lua_Number k = lua_tonumber(L,-2);
+			lua_Integer x = lua_tointeger(L,-2);
+			if (k == (lua_Number)x && x>0 && x<=array_size) {
+				lua_pop(L,1);
+				continue;
+			}
+		}
+		_pack_one(L,wb,-2,depth);
+		_pack_one(L,wb,-1,depth);
+		lua_pop(L, 1);
+	}
+	wb_nil(wb);
+}
+
+static void
+wb_table(lua_State *L, struct write_block *wb, int index, int depth) {
+	if (index < 0) {
+		index = lua_gettop(L) + index + 1;
+	}
+	int array_size = wb_table_array(L, wb, index, depth);
+	wb_table_hash(L, wb, index, depth, array_size);
+}
+
+static void
+_pack_one(lua_State *L, struct write_block *b, int index, int depth) {
+	if (depth > MAX_DEPTH) {
+		wb_free(b);
+		luaL_error(L, "serialize can't pack too depth table");
+	}
+	int type = lua_type(L,index);
+	switch(type) {
+	case LUA_TNIL:
+		wb_nil(b);
+		break;
+	case LUA_TNUMBER: {
+		lua_Integer x = lua_tointeger(L,index);
+		lua_Number n = lua_tonumber(L,index);
+		if ((lua_Number)x==n) {
+			wb_integer(b, x);
+		} else {
+			wb_number(b,n);
+		}
+		break;
+	}
+	case LUA_TBOOLEAN: 
+		wb_boolean(b, lua_toboolean(L,index));
+		break;
+	case LUA_TSTRING: {
+		size_t sz = 0;
+		const char *str = lua_tolstring(L,index,&sz);
+		if (sz >= 0x10000) {
+			wb_free(b);
+			luaL_error(L,"string is too long to serialize");
+		}
+		wb_string(b, str, sz);
+		break;
+	}
+	case LUA_TLIGHTUSERDATA:
+		wb_pointer(b, lua_touserdata(L,index));
+		break;
+	case LUA_TTABLE:
+		wb_table(L, b, index, depth+1);
+		break;
+	default:
+		wb_free(b);
+		luaL_error(L, "Unsupport type %s to serialize", lua_typename(L, type));
+	}
+}
+
+static void
+_pack_from(lua_State *L, struct write_block *b, int from) {
+	int n = lua_gettop(L) - from;
+	int i;
+	for (i=1;i<=n;i++) {
+		_pack_one(L, b , from + i, 0);
+	}
+}
+
+#ifdef LUAMODULE
+
+static int
+_pack(lua_State *L) {
+	struct write_block b;
+	wb_init(&b, NULL);
+	_pack_from(L,&b,0);
+	struct block * ret = wb_close(&b);
+	lua_pushlightuserdata(L,ret);
+	return 1;
+}
+
+static int
+_append(lua_State *L) {
+	struct write_block b;
+	wb_init(&b, lua_touserdata(L,1));
+	_pack_from(L,&b,1);
+	struct block * ret = wb_close(&b);
+	lua_pushlightuserdata(L,ret);
+	return 1;
+}
+
+#endif
+
+static inline void
+__invalid_stream(lua_State *L, struct read_block *rb, int line) {
+	int len = rb->len;
+	if (rb->buffer == NULL) {
+		rb_close(rb);
+	}
+	luaL_error(L, "Invalid serialize stream %d (line:%d)", len, line);
+}
+
+#define _invalid_stream(L,rb) __invalid_stream(L,rb,__LINE__)
+
+static double
+_get_number(lua_State *L, struct read_block *rb, int cookie) {
+	switch (cookie) {
+	case 0:
+		return 0;
+	case 1: {
+		uint8_t n = 0;
+		uint8_t * pn = rb_read(rb,&n,1);
+		if (pn == NULL)
+			_invalid_stream(L,rb);
+		return *pn;
+	}
+	case 2: {
+		uint16_t n = 0;
+		uint16_t * pn = rb_read(rb,&n,2);
+		if (pn == NULL)
+			_invalid_stream(L,rb);
+		return *pn;
+	}
+	case 4: {
+		int n = 0;
+		int * pn = rb_read(rb,&n,4);
+		if (pn == NULL)
+			_invalid_stream(L,rb);
+		return *pn;
+	}
+	case 8: {
+		double n = 0;
+		double * pn = rb_read(rb,&n,8);
+		if (pn == NULL)
+			_invalid_stream(L,rb);
+		return *pn;
+	}
+	default:
+		_invalid_stream(L,rb);
+		return 0;
+	}
+}
+
+static void *
+_get_pointer(lua_State *L, struct read_block *rb) {
+	void * userdata = 0;
+	void ** v = (void **)rb_read(rb,&userdata,sizeof(userdata));
+	if (v == NULL) {
+		_invalid_stream(L,rb);
+	}
+	return *v;
+}
+
+static void
+_get_buffer(lua_State *L, struct read_block *rb, int len) {
+	char tmp[len];
+	char * p = rb_read(rb,tmp,len);
+	lua_pushlstring(L,p,len);
+}
+
+static void _unpack_one(lua_State *L, struct read_block *rb);
+
+static void
+_unpack_table(lua_State *L, struct read_block *rb, int array_size) {
+	if (array_size == MAX_COOKIE-1) {
+		uint8_t type = 0;
+		uint8_t *t = rb_read(rb, &type, 1);
+		if (t==NULL || (*t & 7) != TYPE_NUMBER) {
+			_invalid_stream(L,rb);
+		}
+		array_size = (int)_get_number(L,rb,*t >> 3);
+	}
+	lua_createtable(L,array_size,0);
+	int i;
+	for (i=1;i<=array_size;i++) {
+		_unpack_one(L,rb);
+		lua_rawseti(L,-2,i);
+	}
+	for (;;) {
+		_unpack_one(L,rb);
+		if (lua_isnil(L,-1)) {
+			lua_pop(L,1);
+			return;
+		}
+		_unpack_one(L,rb);
+		lua_rawset(L,-3);
+	}
+}
+
+static void
+_push_value(lua_State *L, struct read_block *rb, int type, int cookie) {
+	switch(type) {
+	case TYPE_NIL:
+		lua_pushnil(L);
+		break;
+	case TYPE_BOOLEAN:
+		lua_pushboolean(L,cookie);
+		break;
+	case TYPE_NUMBER:
+		lua_pushnumber(L,_get_number(L,rb,cookie));
+		break;
+	case TYPE_USERDATA:
+		lua_pushlightuserdata(L,_get_pointer(L,rb));
+		break;
+	case TYPE_SHORT_STRING:
+		_get_buffer(L,rb,cookie);
+		break;
+	case TYPE_LONG_STRING: {
+		uint16_t len = 0;
+		uint16_t *plen = rb_read(rb, &len, 2);
+		if (plen == NULL) {
+			_invalid_stream(L,rb);
+		}
+		_get_buffer(L,rb,*plen);
+		break;
+	}
+	case TYPE_TABLE: {
+		_unpack_table(L,rb,cookie);
+		break;
+	}
+	}
+}
+
+static void
+_unpack_one(lua_State *L, struct read_block *rb) {
+	uint8_t type = 0;
+	uint8_t *t = rb_read(rb, &type, 1);
+	if (t==NULL) {
+		_invalid_stream(L, rb);
+	}
+	_push_value(L, rb, *t & 0x7, *t>>3);
+}
+
+#ifdef LUAMODULE
+
+static int
+_unpack(lua_State *L) {
+	struct block * blk = lua_touserdata(L,1);
+	if (blk == NULL) {
+		return luaL_error(L, "Need a block to unpack");
+	}
+	lua_settop(L,0);
+	struct read_block rb;
+	rb_init(&rb, blk);
+
+	int i;
+	for (i=0;;i++) {
+		if (i%16==15) {
+			lua_checkstack(L,i);
+		}
+		uint8_t type = 0;
+		uint8_t *t = rb_read(&rb, &type, 1);
+		if (t==NULL)
+			break;
+		_push_value(L, &rb, *t & 0x7, *t>>3);
+	}
+
+	rb_close(&rb);
+
+	return lua_gettop(L);
+}
+
+static int
+_dump_mem(const char * buffer, int len, int size) {
+	int i;
+	for (i=0;i<len && i<size;i++) {
+		printf("%02x ",(unsigned char)buffer[i]);
+	}
+	return size - len;
+}
+
+static int
+_dump(lua_State *L) {
+	struct block *b = lua_touserdata(L,1);
+	if (b==NULL) {
+		return luaL_error(L, "dump null pointer");
+	}
+	int len = 0;
+	memcpy(&len, b->buffer ,sizeof(len));
+	len -= sizeof(len);
+	printf("Len = %d\n",len);
+	len = _dump_mem(b->buffer + sizeof(len), BLOCK_SIZE - sizeof(len) , len);
+	while (len > 0) {
+		b=b->next;
+		len = _dump_mem(b->buffer, BLOCK_SIZE , len);
+	}
+	printf("\n");
+	return 0;
+}
+
+#endif
+
+static void
+_seri(lua_State *L, struct block *b) {
+	uint32_t len = 0;
+	memcpy(&len, b->buffer ,sizeof(len));
+
+	len -= 4;
+	uint8_t * buffer = malloc(len);
+	uint8_t * ptr = buffer;
+	int sz = len;
+	if (len < BLOCK_SIZE - 4) {
+		memcpy(ptr, b->buffer+4, len);
+	} else {
+		memcpy(ptr, b->buffer+4, BLOCK_SIZE-4);
+		ptr += BLOCK_SIZE-4;
+		len -= BLOCK_SIZE-4;
+		b = b->next;
+
+		while(len>0) {
+			if (len >= BLOCK_SIZE) {
+				memcpy(ptr, b->buffer, BLOCK_SIZE);
+				ptr += BLOCK_SIZE;
+				len -= BLOCK_SIZE;
+			} else {
+				memcpy(ptr, b->buffer, len);
+				break;
+			}
+			b = b->next;
+		}
+	}
+	
+	lua_pushlightuserdata(L, buffer);
+	lua_pushinteger(L, sz);
+}
+
+#ifdef LUAMODULE
+
+static int
+_serialize(lua_State *L) {
+	struct block *b = lua_touserdata(L,1);
+	if (b==NULL) {
+		return luaL_error(L, "dump null pointer");
+	}
+
+	_seri(L,b);
+
+	return 2;
+}
+
+#endif
+
+int
+_luaseri_unpack(lua_State *L) {
+	if (lua_isnoneornil(L,1)) {
+		return 0;
+	}
+	void * buffer = lua_touserdata(L,1);
+	if (buffer == NULL) {
+		return luaL_error(L, "deserialize null pointer");
+	}
+	int len = luaL_checkinteger(L,2);
+
+	lua_settop(L,0);
+	struct read_block rb;
+	rball_init(&rb, buffer, len);
+
+	int i;
+	for (i=0;;i++) {
+		if (i%16==15) {
+			lua_checkstack(L,i);
+		}
+		uint8_t type = 0;
+		uint8_t *t = rb_read(&rb, &type, 1);
+		if (t==NULL)
+			break;
+		_push_value(L, &rb, *t & 0x7, *t>>3);
+	}
+
+	// Need not free buffer
+
+	return lua_gettop(L);
+}
+
+int
+_luaseri_pack(lua_State *L) {
+	struct write_block wb;
+	wb_init(&wb, NULL);
+	_pack_from(L,&wb,0);
+	struct block * b = wb_close(&wb);
+	_seri(L,b);
+
+	while (b) {
+		struct block * next = b->next;
+		free(b);
+		b = next;
+	}
+
+	return 2;
+}
+
+#ifdef LUAMODULE
+
+int
+luaopen_luaseri(lua_State *L) {
+	luaL_Reg l[] = {
+		{ "pack", _pack },
+		{ "unpack", _unpack },
+		{ "append", _append },
+		{ "serialize", _serialize },
+		{ "deserialize", _luaseri_unpack },
+		{ "dump", _dump },
+		{ NULL, NULL },
+	};
+	luaL_newlib(L,l);
+	return 1;
+}
+
+#endif

lua-serialize/test.lua

@@ -0,0 +1,28 @@
+s = require "luaseri"
+
+a = s.pack { hello={3,4}, false, 1,2,3,4,5,6,7,8,9,0,1,2,3,4,5,6,7,8,9,0,1,2,3,4,5,6,7,8,9,0,1,2,3,4,5,6,7,8,9 }
+
+s.dump(a)
+
+a = s.append(a, 42,4.2,-1,1000,80000,"hello",true,false,nil,"1234567890123456789012345678901234567890")
+
+s.dump(a)
+print(a)
+
+function pr(t,...)
+	for k,v in pairs(t) do
+		print(k,v)
+	end
+	print(...)
+end
+
+print ("------")
+
+local seri, length = s.serialize(a)
+print(seri, length)
+
+pr(s.unpack(a))
+
+print("-------")
+
+pr(s.deserialize(seri, length))

lua-serialize/test2.lua

@@ -0,0 +1,15 @@
+local s = require "luaseri"
+
+addressbook = {
+	name = "Alice",
+	id = 12345,
+	phone = {
+		{ number = "1301234567" },
+		{ number = "87654321", type = "WORK" },
+	}
+}
+
+for i=1,100000 do
+	local u = s.pack (addressbook)
+	local t = s.unpack(u)
+end

lua-skynet.c

@@ -1,4 +1,5 @@
 #include "skynet.h"
+#include "lua-serialize/luaseri.h"
 
 #include <lua.h>
 #include <lauxlib.h>
@@ -35,8 +36,9 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr,
 	} else {
 		lua_pushinteger(L, session);
 		lua_pushstring(L, addr);
-		lua_pushlstring(L, msg, sz);
-		r = lua_pcall(L, 3, 0 , trace);
+		lua_pushlightuserdata(L,(void *)msg);
+		lua_pushinteger(L,sz);
+		r = lua_pcall(L, 4, 0 , trace);
 	}
 	if (r == LUA_OK) 
 		return;
@@ -138,6 +140,17 @@ _error(lua_State *L) {
 	return 0;
 }
 
+static int
+_tostring(lua_State *L) {
+	if (lua_isnoneornil(L,1)) {
+		return 0;
+	}
+	char * msg = lua_touserdata(L,1);
+	int sz = luaL_checkinteger(L,2);
+	lua_pushlstring(L,msg,sz);
+	return 1;
+}
+
 int
 luaopen_skynet_c(lua_State *L) {
 	luaL_checkversion(L);
@@ -146,6 +159,9 @@ luaopen_skynet_c(lua_State *L) {
 		{ "command" , _command },
 		{ "callback" , _callback },
 		{ "error", _error },
+		{ "tostring", _tostring },
+		{ "pack", _luaseri_pack },
+		{ "unpack", _luaseri_unpack },
 		{ NULL, NULL },
 	};
 	lua_pushcfunction(L, traceback);

redis-cli.lua

@@ -2,6 +2,8 @@ local skynet = require "skynet"
 local string = string
 local table = table
 local tonumber = tonumber
+local ipairs = ipairs
+local unpack = unpack
 local redis_server = ...
 local fd
 local write_fd
@@ -15,36 +17,67 @@ local function init_fd(fdstr)
 	read_fd = "READ " .. fd .. " "
 end
 
-skynet.dispatch(function(message, ...)
-	skynet.send(".connection", write_fd .. message)
+local function compose_message(msg)
+	local lines = { "*" .. #msg }
+	for _,v in ipairs(msg) do
+		table.insert(lines,"$"..#v)
+		table.insert(lines,v)
+	end
+	table.insert(lines,"")
+
+	local cmd =  table.concat(lines,"\r\n")
+	return cmd
+end
+
+local redcmd = {}
+
+redcmd[42] = function(data)	-- '*'
+	local n = tonumber(data)
+	if n < 1 then
+		skynet.ret(skynet.pack(true, nil))
+		return
+	end
+	local bulk = {}
+	for i = 1,n do
+		local line = skynet.call(".connection", readline_fd)
+		local bytes = tonumber(string.sub(line,2) + 2)
+		local data = skynet.call(".connection", read_fd .. bytes)
+		table.insert(result, string.sub(data,1,-3))
+	end
+	skynet.ret(skynet.pack(unpack(bulk)))
+end
+
+redcmd[36] = function(data) -- '$'
+	local bytes = tonumber(data)
+	if bytes < 0 then
+		skynet.ret(skynet.pack(true, nil))
+		return
+	end
+	local firstline = skynet.call(".connection", read_fd .. (bytes + 2))
+	skynet.ret(skynet.pack(string.sub(firstline,1,-3)))
+end
+
+redcmd[43] = function(data) -- '+'
+	skynet.ret(skynet.pack(true, data))
+end
+
+redcmd[45] = function(data) -- '-'
+	skynet.ret(skynet.pack(false, data))
+end
+
+redcmd[58] = function(data) -- ':'
+	skynet.ret(skynet.pack(true, tonumber(data)))
+end
+
+skynet.dispatch(function(msg, sz, session, address)
+	local message = { skynet.unpack(msg,sz) }
+	skynet.send(".connection", write_fd .. compose_message(message))
 	local result = skynet.call(".connection", readline_fd)
 	local firstchar = string.byte(result)
-	if firstchar == 42 then	-- '*'
-		local n = tonumber(string.sub(result,2))
-		if n < 1 then
-			skynet.ret(result .. "\r\n")
-			return
-		end
-		local bulk = { result }
-		for i = 1,n do
-			local line = skynet.call(".connection", readline_fd)
-			table.insert(result, line)
-			local bytes = tonumber(string.sub(line,2) + 2)
-			table.insert(result, bytes)
-		end
-		table.insert(result,"")
-		skynet.ret(table.concat(result,"\r\n"))
-	elseif firstchar == 36 then -- '$'
-		local bytes = tonumber(string.sub(result,2))
-		if bytes < 0 then
-			skynet.ret(result .. "\r\n")
-			return
-		end
-		local firstline = skynet.call(".connection", read_fd .. (bytes + 2))
-		skynet.ret(result .. "\r\n" .. firstline)
-	else
-		skynet.ret(result .. "\r\n")
-	end
+	local data = string.sub(result,2)
+	local f = redcmd[firstchar]
+	assert(f)
+	f(data)
 end)
 
 skynet.start(function()

simpledb.lua

@@ -13,7 +13,8 @@ function command.SET(key, value)
 	skynet.ret(last)
 end
 
-skynet.dispatch(function(message, session, from)
+skynet.dispatch(function(msg, sz , session, from)
+	local message = skynet.tostring(msg,sz)
 	print("simpledb",message, from, session)
 	local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)")
 	local f = command[cmd]

skynet.lua

@@ -9,14 +9,14 @@ local session_id_coroutine = {}
 local session_coroutine_id = {}
 local session_coroutine_address = {}
 
-local function suspend(co, result, command, param)
+local function suspend(co, result, command, param, size)
 	assert(result, command)
 	if command == "CALL" or command == "SLEEP" then
 		session_id_coroutine[param] = co
 	elseif command == "RETURN" then
 		local co_session = session_coroutine_id[co]
 		local co_address = session_coroutine_address[co]
-		c.send(co_address, co_session, param)
+		c.send(co_address, co_session, param, size)
 	else
 		assert(command == nil, command)
 		session_coroutine_id[co] = nil
@@ -70,29 +70,39 @@ function skynet.kill(name)
 end
 
 skynet.send = c.send
+skynet.pack = c.pack
+skynet.tostring = c.tostring
+skynet.unpack = c.unpack
 
-function skynet.call(addr, message)
-	local session = c.send(addr, -1, message)
-	return coroutine.yield("CALL", session)
+function skynet.call(addr, deseri , ...)
+	local t = type(deseri)
+	if t == "function" then
+		local session = c.send(addr, -1, ...)
+		return deseri(coroutine.yield("CALL", session))
+	else
+		assert(t=="string")
+		local session = c.send(addr, -1, deseri)
+		return c.tostring(coroutine.yield("CALL", session))
+	end
 end
 
-function skynet.ret(message)
-	coroutine.yield("RETURN", message)
+function skynet.ret(...)
+	coroutine.yield("RETURN", ...)
 end
 
 function skynet.dispatch(f)
-	c.callback(function(session, address , message)
+	c.callback(function(session, address , msg, sz)
 		if session <= 0 then
 			session = - session
 			co = coroutine.create(f)
 			session_coroutine_id[co] = session
 			session_coroutine_address[co] = address
-			suspend(co, coroutine.resume(co, message, session, address))
+			suspend(co, coroutine.resume(co, msg, sz, session, address))
 		else
 			local co = session_id_coroutine[session]
 			assert(co, session)
 			session_id_coroutine[session] = nil
-			suspend(co, coroutine.resume(co, message))
+			suspend(co, coroutine.resume(co, msg, sz))
 		end
 	end)
 end

testredis.lua

@@ -5,11 +5,10 @@ skynet.dispatch()
 local command ="*2\r\n$3\r\nGET\r\n$1\r\nA\r\n"
 
 skynet.start(function()
-	local cli = skynet.call(".launcher","broker redis snlua redis-cli.lua 127.0.0.1:6379")
+	local cli = skynet.call(".launcher","broker redis snlua redis-cli.lua 127.0.0.1:7379")
 	print("redis-cli:", cli)
 	assert(cli)
-	local result = skynet.call(cli, command)
-	print(result)
+	print(skynet.call(cli, skynet.unpack, skynet.pack("GET","A")))
 	skynet.exit()
 end)
 

watchdog.lua

@@ -31,7 +31,8 @@ function command:data(data)
 	end
 end
 
-skynet.dispatch(function(message)
+skynet.dispatch(function(msg, sz)
+	local message = skynet.tostring(msg,sz)
 	local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)")
 	id = tonumber(id)
 	local f = command[cmd]