update history

merge dev version

.gitignore

@@ -1,6 +1,7 @@
 *.o
 *.a
 skynet
+skynet.pid
 3rd/lua/lua
 3rd/lua/luac
 cservice

HISTORY.md

@@ -1,3 +1,13 @@
+v0.3.0 (2014-6-2)
+-----------
+* Add cluster support
+* Add single node mode
+* Add daemon mode
+* Bugfix: update lua-bson (signed 32bit int bug / check string length)
+* Optimize timer
+* Simplify message queue and optimize message dispatch
+* Use jemalloc release 3.6.0
+
 v0.2.1 (2014-5-19)
 -----------
 * Bugfix: check all the events already read after socket close

Makefile

@@ -40,13 +40,15 @@ jemalloc : $(MALLOC_STATICLIB)
 
 # skynet
 
-CSERVICE = snlua logger gate master harbor
-LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile multicast
+CSERVICE = snlua logger gate master harbor dummy
+LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack \
+  cjson clientsocket memory profile multicast \
+  cluster
 
 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 \
   skynet_harbor.c skynet_env.c skynet_monitor.c skynet_socket.c socket_server.c \
-  malloc_hook.c
+  malloc_hook.c skynet_daemon.c
 
 all : \
   $(SKYNET_BUILD_PATH)/skynet \
@@ -105,6 +107,9 @@ $(LUA_CLIB_PATH)/profile.so : lualib-src/lua-profile.c | $(LUA_CLIB_PATH)
 $(LUA_CLIB_PATH)/multicast.so : lualib-src/lua-multicast.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ 
 
+$(LUA_CLIB_PATH)/cluster.so : lualib-src/lua-cluster.c | $(LUA_CLIB_PATH)
+	$(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ 
+
 clean :
 	rm -f $(SKYNET_BUILD_PATH)/skynet $(CSERVICE_PATH)/*.so $(LUA_CLIB_PATH)/*.so
 

examples/agent.lua

@@ -8,7 +8,7 @@ local CMD = {}
 local client_fd
 
 local function send_client(v)
-	socket.write(client_fd, netpack.pack(jsonpack.pack(0,v)))
+	socket.write(client_fd, netpack.pack(jsonpack.pack(0, {true, v})))
 end
 
 local function response_client(session,v)

examples/cluster1.lua

@@ -0,0 +1,9 @@
+local skynet = require "skynet"
+local cluster = require "cluster"
+
+skynet.start(function()
+	skynet.newservice("simpledb")
+	print(skynet.call("SIMPLEDB", "lua", "SET", "a", "foobar"))
+	print(skynet.call("SIMPLEDB", "lua", "GET", "a"))
+	cluster.open(2528)
+end)

examples/cluster2.lua

@@ -0,0 +1,6 @@
+local skynet = require "skynet"
+local cluster = require "cluster"
+
+skynet.start(function()
+	print(cluster.call("db", "SIMPLEDB", "GET", "a"))
+end)

examples/clustername.lua

@@ -0,0 +1 @@
+db = "127.0.0.1:2528"

examples/config

@@ -12,3 +12,4 @@ lualoader = "lualib/loader.lua"
 -- preload = "./examples/preload.lua"	-- run preload.lua before every lua service run
 snax = root.."examples/?.lua;"..root.."test/?.lua"
 cpath = root.."cservice/?.so"
+-- daemon = "./skynet.pid"

examples/config.c1

@@ -0,0 +1,9 @@
+thread = 8
+logger = nil
+harbor = 0
+start = "cluster1"
+bootstrap = "snlua bootstrap"	-- The service for bootstrap
+luaservice = "./service/?.lua;./test/?.lua;./examples/?.lua"
+lualoader = "lualib/loader.lua"
+cpath = "./cservice/?.so"
+cluster = "./examples/clustername.lua"

examples/config.c2

@@ -0,0 +1,9 @@
+thread = 8
+logger = nil
+harbor = 0
+start = "cluster2"
+bootstrap = "snlua bootstrap"	-- The service for bootstrap
+luaservice = "./service/?.lua;./test/?.lua;./examples/?.lua"
+lualoader = "lualib/loader.lua"
+cpath = "./cservice/?.so"
+cluster = "./examples/clustername.lua"

lualib-src/lua-bson.c

@@ -1,5 +1,3 @@
-#include "skynet_malloc.h"
-
 #include <lua.h>
 #include <lauxlib.h>
 
@@ -73,7 +71,7 @@ struct bson_reader {
 static inline void
 bson_destroy(struct bson *b) {
 	if (b->ptr != b->buffer) {
-		skynet_free(b->ptr);
+		free(b->ptr);
 	}
 }
 
@@ -93,10 +91,10 @@ bson_reserve(struct bson *b, int sz) {
 	} while (b->cap <= b->size + sz);
 
 	if (b->ptr == b->buffer) {
-		b->ptr = skynet_malloc(b->cap);
+		b->ptr = malloc(b->cap);
 		memcpy(b->ptr, b->buffer, b->size);
 	} else {
-		b->ptr = skynet_realloc(b->ptr, b->cap);
+		b->ptr = realloc(b->ptr, b->cap);
 	}
 }
 
@@ -258,13 +256,13 @@ append_key(struct bson *bs, int type, const char *key, size_t sz) {
 
 static void
 append_number(struct bson *bs, lua_State *L, const char *key, size_t sz) {
-	lua_Integer i = lua_tointeger(L, -1);
+	int64_t i = lua_tointeger(L, -1);
 	lua_Number d = lua_tonumber(L,-1);
 	if (i != d) {
 		append_key(bs, BSON_REAL, key, sz);
 		write_double(bs, d);
 	} else {
-		int si = (int64_t)i >> 32;
+		int si = i >> 31;
 		if (si == 0 || si == -1) {
 			append_key(bs, BSON_INT32, key, sz);
 			write_int32(bs, i);
@@ -459,7 +457,7 @@ pack_dict(lua_State *L, struct bson *b, bool isarray) {
 }
 
 static void
-pack_sorted_dict(lua_State *L, struct bson *b, int n) {
+pack_ordered_dict(lua_State *L, struct bson *b, int n) {
 	int length = reserve_length(b);
 	int i;
 	for (i=0;i<n;i+=2) {
@@ -534,6 +532,9 @@ unpack_dict(lua_State *L, struct bson_reader *br, bool array) {
 			break;
 		case BSON_STRING: {
 			int sz = read_int32(L, &t);
+			if (sz == 0) {
+				luaL_error(L, "Invalid bson string , length = 0");
+			}
 			lua_pushlstring(L, read_bytes(L, &t, sz), sz-1);
 			break;
 		}
@@ -806,6 +807,9 @@ lreplace(lua_State *L) {
 		write_int64(&b, i);
 		break;
 	}
+	default:
+		luaL_error(L, "Can't replace type %d", type);
+		break;
 	}
 	return 0;
 }
@@ -865,7 +869,7 @@ lencode_order(lua_State *L) {
 	if (n%2 != 0) {
 		return luaL_error(L, "Invalid ordered dict");
 	}
-	pack_sorted_dict(L, &b, n);
+	pack_ordered_dict(L, &b, n);
 	lua_settop(L,1);
 	void * ud = lua_newuserdata(L, b.size);
 	memcpy(ud, b.ptr, b.size);
@@ -1190,3 +1194,4 @@ luaopen_bson(lua_State *L) {
 
 	return 1;
 }
+

lualib-src/lua-cluster.c

@@ -0,0 +1,205 @@
+#include <lua.h>
+#include <lauxlib.h>
+#include <string.h>
+
+#include "skynet.h"
+
+/*
+	uint32_t/string addr 
+	uint32_t/session session
+	lightuserdata msg
+	uint32_t sz
+
+	return 
+		string request
+		uint32_t next_session
+ */
+
+#define TEMP_LENGTH 0x10002
+
+static void
+fill_uint32(uint8_t * buf, uint32_t n) {
+	buf[0] = n & 0xff;
+	buf[1] = (n >> 8) & 0xff;
+	buf[2] = (n >> 16) & 0xff;
+	buf[3] = (n >> 24) & 0xff;
+}
+
+static void
+fill_header(lua_State *L, uint8_t *buf, int sz, void *msg) {
+	if (sz >= 0x10000) {
+		skynet_free(msg);
+		luaL_error(L, "request message is too long %d", sz);
+	}
+	buf[0] = (sz >> 8) & 0xff;
+	buf[1] = sz & 0xff;
+}
+
+static void
+packreq_number(lua_State *L, int session, void * msg, size_t sz) {
+	uint32_t addr = lua_tounsigned(L,1);
+	uint8_t buf[TEMP_LENGTH];
+	fill_header(L, buf, sz+9, msg);
+	buf[2] = 0;
+	fill_uint32(buf+3, addr);
+	fill_uint32(buf+7, (uint32_t)session);
+	memcpy(buf+11,msg,sz);
+
+	lua_pushlstring(L, (const char *)buf, sz+11);
+}
+
+static void
+packreq_string(lua_State *L, int session, void * msg, size_t sz) {
+	size_t namelen = 0;
+	const char *name = lua_tolstring(L, 1, &namelen);
+	if (name == NULL || namelen < 1 || namelen > 255) {
+		skynet_free(msg);
+		luaL_error(L, "name is too long %s", name);
+	}
+
+	uint8_t buf[TEMP_LENGTH];
+	fill_header(L, buf, sz+5+namelen, msg);
+	buf[2] = (uint8_t)namelen;
+	memcpy(buf+3, name, namelen);
+	fill_uint32(buf+3+namelen, (uint32_t)session);
+	memcpy(buf+7+namelen,msg,sz);
+
+	lua_pushlstring(L, (const char *)buf, sz+7+namelen);
+}
+
+static int
+lpackrequest(lua_State *L) {
+	void *msg = lua_touserdata(L,3);
+	if (msg == NULL) {
+		return luaL_error(L, "Invalid request message");
+	}
+	size_t sz = luaL_checkunsigned(L,4);
+	int session = luaL_checkinteger(L,2);
+	if (session <= 0) {
+		return luaL_error(L, "Invalid request session %d", session);
+	}
+	int addr_type = lua_type(L,1);
+	if (addr_type == LUA_TNUMBER) {
+		packreq_number(L, session, msg, sz);
+	} else {
+		packreq_string(L, session, msg, sz);
+	}
+	if (++session < 0) {
+		session = 1;
+	}
+	skynet_free(msg);
+	lua_pushinteger(L, session);
+	return 2;
+}
+
+/*
+	string packed message
+	return 	
+		uint32_t or string addr
+		int session
+		string msg
+ */
+
+static inline uint32_t
+unpack_uint32(const uint8_t * buf) {
+	return buf[0] | buf[1]<<8 | buf[2]<<16 | buf[3]<<24;
+}
+
+static int
+unpackreq_number(lua_State *L, const uint8_t * buf, size_t sz) {
+	if (sz < 9) {
+		return luaL_error(L, "Invalid cluster message");
+	}
+	uint32_t address = unpack_uint32(buf+1);
+	uint32_t session = unpack_uint32(buf+5);
+	lua_pushunsigned(L, address);
+	lua_pushunsigned(L, session);
+	lua_pushlstring(L, (const char *)buf+9, sz-9);
+
+	return 3;
+}
+
+static int
+unpackreq_string(lua_State *L, const uint8_t * buf, size_t sz) {
+	size_t namesz = buf[0];
+	if (sz < namesz + 5) {
+		return luaL_error(L, "Invalid cluster message");
+	}
+	lua_pushlstring(L, (const char *)buf+1, namesz);
+	uint32_t session = unpack_uint32(buf + namesz + 1);
+	lua_pushunsigned(L, session);
+	lua_pushlstring(L, (const char *)buf+1+namesz+4, sz - namesz - 5);
+
+	return 3;
+}
+
+static int
+lunpackrequest(lua_State *L) {
+	size_t sz;
+	const char *msg = luaL_checklstring(L,1,&sz);
+	if (msg[0] == 0) {
+		return unpackreq_number(L, (const uint8_t *)msg, sz);
+	} else {
+		return unpackreq_string(L, (const uint8_t *)msg, sz);
+	}
+}
+
+/*
+	int session
+	lightuserdata msg
+	int sz
+	return string response
+ */
+static int
+lpackresponse(lua_State *L) {
+	uint32_t session = luaL_checkunsigned(L,1);
+	void * msg = lua_touserdata(L,2);
+	size_t sz = luaL_checkunsigned(L, 3);
+
+	uint8_t buf[TEMP_LENGTH];
+	fill_header(L, buf, sz+4, msg);
+	fill_uint32(buf+2, session);
+	memcpy(buf+6,msg,sz);
+
+	skynet_free(msg);
+
+	lua_pushlstring(L, (const char *)buf, sz+6);
+
+	return 1;
+}
+
+/*
+	string packed response
+	return integer session
+		boolean ok
+		string msg
+ */
+static int
+lunpackresponse(lua_State *L) {
+	size_t sz;
+	const char * buf = luaL_checklstring(L, 1, &sz);
+	if (sz < 4) {
+		return 0;
+	}
+	uint32_t session = unpack_uint32((const uint8_t *)buf);
+	lua_pushunsigned(L, session);
+	lua_pushboolean(L, 1);
+	lua_pushlstring(L, buf+4, sz-4);
+
+	return 3;
+}
+
+int
+luaopen_cluster_c(lua_State *L) {
+	luaL_Reg l[] = {
+		{ "packrequest", lpackrequest },
+		{ "unpackrequest", lunpackrequest },
+		{ "packresponse", lpackresponse },
+		{ "unpackresponse", lunpackresponse },
+		{ NULL, NULL },
+	};
+	luaL_checkversion(L);
+	luaL_newlib(L,l);
+
+	return 1;
+}

lualib-src/lua-socket.c

@@ -191,7 +191,7 @@ pop_lstring(lua_State *L, struct socket_buffer *sb, int sz, int skip) {
 /*
 	userdata send_buffer
 	table pool
-	integer sz
+	integer sz or string (big endian)
  */
 static int
 lpopbuffer(lua_State *L) {
@@ -200,7 +200,23 @@ lpopbuffer(lua_State *L) {
 		return luaL_error(L, "Need buffer object at param 1");
 	}
 	luaL_checktype(L,2,LUA_TTABLE);
-	int sz = luaL_checkinteger(L,3);
+	int type = lua_type(L,3);
+	int sz;
+	if (type == LUA_TNUMBER) {
+		sz = lua_tointeger(L,3);
+	} else {
+		size_t len;
+		const uint8_t * s = (const uint8_t *)luaL_checklstring(L, 3, &len);
+		if (len > 4 || len < 1) {
+			return luaL_error(L, "Invalid read %s", s);
+		}
+		int i;
+		sz = 0;
+		for (i=0;i<(int)len;i++) {
+			sz <<= 8;
+			sz |= s[i];
+		}
+	}
 	if (sb->size < sz || sz == 0) {
 		lua_pushnil(L);
 	} else {

lualib/cluster.lua

@@ -0,0 +1,19 @@
+local skynet = require "skynet"
+
+local clusterd
+local cluster = {}
+
+function cluster.call(node, address, ...)
+	-- skynet.pack(...) will free by cluster.c.packrequest
+	return skynet.call(clusterd, "lua", "req", node, address, skynet.pack(...))
+end
+
+function cluster.open(port)
+	skynet.call(clusterd, "lua", "listen", "0.0.0.0", port)
+end
+
+skynet.init(function()
+	clusterd = skynet.uniqueservice("clusterd")
+end)
+
+return cluster

service-src/service_dummy.c

@@ -0,0 +1,292 @@
+#include "skynet.h"
+#include "skynet_harbor.h"
+
+#include <string.h>
+#include <assert.h>
+
+#define HASH_SIZE 4096
+#define DEFAULT_QUEUE_SIZE 1024
+
+struct msg {
+	uint8_t * buffer;
+	size_t size;
+};
+
+struct msg_queue {
+	int size;
+	int head;
+	int tail;
+	struct msg * data;
+};
+
+struct keyvalue {
+	struct keyvalue * next;
+	char key[GLOBALNAME_LENGTH];
+	uint32_t hash;
+	uint32_t value;
+	struct msg_queue * queue;
+};
+
+struct hashmap {
+	struct keyvalue *node[HASH_SIZE];
+};
+
+/*
+	message type (8bits) is in destination high 8bits
+	harbor id (8bits) is also in that place , but  remote message doesn't need harbor id.
+ */
+struct remote_message_header {
+	uint32_t source;
+	uint32_t destination;
+	uint32_t session;
+};
+
+// 12 is sizeof(struct remote_message_header)
+#define HEADER_COOKIE_LENGTH 12
+
+struct dummy {
+	struct skynet_context *ctx;
+	struct hashmap * map;
+};
+
+// hash table
+
+static void
+_push_queue(struct msg_queue * queue, const void * buffer, size_t sz, struct remote_message_header * header) {
+	// If there is only 1 free slot which is reserved to distinguish full/empty
+	// of circular buffer, expand it.
+	if (((queue->tail + 1) % queue->size) == queue->head) {
+		struct msg * new_buffer = skynet_malloc(queue->size * 2 * sizeof(struct msg));
+		int i;
+		for (i=0;i<queue->size-1;i++) {
+			new_buffer[i] = queue->data[(i+queue->head) % queue->size];
+		}
+		skynet_free(queue->data);
+		queue->data = new_buffer;
+		queue->head = 0;
+		queue->tail = queue->size - 1;
+		queue->size *= 2;
+	}
+	struct msg * slot = &queue->data[queue->tail];
+	queue->tail = (queue->tail + 1) % queue->size;
+
+	slot->buffer = skynet_malloc(sz + sizeof(*header));
+	memcpy(slot->buffer, buffer, sz);
+	memcpy(slot->buffer + sz, header, sizeof(*header));
+	slot->size = sz + sizeof(*header);
+}
+
+static struct msg *
+_pop_queue(struct msg_queue * queue) {
+	if (queue->head == queue->tail) {
+		return NULL;
+	}
+	struct msg * slot = &queue->data[queue->head];
+	queue->head = (queue->head + 1) % queue->size;
+	return slot;
+}
+
+static struct msg_queue *
+_new_queue() {
+	struct msg_queue * queue = skynet_malloc(sizeof(*queue));
+	queue->size = DEFAULT_QUEUE_SIZE;
+	queue->head = 0;
+	queue->tail = 0;
+	queue->data = skynet_malloc(DEFAULT_QUEUE_SIZE * sizeof(struct msg));
+
+	return queue;
+}
+
+static void
+_release_queue(struct msg_queue *queue) {
+	if (queue == NULL)
+		return;
+	struct msg * m = _pop_queue(queue);
+	while (m) {
+		skynet_free(m->buffer);
+		m = _pop_queue(queue);
+	}
+	skynet_free(queue->data);
+	skynet_free(queue);
+}
+
+static struct keyvalue *
+_hash_search(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) {
+	uint32_t *ptr = (uint32_t*) name;
+	uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3];
+	struct keyvalue * node = hash->node[h % HASH_SIZE];
+	while (node) {
+		if (node->hash == h && strncmp(node->key, name, GLOBALNAME_LENGTH) == 0) {
+			return node;
+		}
+		node = node->next;
+	}
+	return NULL;
+}
+
+static struct keyvalue *
+_hash_insert(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) {
+	uint32_t *ptr = (uint32_t *)name;
+	uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3];
+	struct keyvalue ** pkv = &hash->node[h % HASH_SIZE];
+	struct keyvalue * node = skynet_malloc(sizeof(*node));
+	memcpy(node->key, name, GLOBALNAME_LENGTH);
+	node->next = *pkv;
+	node->queue = NULL;
+	node->hash = h;
+	node->value = 0;
+	*pkv = node;
+
+	return node;
+}
+
+static struct hashmap * 
+_hash_new() {
+	struct hashmap * h = skynet_malloc(sizeof(struct hashmap));
+	memset(h,0,sizeof(*h));
+	return h;
+}
+
+static void
+_hash_delete(struct hashmap *hash) {
+	int i;
+	for (i=0;i<HASH_SIZE;i++) {
+		struct keyvalue * node = hash->node[i];
+		while (node) {
+			struct keyvalue * next = node->next;
+			_release_queue(node->queue);
+			skynet_free(node);
+			node = next;
+		}
+	}
+	skynet_free(hash);
+}
+
+///////////////
+
+struct dummy *
+dummy_create(void) {
+	struct dummy * d = skynet_malloc(sizeof(*d));
+	d->map = _hash_new();
+	return d;
+}
+
+void
+dummy_release(struct dummy *d) {
+	_hash_delete(d->map);
+	skynet_free(d);
+}
+
+static inline void
+to_bigendian(uint8_t *buffer, uint32_t n) {
+	buffer[0] = (n >> 24) & 0xff;
+	buffer[1] = (n >> 16) & 0xff;
+	buffer[2] = (n >> 8) & 0xff;
+	buffer[3] = n & 0xff;
+}
+
+static inline void
+_header_to_message(const struct remote_message_header * header, uint8_t * message) {
+	to_bigendian(message , header->source);
+	to_bigendian(message+4 , header->destination);
+	to_bigendian(message+8 , header->session);
+}
+
+static inline uint32_t
+from_bigendian(uint32_t n) {
+	union {
+		uint32_t big;
+		uint8_t bytes[4];
+	} u;
+	u.big = n;
+	return u.bytes[0] << 24 | u.bytes[1] << 16 | u.bytes[2] << 8 | u.bytes[3];
+}
+
+static inline void
+_message_to_header(const uint32_t *message, struct remote_message_header *header) {
+	header->source = from_bigendian(message[0]);
+	header->destination = from_bigendian(message[1]);
+	header->session = from_bigendian(message[2]);
+}
+
+static void
+_dispatch_queue(struct dummy *h, struct msg_queue * queue, uint32_t handle,  const char name[GLOBALNAME_LENGTH] ) {
+	struct msg * m = _pop_queue(queue);
+	while (m) {
+		struct remote_message_header cookie;
+		uint8_t *ptr = m->buffer + m->size - sizeof(cookie);
+		memcpy(&cookie, ptr, sizeof(cookie));
+		int type = cookie.destination >> HANDLE_REMOTE_SHIFT;
+		skynet_send(h->ctx, cookie.source, handle , type | PTYPE_TAG_DONTCOPY, cookie.session, m->buffer, m->size - sizeof(cookie));
+		m = _pop_queue(queue);
+	}
+}
+
+static void
+_update_name(struct dummy *h, const char name[GLOBALNAME_LENGTH], uint32_t handle) {
+	struct keyvalue * node = _hash_search(h->map, name);
+	if (node == NULL) {
+		node = _hash_insert(h->map, name);
+	}
+	node->value = handle;
+	if (node->queue) {
+		_dispatch_queue(h, node->queue, handle, name);
+		_release_queue(node->queue);
+		node->queue = NULL;
+	}
+}
+
+static void
+_send_name(struct dummy *h, uint32_t source, const char name[GLOBALNAME_LENGTH], int type, int session, const char * msg, size_t sz) {
+	struct keyvalue * node = _hash_search(h->map, name);
+	if (node == NULL) {
+		node = _hash_insert(h->map, name);
+	}
+	if (node->value == 0) {
+		if (node->queue == NULL) {
+			node->queue = _new_queue();
+		}
+		struct remote_message_header header;
+		header.source = source;
+		header.destination = type << HANDLE_REMOTE_SHIFT;
+		header.session = (uint32_t)session;
+		_push_queue(node->queue, msg, sz, &header);
+	} else {
+		// local message
+		skynet_send(h->ctx, source, node->value , type | PTYPE_TAG_DONTCOPY, session, (void *)msg, sz);
+	}
+}
+
+static int
+_mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
+	struct dummy * h = ud;
+	switch (type) {
+	case PTYPE_SYSTEM: {
+		// register name message
+		const struct remote_message *rmsg = msg;
+		assert (sz == sizeof(rmsg->destination));
+		_update_name(h, rmsg->destination.name, rmsg->destination.handle);
+		return 0;
+	}
+	default: {
+		// remote message out
+		const struct remote_message *rmsg = msg;
+		if (rmsg->destination.handle == 0) {
+			_send_name(h, source , rmsg->destination.name, type, session, rmsg->message, rmsg->sz);
+		} else {
+			// local message
+			skynet_send(context, source, rmsg->destination.handle , type | PTYPE_TAG_DONTCOPY, session, (void *)rmsg->message, rmsg->sz);
+		}
+		return 0;
+	}
+	}
+}
+
+int
+dummy_init(struct dummy *d, struct skynet_context *ctx, const char * args) {
+	d->ctx = ctx;
+	skynet_harbor_start(ctx);
+	skynet_callback(ctx, d, _mainloop);
+
+	return 0;
+}

service-src/service_logger.c

@@ -28,7 +28,7 @@ logger_release(struct logger * inst) {
 static int
 _logger(struct skynet_context * context, void *ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
 	struct logger * inst = ud;
-	fprintf(inst->handle, "[:%x] ",source);
+	fprintf(inst->handle, "[:%08x] ",source);
 	fwrite(msg, sz , 1, inst->handle);
 	fprintf(inst->handle, "\n");
 	fflush(inst->handle);

service/bootstrap.lua

@@ -4,17 +4,24 @@ skynet.start(function()
 	assert(skynet.launch("logger", skynet.getenv "logger"))
 
 	local standalone = skynet.getenv "standalone"
-	local master_addr = skynet.getenv "master"
+	local harbor_id = tonumber(skynet.getenv "harbor")
+	if harbor_id == 0 then
+		assert(standalone ==  nil)
+		standalone = true
+		skynet.setenv("standalone", "true")
+		assert(skynet.launch("dummy"))
+	else
+		local master_addr = skynet.getenv "master"
 
-	if standalone then
-		assert(skynet.launch("master", master_addr))
+		if standalone then
+			assert(skynet.launch("master", master_addr))
+		end
+
+		local local_addr = skynet.getenv "address"
+
+		assert(skynet.launch("harbor",master_addr, local_addr, harbor_id))
 	end
 
-	local local_addr = skynet.getenv "address"
-	local harbor_id = skynet.getenv "harbor"
-
-	assert(skynet.launch("harbor",master_addr, local_addr, harbor_id))
-
 	local launcher = assert(skynet.launch("snlua","launcher"))
 	skynet.name(".launcher", launcher)
 

service/clusterd.lua

@@ -0,0 +1,69 @@
+local skynet = require "skynet"
+local sc = require "socketchannel"
+local socket = require "socket"
+local cluster = require "cluster.c"
+
+local config_name = skynet.getenv "cluster"
+local node_address = {}
+assert(loadfile(config_name, "t", node_address))()
+local node_session = {}
+local command = {}
+
+local function read_response(sock)
+	local sz = sock:read(2)
+	local msg = sock:read(sz)
+	return cluster.unpackresponse(msg)	-- session, ok, data
+end
+
+local function open_channel(t, key)
+	local host, port = string.match(node_address[key], "([^:]+):(.*)$")
+	local c = sc.channel {
+		host = host,
+		port = tonumber(port),
+		response = read_response,
+	}
+	assert(c:connect(true))
+	t[key] = c
+	node_session[key] = 1
+	return c
+end
+
+local node_channel = setmetatable({}, { __index = open_channel })
+
+function command.listen(source, addr, port)
+	local gate = skynet.newservice("gate")
+	skynet.call(gate, "lua", "open", { address = addr, port = port })
+	skynet.ret(skynet.pack(nil))
+end
+
+function command.req(source, node, addr, msg, sz)
+	local request
+	local c = node_channel[node]
+	local session = node_session[node]
+	-- msg is a local pointer, cluster.packrequest will free it
+	request, node_session[node] = cluster.packrequest(addr, session , msg, sz)
+	skynet.ret(c:request(request, session))
+end
+
+local request_fd = {}
+
+function command.socket(source, subcmd, fd, msg)
+	if subcmd == "data" then
+		local addr, session, msg = cluster.unpackrequest(msg)
+		local msg, sz = skynet.rawcall(addr, "lua", msg)
+		local response = cluster.packresponse(session, msg, sz)
+		socket.write(fd, response)
+	elseif subcmd == "open" then
+		skynet.error(string.format("socket accept from %s", msg))
+		skynet.call(source, "lua", "accept", fd)
+	else
+		skynet.error(string.format("socket %s %d : %s", subcmd, fd, msg))
+	end
+end
+
+skynet.start(function()
+	skynet.dispatch("lua", function(_, source, cmd, ...)
+		local f = assert(command[cmd])
+		f(source, ...)
+	end)
+end)

skynet-src/skynet_daemon.c

@@ -0,0 +1,96 @@
+#include <stdio.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/file.h>
+#include <signal.h>
+#include <errno.h>
+#include <stdlib.h>
+
+#include "skynet_daemon.h"
+
+static int
+check_pid(const char *pidfile) {
+	int pid = 0;
+	FILE *f = fopen(pidfile,"r");
+	if (f == NULL)
+		return 0;
+	int n = fscanf(f,"%d", &pid);
+	fclose(f);
+
+	if (n !=1 || pid == 0 || pid == getpid()) {
+		return 0;
+	}
+
+	if (kill(pid, 0) && errno == ESRCH)
+		return 0;
+
+	return pid;
+}
+
+static int 
+write_pid(const char *pidfile) {
+	FILE *f;
+	int pid = 0;
+	int fd = open(pidfile, O_RDWR|O_CREAT, 0644);
+	if (fd == -1) {
+		fprintf(stderr, "Can't create %s.\n", pidfile);
+		return 0;
+	}
+	f = fdopen(fd, "r+");
+	if (f == NULL) {
+		fprintf(stderr, "Can't open %s.\n", pidfile);
+		return 0;
+	}
+
+	if (flock(fd, LOCK_EX|LOCK_NB) == -1) {
+		int n = fscanf(f, "%d", &pid);
+		fclose(f);
+		if (n != 1) {
+			fprintf(stderr, "Can't lock and read pidfile.\n");
+		} else {
+			fprintf(stderr, "Can't lock pidfile, lock is held by pid %d.\n", pid);
+		}
+		return 0;
+	}
+	
+	pid = getpid();
+	if (!fprintf(f,"%d\n", pid)) {
+		fprintf(stderr, "Can't write pid.\n");
+		close(fd);
+		return 0;
+	}
+	fflush(f);
+
+	return pid;
+}
+
+int
+daemon_init(const char *pidfile) {
+	int pid = check_pid(pidfile);
+
+	if (pid) {
+		fprintf(stderr, "Skynet is already running, pid = %d.\n", pid);
+		return 1;
+	}
+
+#ifdef __APPLE__
+	fprintf(stderr, "'daemon' is deprecated: first deprecated in OS X 10.5 , use launchd instead.\n");
+#else
+	if (daemon(1,0)) {
+		fprintf(stderr, "Can't daemonize.\n");
+		return 1;
+	}
+#endif
+
+	pid = write_pid(pidfile);
+	if (pid == 0) {
+		return 1;
+	}
+
+	return 0;
+}
+
+int 
+daemon_exit(const char *pidfile) {
+	return unlink(pidfile);
+}

skynet-src/skynet_daemon.h

@@ -0,0 +1,7 @@
+#ifndef skynet_daemon_h
+#define skynet_daemon_h
+
+int daemon_init(const char *pidfile);
+int daemon_exit(const char *pidfile);
+
+#endif

skynet-src/skynet_harbor.c

@@ -7,18 +7,20 @@
 #include <assert.h>
 
 static struct skynet_context * REMOTE = 0;
-static unsigned int HARBOR = 0;
+static unsigned int HARBOR = ~0;
 
 void 
 skynet_harbor_send(struct remote_message *rmsg, uint32_t source, int session) {
 	int type = rmsg->sz >> HANDLE_REMOTE_SHIFT;
 	rmsg->sz &= HANDLE_MASK;
-	assert(type != PTYPE_SYSTEM && type != PTYPE_HARBOR);
+	assert(type != PTYPE_SYSTEM && type != PTYPE_HARBOR && REMOTE);
 	skynet_context_send(REMOTE, rmsg, sizeof(*rmsg) , source, type , session);
 }
 
 void 
 skynet_harbor_register(struct remote_name *rname) {
+	if (REMOTE == NULL)
+		return;
 	int i;
 	int number = 1;
 	for (i=0;i<GLOBALNAME_LENGTH;i++) {
@@ -34,7 +36,7 @@ skynet_harbor_register(struct remote_name *rname) {
 
 int 
 skynet_harbor_message_isremote(uint32_t handle) {
-	assert(HARBOR != 0);
+	assert(HARBOR != ~0);
 	int h = (handle & ~HANDLE_MASK);
 	return h != HARBOR && h !=0;
 }

skynet-src/skynet_imp.h

@@ -4,6 +4,7 @@
 struct skynet_config {
 	int thread;
 	int harbor;
+	const char * daemon;
 	const char * module_path;
 	const char * bootstrap;
 };

skynet-src/skynet_main.c

@@ -35,6 +35,7 @@ optboolean(const char *key, int opt) {
 	return strcmp(str,"true")==0;
 }
 */
+
 static const char *
 optstring(const char *key,const char * opt) {
 	const char * str = skynet_getenv(key);
@@ -109,21 +110,16 @@ main(int argc, char *argv[]) {
 	} 
 	_init_env(L);
 
-#ifdef LUA_CACHELIB
-  printf("Skynet lua code cache enable\n");
-#endif
-
 	config.thread =  optint("thread",8);
 	config.module_path = optstring("cpath","./cservice/?.so");
 	config.harbor = optint("harbor", 1);
 	config.bootstrap = optstring("bootstrap","snlua bootstrap");
+	config.daemon = optstring("daemon", NULL);
 
 	lua_close(L);
 
 	skynet_start(&config);
 	skynet_globalexit();
 
-	printf("skynet exit\n");
-
 	return 0;
 }

skynet-src/skynet_mq.c

@@ -34,7 +34,6 @@ struct global_queue {
 	struct message_queue ** queue;
 	// We use a separated flag array to ensure the mq is pushed.
 	// See the comments below.
-	bool * flag;
 	struct message_queue *list;
 };
 
@@ -45,11 +44,14 @@ static struct global_queue *Q = NULL;
 
 #define GP(p) ((p) % MAX_GLOBAL_MQ)
 
-static void 
+void 
 skynet_globalmq_push(struct message_queue * queue) {
 	struct global_queue *q= Q;
 
-	if (q->flag[GP(q->tail)]) {
+	uint32_t tail = GP(__sync_fetch_and_add(&q->tail,1));
+
+	// only one thread can set the slot (change q->queue[tail] from NULL to queue)
+	if (!__sync_bool_compare_and_swap(&q->queue[tail], NULL, queue)) {
 		// The queue may full seldom, save queue in list
 		assert(queue->next == NULL);
 		struct message_queue * last;
@@ -60,14 +62,6 @@ skynet_globalmq_push(struct message_queue * queue) {
 
 		return;
 	}
-
-	uint32_t tail = GP(__sync_fetch_and_add(&q->tail,1));
-	// The thread would suspend here, and the q->queue[tail] is last version ,
-	// but the queue tail is increased.
-	// So we set q->flag[tail] after changing q->queue[tail].
-	q->queue[tail] = queue;
-	__sync_synchronize();
-	q->flag[tail] = true;
 }
 
 struct message_queue * 
@@ -93,18 +87,18 @@ skynet_globalmq_pop() {
 		}
 	}
 
-	// Check the flag first, if the flag is false, the pushing may not complete.
-	if(!q->flag[head_ptr]) {
+	struct message_queue * mq = q->queue[head_ptr];
+	if (mq == NULL) {
+		// globalmq push not complete
 		return NULL;
 	}
-
-	__sync_synchronize();
-
-	struct message_queue * mq = q->queue[head_ptr];
 	if (!__sync_bool_compare_and_swap(&q->head, head, head+1)) {
 		return NULL;
 	}
-	q->flag[head_ptr] = false;
+	// only one thread can get the slot (change q->queue[head_ptr] to NULL)
+	if (!__sync_bool_compare_and_swap(&q->queue[head_ptr], mq, NULL)) {
+		return NULL;
+	}
 
 	return mq;
 }
@@ -220,26 +214,10 @@ skynet_mq_init() {
 	struct global_queue *q = skynet_malloc(sizeof(*q));
 	memset(q,0,sizeof(*q));
 	q->queue = skynet_malloc(MAX_GLOBAL_MQ * sizeof(struct message_queue *));
-	q->flag = skynet_malloc(MAX_GLOBAL_MQ * sizeof(bool));
-	memset(q->flag, 0, sizeof(bool) * MAX_GLOBAL_MQ);
+	memset(q->queue, 0, sizeof(struct message_queue *) * MAX_GLOBAL_MQ);
 	Q=q;
 }
 
-void 
-skynet_mq_force_push(struct message_queue * queue) {
-	assert(queue->in_global);
-	skynet_globalmq_push(queue);
-}
-
-void 
-skynet_mq_pushglobal(struct message_queue *queue) {
-	LOCK(queue)
-	assert(queue->in_global);
-	skynet_globalmq_push(queue);
-	queue->in_global = MQ_IN_GLOBAL;
-	UNLOCK(queue)
-}
-
 void 
 skynet_mq_mark_release(struct message_queue *q) {
 	LOCK(q)
@@ -268,7 +246,7 @@ skynet_mq_release(struct message_queue *q, message_drop drop_func, void *ud) {
 		UNLOCK(q)
 		_drop_queue(q, drop_func, ud);
 	} else {
-		skynet_mq_force_push(q);
+		skynet_globalmq_push(q);
 		UNLOCK(q)
 	}
 }

skynet-src/skynet_mq.h

@@ -13,6 +13,7 @@ struct skynet_message {
 
 struct message_queue;
 
+void skynet_globalmq_push(struct message_queue * queue);
 struct message_queue * skynet_globalmq_pop(void);
 
 struct message_queue * skynet_mq_create(uint32_t handle);
@@ -30,9 +31,6 @@ 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);
 
-void skynet_mq_force_push(struct message_queue *q);
-void skynet_mq_pushglobal(struct message_queue *q);
-
 void skynet_mq_init();
 
 #endif

skynet-src/skynet_server.c

@@ -139,7 +139,7 @@ skynet_context_new(const char * name, const char *param) {
 		if (ret) {
 			ctx->init = true;
 		}
-		skynet_mq_force_push(queue);
+		skynet_globalmq_push(queue);
 		if (ret) {
 			skynet_error(ret, "LAUNCH %s %s", name, param ? param : "");
 		}
@@ -228,11 +228,13 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
 	CHECKCALLING_END(ctx)
 }
 
-int
-skynet_context_message_dispatch(struct skynet_monitor *sm) {
-	struct message_queue * q = skynet_globalmq_pop();
-	if (q==NULL)
-		return 1;
+struct message_queue * 
+skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q) {
+	if (q == NULL) {
+		q = skynet_globalmq_pop();
+		if (q==NULL)
+			return NULL;
+	}
 
 	uint32_t handle = skynet_mq_handle(q);
 
@@ -240,13 +242,13 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) {
 	if (ctx == NULL) {
 		struct drop_t d = { handle };
 		skynet_mq_release(q, drop_message, &d);
-		return 0;
+		return skynet_globalmq_pop();
 	}
 
 	struct skynet_message msg;
 	if (skynet_mq_pop(q,&msg)) {
 		skynet_context_release(ctx);
-		return 0;
+		return skynet_globalmq_pop();
 	}
 
 	skynet_monitor_trigger(sm, msg.source , handle);
@@ -258,12 +260,18 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) {
 	}
 
 	assert(q == ctx->queue);
-	skynet_mq_pushglobal(q);
+	struct message_queue *nq = skynet_globalmq_pop();
+	if (nq) {
+		// If global mq is not empty , push q back, and return next queue (nq)
+		// Else (global mq is empty or block, don't push q back, and return q again (for next dispatch)
+		skynet_globalmq_push(q);
+		q = nq;
+	} 
 	skynet_context_release(ctx);
 
 	skynet_monitor_trigger(sm, 0,0);
 
-	return 0;
+	return q;
 }
 
 static void

skynet-src/skynet_server.h

@@ -16,7 +16,7 @@ void skynet_context_init(struct skynet_context *, uint32_t handle);
 int skynet_context_push(uint32_t handle, struct skynet_message *message);
 void skynet_context_send(struct skynet_context * context, void * msg, size_t sz, uint32_t source, int type, int session);
 int skynet_context_newsession(struct skynet_context *);
-int skynet_context_message_dispatch(struct skynet_monitor *);	// return 1 when block
+struct message_queue * skynet_context_message_dispatch(struct skynet_monitor *, struct message_queue *);	// return next queue
 int skynet_context_total();
 
 void skynet_context_endless(uint32_t handle);	// for monitor

skynet-src/skynet_start.c

@@ -7,6 +7,7 @@
 #include "skynet_timer.h"
 #include "skynet_monitor.h"
 #include "skynet_socket.h"
+#include "skynet_daemon.h"
 
 #include <pthread.h>
 #include <unistd.h>
@@ -120,8 +121,10 @@ _worker(void *p) {
 	struct monitor *m = wp->m;
 	struct skynet_monitor *sm = m->m[id];
 	skynet_initthread(THREAD_WORKER);
+	struct message_queue * q = NULL;
 	for (;;) {
-		if (skynet_context_message_dispatch(sm)) {
+		q = skynet_context_message_dispatch(sm, q);
+		if (q == NULL) {
 			CHECK_ABORT
 			if (pthread_mutex_lock(&m->mutex) == 0) {
 				++ m->sleep;
@@ -195,6 +198,11 @@ bootstrap(const char * cmdline) {
 
 void 
 skynet_start(struct skynet_config * config) {
+	if (config->daemon) {
+		if (daemon_init(config->daemon)) {
+			exit(1);
+		}
+	}
 	skynet_harbor_init(config->harbor);
 	skynet_handle_init(config->harbor);
 	skynet_mq_init();
@@ -206,4 +214,7 @@ skynet_start(struct skynet_config * config) {
 
 	_start(config->thread);
 	skynet_socket_free();
+	if (config->daemon) {
+		daemon_exit(config->daemon);
+	}
 }

skynet-src/skynet_timer.c

@@ -16,6 +16,9 @@
 
 typedef void (*timer_execute_func)(void *ud,void *arg);
 
+#define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {}
+#define UNLOCK(q) __sync_lock_release(&(q)->lock);
+
 #define TIME_NEAR_SHIFT 8
 #define TIME_NEAR (1 << TIME_NEAR_SHIFT)
 #define TIME_LEVEL_SHIFT 6
@@ -50,8 +53,7 @@ struct timer {
 static struct timer * TI = NULL;
 
 static inline struct timer_node *
-link_clear(struct link_list *list)
-{
+link_clear(struct link_list *list) {
 	struct timer_node * ret = list->head.next;
 	list->head.next = 0;
 	list->tail = &(list->head);
@@ -60,23 +62,20 @@ link_clear(struct link_list *list)
 }
 
 static inline void
-link(struct link_list *list,struct timer_node *node)
-{
+link(struct link_list *list,struct timer_node *node) {
 	list->tail->next = node;
 	list->tail = node;
 	node->next=0;
 }
 
 static void
-add_node(struct timer *T,struct timer_node *node)
-{
+add_node(struct timer *T,struct timer_node *node) {
 	int time=node->expire;
 	int current_time=T->time;
 	
 	if ((time|TIME_NEAR_MASK)==(current_time|TIME_NEAR_MASK)) {
 		link(&T->near[time&TIME_NEAR_MASK],node);
-	}
-	else {
+	} else {
 		int i;
 		int mask=TIME_NEAR << TIME_LEVEL_SHIFT;
 		for (i=0;i<3;i++) {
@@ -90,21 +89,21 @@ add_node(struct timer *T,struct timer_node *node)
 }
 
 static void
-timer_add(struct timer *T,void *arg,size_t sz,int time)
-{
+timer_add(struct timer *T,void *arg,size_t sz,int time) {
 	struct timer_node *node = (struct timer_node *)skynet_malloc(sizeof(*node)+sz);
 	memcpy(node+1,arg,sz);
 
-	while (__sync_lock_test_and_set(&T->lock,1)) {};
+	LOCK(T);
 
 		node->expire=time+T->time;
 		add_node(T,node);
 
-	__sync_lock_release(&T->lock);
+	UNLOCK(T);
 }
 
 static void
 timer_shift(struct timer *T) {
+	LOCK(T);
 	int mask = TIME_NEAR;
 	int time = (++T->time) >> TIME_NEAR_SHIFT;
 	int i=0;
@@ -125,50 +124,57 @@ timer_shift(struct timer *T) {
 		time >>= TIME_LEVEL_SHIFT;
 		++i;
 	}	
+	UNLOCK(T);
+}
+
+static inline void
+dispatch_list(struct timer_node *current) {
+	do {
+		struct timer_event * event = (struct timer_event *)(current+1);
+		struct skynet_message message;
+		message.source = 0;
+		message.session = event->session;
+		message.data = NULL;
+		message.sz = PTYPE_RESPONSE << HANDLE_REMOTE_SHIFT;
+
+		skynet_context_push(event->handle, &message);
+		
+		struct timer_node * temp = current;
+		current=current->next;
+		skynet_free(temp);	
+	} while (current);
 }
 
 static inline void
 timer_execute(struct timer *T) {
+	LOCK(T);
 	int idx = T->time & TIME_NEAR_MASK;
 	
 	while (T->near[idx].head.next) {
 		struct timer_node *current = link_clear(&T->near[idx]);
-		
-		do {
-			struct timer_event * event = (struct timer_event *)(current+1);
-			struct skynet_message message;
-			message.source = 0;
-			message.session = event->session;
-			message.data = NULL;
-			message.sz = PTYPE_RESPONSE << HANDLE_REMOTE_SHIFT;
-
-			skynet_context_push(event->handle, &message);
-			
-			struct timer_node * temp = current;
-			current=current->next;
-			skynet_free(temp);	
-		} while (current);
+		UNLOCK(T);
+		// dispatch_list don't need lock T
+		dispatch_list(current);
+		LOCK(T);
 	}
+
+	UNLOCK(T);
 }
 
 static void 
-timer_update(struct timer *T)
-{
-	while (__sync_lock_test_and_set(&T->lock,1)) {};
-
+timer_update(struct timer *T) {
 	// try to dispatch timeout 0 (rare condition)
 	timer_execute(T);
 
 	// shift time first, and then dispatch timer message
 	timer_shift(T);
+
 	timer_execute(T);
 
-	__sync_lock_release(&T->lock);
 }
 
 static struct timer *
-timer_create_timer()
-{
+timer_create_timer() {
 	struct timer *r=(struct timer *)skynet_malloc(sizeof(struct timer));
 	memset(r,0,sizeof(*r));