skynet harbor

Makefile

@@ -1,10 +1,7 @@
-all : skynet blackhole.so snlua.so logger.so skynet.so gate.so client
+all : skynet snlua.so logger.so skynet.so gate.so client skynet-master
 
-skynet : skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c skynet_server.c skynet_start.c skynet_timer.c skynet_error.c
-	gcc -Wall -g -Wl,-E -o $@ $^ -lpthread -ldl -lrt -Wl,-E -llua -lm
-
-blackhole.so : skynet_blackhole.c
-	gcc -Wall -g -fPIC --shared $^ -o $@
+skynet : 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
+	gcc -Wall -g -Wl,-E -o $@ $^ -lpthread -ldl -lrt -Wl,-E -llua -lm -lzmq
 
 logger.so : skynet_logger.c
 	gcc -Wall -g -fPIC --shared $^ -o $@
@@ -21,6 +18,9 @@ skynet.so : lua-skynet.c
 client : client.c
 	gcc -Wall -g $^ -o $@
 
-clean :
-	rm skynet client *.so
+skynet-master : master/master.c
+	gcc -g -Wall -o $@ $^ -lzmq
+
+clean :
+	rm skynet client skynet-master *.so
 

README.md

@@ -0,0 +1,18 @@
+## Build
+
+Install zeromq 2.2 and lua 5.2 first.
+
+```
+make
+```
+
+## Test
+
+Run these in different console
+
+```
+./skynet-master		# Launch master server first
+./skynet config		# Launch first skynet node  (Gate server)
+./skynet config_log	# Launch second skynet node (Global logger server)
+./clinet 127.0.0.1 8888	# Launch a client, and try to input some words.
+```

config

@@ -2,3 +2,7 @@ thread = 8
 mqueue = 256
 cpath = "./"
 logger = nil
+harbor = 1
+address = "tcp://127.0.0.1:2525"
+master = "tcp://127.0.0.1:2012"
+start = "main.lua"

config_log

@@ -0,0 +1,8 @@
+thread = 2
+mqueue = 256
+cpath = "./"
+logger = nil
+harbor = 2
+address = "tcp://127.0.0.1:2526"
+master = "tcp://127.0.0.1:2012"
+start = "main_log.lua"

globallog.lua

@@ -0,0 +1,7 @@
+local skynet = require "skynet"
+
+skynet.callback(function(from , message)
+	print("[GLOBALLOG]",from,message)
+end)
+
+skynet.command("REG","LOG")

main.lua

@@ -6,7 +6,7 @@ local console = skynet.command("LAUNCH","snlua console.lua")
 print("console",console)
 local watchdog = skynet.command("LAUNCH","snlua watchdog.lua")
 print("watchdog",watchdog)
-local gate = skynet.command("LAUNCH","gate 2525 4 0")
+local gate = skynet.command("LAUNCH","gate 8888 4 0")
 print("gate",gate)
 
 skynet.command("EXIT")

main_log.lua

@@ -0,0 +1,8 @@
+local skynet = require "skynet"
+
+print("Log server start")
+
+local log = skynet.command("LAUNCH","snlua globallog.lua")
+print("log",log)
+
+skynet.command("EXIT")

master/Makefile

@@ -0,0 +1,7 @@
+all : master test
+
+master : master.c
+	gcc -g -Wall -o $@ $^ -lzmq
+
+test : test.c
+	gcc -g -Wall -o $@ $^ -lzmq

master/master.c

@@ -0,0 +1,250 @@
+#include <zmq.h>
+#include <stdio.h>
+#include <string.h>
+#include <stdlib.h>
+#include <stdint.h>
+
+#define HASH_SIZE 4096
+#define MAX_SLAVE 255
+
+static uint32_t
+calc_hash(const char *name) {
+	int i;
+	uint32_t h = 0;
+	for (i=0;name[i];i++) {
+	    h = h ^ ((h<<5)+(h>>2)+(uint8_t)name[i]);
+	}
+	h ^= i;
+	return h;
+}
+
+struct keyvalue {
+	struct keyvalue * next;
+	uint32_t hash;
+	char * key;
+	size_t value_size;
+	char * value;
+};
+
+struct hashmap {
+	struct keyvalue *node[HASH_SIZE];
+	void * zmq;
+	void * slave[MAX_SLAVE];
+};
+
+static struct hashmap *
+_hash_new(void *zmq) {
+	struct hashmap * hash = malloc(sizeof(*hash));
+	memset(hash, 0, sizeof(*hash));
+	hash->zmq = zmq;
+	return hash;
+}
+
+static struct keyvalue *
+_hash_search(struct hashmap * hash, const char * key) {
+	uint32_t h = calc_hash(key);
+	struct keyvalue * n = hash->node[h & (HASH_SIZE-1)];
+	while (n) {
+		if (n->hash == h && strcmp(n->key, key) == 0) {
+			return n;
+		}
+		n = n->next;
+	}
+	return NULL;
+}
+
+static void
+_hash_insert(struct hashmap * hash, const char * key, const char *value) {
+	uint32_t h = calc_hash(key);
+	struct keyvalue * node = malloc(sizeof(*node));
+	node->next = hash->node[h & (HASH_SIZE-1)];
+	node->hash = h;
+	node->key = strdup(key);
+	node->value_size = strlen(value);
+	node->value = malloc(node->value_size+1);
+	memcpy(node->value, value, node->value_size+1);
+
+	hash->node[h & (HASH_SIZE-1)] = node;
+}
+
+static void
+_hash_delete(struct hashmap * hash, const char * key) {
+	uint32_t h = calc_hash(key);
+	struct keyvalue ** ptr = &hash->node[h & (HASH_SIZE-1)];
+	while(*ptr) {
+		struct keyvalue *n = *ptr;
+		if (n->hash == h && strcmp(n->key, key) == 0) {
+			*ptr = n->next;
+			free(n->key);
+			free(n->value);
+			free(n);
+			return ;
+		}
+		ptr = &(n->next);
+	}
+}
+
+static int
+_hash_bind(struct hashmap * hash, int slave, const char *addr) {
+	void *pub = NULL;
+	if (addr) {
+		pub = zmq_socket(hash->zmq, ZMQ_PUSH);
+		int r = zmq_connect(pub, addr);
+		if (r<0) {
+			fprintf(stderr,"Can't connect to [%d] %s\n",slave,addr);
+			return -1;
+		}
+		printf("Connect to [%d] %s\n",slave,addr);
+	}
+	void * old_pub = hash->slave[slave-1];
+	hash->slave[slave-1] = pub;
+	if (old_pub) {
+		zmq_close(old_pub);
+	}
+
+	return 0;
+}
+
+static void
+broadcast(struct hashmap * hash, zmq_msg_t * msg) {
+	int i;
+	for (i=0;i<MAX_SLAVE;i++) {
+		void * pub = hash->slave[i];
+		if (pub) {
+			zmq_msg_t part;
+			zmq_msg_init(&part);
+			int rc = zmq_send(pub, &part, ZMQ_SNDMORE);
+			if (rc != 0) {
+				fprintf(stderr,"Can't publish to %d : %s",i+1,zmq_strerror(errno));
+			}
+			zmq_msg_close(&part);
+			zmq_msg_init(&part);
+			zmq_msg_copy(&part,msg);
+			rc = zmq_send(pub, &part, 0);
+			if (rc != 0) {
+				fprintf(stderr,"Can't publish to %d : %s",i+1,zmq_strerror(errno));
+			}
+			zmq_msg_close(&part);
+		}
+	}
+}
+
+static void
+replace(void * responder, struct hashmap * map, const char *key, const char *value) {
+//	printf("Replace %s %s\n",key,value);
+	struct keyvalue * node = _hash_search(map, key);
+	zmq_msg_t reply;
+	if (node == NULL) {
+		_hash_insert(map, key, value);
+		zmq_msg_init_size (&reply, 0);
+	} else {
+		zmq_msg_init_size (&reply, node->value_size);
+		memcpy (zmq_msg_data (&reply), node->value, node->value_size);
+
+		free(node->value);
+		node->value_size = strlen(value);
+		node->value = malloc(node->value_size + 1);
+		memcpy(node->value, value, node->value_size +1);
+	}
+	zmq_send (responder, &reply, 0);
+	zmq_msg_close (&reply);
+}
+
+static void
+erase(void * responder, struct hashmap *map, const char *key) {
+//	printf("Erase %s\n",key);
+	struct keyvalue * node = _hash_search(map, key);
+	zmq_msg_t reply;
+	if (node == NULL) {
+		zmq_msg_init_size (&reply, 0);
+	} else {
+		zmq_msg_init_size (&reply, node->value_size);
+		memcpy (zmq_msg_data (&reply), node->value, node->value_size);
+		_hash_delete(map, key);
+	}
+	zmq_send (responder, &reply, 0);
+	zmq_msg_close (&reply);
+}
+
+static void
+query(void * responder, struct hashmap *map, const char * key) {
+	struct keyvalue * node = _hash_search(map, key);
+	zmq_msg_t reply;
+	if (node == NULL) {
+		zmq_msg_init_size (&reply, 0);
+	} else {
+		zmq_msg_init_size (&reply, node->value_size);
+		memcpy (zmq_msg_data (&reply), node->value, node->value_size);
+	}
+	zmq_send (responder, &reply, 0);
+	zmq_msg_close (&reply);
+}
+
+static void
+update(void * responder, struct hashmap *map, zmq_msg_t * msg) {
+	size_t sz = zmq_msg_size (msg);
+	const char * command = zmq_msg_data (msg);
+	int i;
+	for (i=0;i<sz;i++) {
+		if (command[i] == '=') {
+			char key[i+1];
+			memcpy(key, command, i);
+			key[i] = '\0';
+			int slave = strtol(key, NULL, 10);
+			if (sz-i == 1) {
+				if (slave > 0 && slave <= MAX_SLAVE) {
+					_hash_bind(map, slave, NULL);
+				}
+				erase(responder, map,key);
+				broadcast(map, msg);
+			} else {
+				char value[sz-i];
+				memcpy(value, command+i+1, sz-i-1);
+				value[sz-i-1] = '\0';
+				if (slave > 0 && slave <= MAX_SLAVE) {
+					_hash_bind(map, slave, value);
+				}
+
+				replace(responder, map, key,value);
+				broadcast(map, msg);
+			}
+			return;
+		}
+	}
+	char key[sz+1];
+	memcpy(key, command, sz);
+	key[sz] = '\0';
+	query(responder, map, key);
+}
+
+int 
+main (int argc, char * argv[]) {
+	const char * default_port = "tcp://127.0.0.1:2012";
+	if (argc > 1) {
+		default_port = argv[2];
+	}
+
+	void *context = zmq_init (1);
+	void *responder = zmq_socket (context, ZMQ_REP);
+
+	int r = zmq_bind(responder, default_port);
+	if (r < 0) {
+		fprintf(stderr, "Can't bind to %s\n",default_port);
+		return 1;
+	}
+	printf("Start master on %s\n",default_port);
+
+	struct hashmap *map = _hash_new(context);
+
+	for (;;) {
+		zmq_msg_t request;
+		zmq_msg_init (&request);
+		zmq_recv (responder, &request, 0);
+		update(responder, map, &request);
+		zmq_msg_close (&request);
+	}
+
+	zmq_close (responder);
+	zmq_term (context);
+	return 0;
+}

master/test.c

@@ -0,0 +1,124 @@
+#include <zmq.h>
+#include <unistd.h>
+#include <stdio.h>
+#include <string.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <assert.h>
+
+static char *
+send_command(void * request, const char * command) {
+	size_t size = strlen(command);
+	zmq_msg_t req;
+	zmq_msg_init_size(&req, size);
+	memcpy(zmq_msg_data(&req) , command, size);
+	zmq_send (request, &req, 0);
+	zmq_msg_close (&req);
+
+	zmq_msg_t reply;
+	zmq_msg_init (&reply);
+	zmq_recv (request, &reply, 0);
+
+	size = zmq_msg_size (&reply);
+
+	char * ret = malloc(size+1);
+	memcpy(ret,zmq_msg_data(&reply),size);
+	ret[size]='\0';
+	zmq_msg_close (&reply);
+
+	return ret;
+}
+
+static void
+pull_sub(void * local) {
+	zmq_msg_t part;
+	zmq_msg_init (&part);
+	zmq_recv (local, &part, 0);
+	zmq_msg_close(&part);
+	zmq_msg_init (&part);
+	zmq_recv (local, &part, 0);
+	size_t sz = zmq_msg_size(&part);
+	char tmp[sz+1];
+	memcpy(tmp,zmq_msg_data(&part),sz);
+	tmp[sz]='\0';
+	printf("%s\n",tmp);
+	zmq_msg_close(&part);
+}
+
+int 
+main (int argc, char * argv[]) {
+	const char * default_master = "tcp://127.0.0.1:2012";
+	int slave = 100;
+	const char * default_local = "tcp://127.0.0.1:5001";
+	if (argc > 1) {
+		default_master = argv[1];
+		if (argc > 2) {
+			slave = strtol(argv[2],NULL,10);
+			if (slave < 1 || slave> 255) {
+				fprintf(stderr,"Slave id must be in [1,255]\n");
+				return 1;
+			}
+			if (argc > 3) {
+				default_local = argv[3];
+			}
+		}
+	}
+
+	printf("master = %s\n",default_master);
+	printf("local = [%d]%s\n",slave,default_local);
+
+	void *context = zmq_init (1);
+	void *request = zmq_socket (context, ZMQ_REQ);
+	void *local = zmq_socket( context, ZMQ_PULL);
+
+	int r = zmq_connect(request, default_master);
+	if (r < 0) {
+		fprintf(stderr, "Can't connect to %s\n",default_master);
+		return 1;
+	}
+	r = zmq_bind(local, default_local);
+	if (r < 0) {
+		fprintf(stderr, "Can't bind to %s\n",default_local);
+		return 1;
+	}
+
+	char tmp[1024];
+	sprintf(tmp,"%d=%s",slave,default_local);
+
+	char * result = send_command(request,tmp);
+	free(result);
+
+	zmq_pollitem_t items[2];
+	items[0].socket = local;
+	items[0].events = ZMQ_POLLIN;
+	items[1].socket = NULL;
+	items[1].fd = STDIN_FILENO;
+	items[1].events = ZMQ_POLLIN;
+
+	for (;;) {
+		int rc = zmq_poll(items,2,-1);
+		assert (rc >= 0);
+		if (items[0].revents) {
+			pull_sub(local);
+		}
+		if (items[1].revents) {
+			char tmp[1024];
+			fgets(tmp,sizeof(tmp),stdin);
+			char * cr = strchr(tmp,'\r');
+			if (cr) {
+				*cr = '\0';
+			}
+			cr = strchr(tmp,'\n');
+			if (cr) {
+				*cr = '\0';
+			}
+			result = send_command(request, tmp);
+			printf("%s\n",result);
+			free(result);
+		}
+	}
+
+	zmq_close (request);
+	zmq_term (context);
+	return 0;
+}

skynet_blackhole.c

@@ -1,22 +0,0 @@
-#include "skynet.h"
-#include "skynet_blackhole.h"
-
-#include <assert.h>
-#include <stdlib.h>
-#include <stdio.h>
-
-static void
-_log_message(struct skynet_context *ctx, void *ud, const char * uid, const void * msg, size_t sz) {
-	const struct blackhole * message = msg;
-	assert(sz == sizeof(*message));
-
-	printf("[blackhole] from %d to %s , message_size = %d\n",message->source, message->destination, (int)message->sz);
-	free(message->data);
-}
-
-int
-blackhole_init(void * inst, struct skynet_context *ctx, const char *args) {
-	skynet_callback(ctx, inst, _log_message);
-	skynet_command(ctx, "REG", "blackhole");
-	return 0;
-}

skynet_blackhole.h

@@ -1,13 +0,0 @@
-#ifndef SKYNET_BLACKHOLE_H
-#define SKYNET_BLACKHOLE_H
-
-#include <stdlib.h>
-
-struct blackhole {
-	int source;
-	char * destination;
-	void * data;
-	size_t sz;
-};
-
-#endif

skynet_error.c

@@ -2,6 +2,7 @@
 #include "skynet_handle.h"
 #include "skynet_mq.h"
 #include "skynet_server.h"
+#include "skynet_system.h"
 
 #include <stdarg.h>
 #include <stdio.h>
@@ -32,7 +33,11 @@ skynet_error(struct skynet_context * context, const char *msg, ...) {
 	}
 
 	struct skynet_message smsg;
-	smsg.source = skynet_context_handle(context);
+	if (context == NULL) {
+		smsg.source = SKYNET_SYSTEM_LOGGER;
+	} else {
+		smsg.source = skynet_context_handle(context);
+	}
 	smsg.destination = logger;
 	smsg.data = strdup(tmp);
 	smsg.sz = len;

skynet_handle.c

@@ -10,13 +10,14 @@
 
 struct handle_name {
 	char * name;
-	int handle;
+	uint32_t handle;
 };
 
 struct handle_storage {
 	struct rwlock lock;
 
-	int handle_index;
+	uint32_t harbor;
+	uint32_t handle_index;
 	int slot_size;
 	struct skynet_context ** slot;
 	
@@ -27,7 +28,7 @@ struct handle_storage {
 
 static struct handle_storage *H = NULL;
 
-int 
+uint32_t
 skynet_handle_register(struct skynet_context *ctx) {
 	struct handle_storage *s = H;
 
@@ -39,14 +40,14 @@ skynet_handle_register(struct skynet_context *ctx) {
 			int hash = (i+s->handle_index) & (s->slot_size-1);
 			if (s->slot[hash] == NULL) {
 				s->slot[hash] = ctx;
-				int handle = s->handle_index + i;
+				uint32_t handle = s->handle_index + i ;
 				skynet_context_init(ctx, handle);
 
 				rwlock_wunlock(&s->lock);
 
 				s->handle_index = handle + 1;
 
-				return handle;
+				return (handle & HANDLE_MASK) | s->harbor;
 			}
 		}
 		struct skynet_context ** new_slot = malloc(s->slot_size * 2 * sizeof(struct skynet_context *));
@@ -63,12 +64,12 @@ skynet_handle_register(struct skynet_context *ctx) {
 }
 
 void
-skynet_handle_retire(int handle) {
+skynet_handle_retire(uint32_t handle) {
 	struct handle_storage *s = H;
 
 	rwlock_wlock(&s->lock);
 
-	int hash = handle & (s->slot_size-1);
+	uint32_t hash = handle & (s->slot_size-1);
 	struct skynet_context * ctx = s->slot[hash];
 	if (skynet_context_handle(ctx) == handle) {
 		skynet_context_release(ctx);
@@ -93,13 +94,13 @@ skynet_handle_retire(int handle) {
 }
 
 struct skynet_context * 
-skynet_handle_grab(int handle) {
+skynet_handle_grab(uint32_t handle) {
 	struct handle_storage *s = H;
 	struct skynet_context * result = NULL;
 
 	rwlock_rlock(&s->lock);
 
-	int hash = handle & (s->slot_size-1);
+	uint32_t hash = handle & (s->slot_size-1);
 	struct skynet_context * ctx = s->slot[hash];
 	if (skynet_context_handle(ctx) == handle) {
 		result = ctx;
@@ -111,13 +112,13 @@ skynet_handle_grab(int handle) {
 	return result;
 }
 
-int 
+uint32_t 
 skynet_handle_findname(const char * name) {
 	struct handle_storage *s = H;
 
 	rwlock_rlock(&s->lock);
 
-	int handle = -1;
+	uint32_t handle = 0;
 
 	int begin = 0;
 	int end = s->name_count - 1;
@@ -127,6 +128,7 @@ skynet_handle_findname(const char * name) {
 		int c = strcmp(n->name, name);
 		if (c==0) {
 			handle = n->handle;
+			handle |= s->harbor;
 			break;
 		}
 		if (c<0) {
@@ -142,7 +144,7 @@ skynet_handle_findname(const char * name) {
 }
 
 static void
-_insert_name_before(struct handle_storage *s, char *name, int handle, int before) {
+_insert_name_before(struct handle_storage *s, char *name, uint32_t handle, int before) {
 	if (s->name_count >= s->name_cap) {
 		s->name_cap *= 2;
 		struct handle_name * n = malloc(s->name_cap * sizeof(struct handle_name));
@@ -167,7 +169,7 @@ _insert_name_before(struct handle_storage *s, char *name, int handle, int before
 }
 
 static const char *
-_insert_name(struct handle_storage *s, const char * name, int handle) {
+_insert_name(struct handle_storage *s, const char * name, uint32_t handle) {
 	int begin = 0;
 	int end = s->name_count - 1;
 	while (begin<=end) {
@@ -191,7 +193,7 @@ _insert_name(struct handle_storage *s, const char * name, int handle) {
 }
 
 const char * 
-skynet_handle_namehandle(int handle, const char *name) {
+skynet_handle_namehandle(uint32_t handle, const char *name) {
 	rwlock_wlock(&H->lock);
 
 	const char * ret = _insert_name(H, name, handle);
@@ -202,7 +204,7 @@ skynet_handle_namehandle(int handle, const char *name) {
 }
 
 void 
-skynet_handle_init(void) {
+skynet_handle_init(int harbor) {
 	assert(H==NULL);
 	struct handle_storage * s = malloc(sizeof(*H));
 	s->slot_size = DEFAULT_SLOT_SIZE;
@@ -210,7 +212,9 @@ skynet_handle_init(void) {
 	memset(s->slot, 0, s->slot_size * sizeof(struct handle_slot *));
 
 	rwlock_init(&s->lock);
-	s->handle_index = 0;
+	// reserve 0 for system
+	s->harbor = (uint32_t) (harbor & 0xff) << HANDLE_REMOTE_SHIFT;
+	s->handle_index = 1;
 	s->name_cap = 2;
 	s->name_count = 0;
 	s->name = malloc(s->name_cap * sizeof(struct handle_name));

skynet_handle.h

@@ -1,15 +1,22 @@
 #ifndef SKYNET_CONTEXT_HANDLE_H
 #define SKYNET_CONTEXT_HANDLE_H
 
+#include <stdint.h>
+
+// reserve high 8 bits for remote id
+// see skynet_harbor.c REMOTE_MAX
+#define HANDLE_MASK 0xffffff
+#define HANDLE_REMOTE_SHIFT 24
+
 struct skynet_context;
 
-int skynet_handle_register(struct skynet_context *);
-void skynet_handle_retire(int handle);
-struct skynet_context * skynet_handle_grab(int handle);
+uint32_t skynet_handle_register(struct skynet_context *);
+void skynet_handle_retire(uint32_t handle);
+struct skynet_context * skynet_handle_grab(uint32_t handle);
 
-int skynet_handle_findname(const char * name);
-const char * skynet_handle_namehandle(int handle, const char *name);
+uint32_t skynet_handle_findname(const char * name);
+const char * skynet_handle_namehandle(uint32_t handle, const char *name);
 
-void skynet_handle_init(void);
+void skynet_handle_init(int harbor);
 
 #endif

skynet_harbor.c

@@ -0,0 +1,609 @@
+#include "skynet_harbor.h"
+#include "skynet_mq.h"
+#include "skynet_handle.h"
+#include "skynet_system.h"
+#include "skynet.h"
+
+#include <zmq.h>
+#include <string.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <assert.h>
+
+#define HASH_SIZE 4096
+#define DEFAULT_QUEUE_SIZE 1024
+
+// see skynet_handle.h for HANDLE_REMOTE_SHIFT
+#define REMOTE_MAX 255
+
+struct keyvalue {
+	struct keyvalue * next;
+	uint32_t hash;
+	char * key;
+	uint32_t value;
+	struct message_queue * queue;
+};
+
+struct hashmap {
+	struct keyvalue *node[HASH_SIZE];
+};
+
+struct remote_header {
+	uint32_t source;
+	uint32_t destination;
+};
+
+struct remote {
+	void *socket;
+	struct message_queue *queue;
+};
+
+struct harbor {
+	void * zmq_context;
+	void * zmq_master_request;
+	void * zmq_local;
+	void * zmq_queue_notice;
+	int notice_event;
+	struct hashmap *map;
+	struct remote remote[REMOTE_MAX];
+	struct message_queue *queue;
+	int harbor;
+
+	int lock;
+};
+
+static struct harbor *Z = NULL;
+
+// todo: optimize for little endian system
+
+static inline void
+buffer_to_remote_header(uint8_t *buffer, struct remote_header *header) {
+	header->source = buffer[0] | buffer[1] << 8 | buffer[2] << 16 | buffer[3] << 24;
+	header->destination = buffer[4] | buffer[5] << 8 | buffer[6] << 16 | buffer[7] << 24;
+}
+
+static inline void
+remote_header_to_buffer(struct remote_header *header, uint8_t *buffer) {
+	buffer[0] = header->source & 0xff;
+	buffer[1] = (header->source >> 8) & 0xff;
+	buffer[2] = (header->source >>16) & 0xff;
+	buffer[3] = (header->source >>24)& 0xff;
+	buffer[4] = (header->destination) & 0xff;
+	buffer[5] = (header->destination >>8) & 0xff;
+	buffer[6] = (header->destination >>16) & 0xff;
+	buffer[7] = (header->destination >>24) & 0xff;
+}
+
+static uint32_t
+calc_hash(const char *name) {
+	int i;
+	uint32_t h = 0;
+	for (i=0;name[i];i++) {
+	    h = h ^ ((h<<5)+(h>>2)+(uint8_t)name[i]);
+	}
+	h ^= i;
+	return h;
+}
+
+static inline void
+_lock() {
+	while (__sync_lock_test_and_set(&Z->lock,1)) {}
+}
+
+static inline void
+_unlock() {
+	__sync_lock_release(&Z->lock);
+}
+
+static struct hashmap *
+_hash_new(void) {
+	struct hashmap * hash = malloc(sizeof(*hash));
+	memset(hash, 0, sizeof(*hash));
+	return hash;
+}
+
+static struct keyvalue *
+_hash_search(struct hashmap * hash, const char * key) {
+	uint32_t h = calc_hash(key);
+	struct keyvalue * n = hash->node[h & (HASH_SIZE-1)];
+	while (n) {
+		if (n->hash == h && strcmp(n->key, key) == 0) {
+			return n;
+		}
+		n = n->next;
+	}
+	return NULL;
+}
+
+static void
+_hash_insert(struct hashmap * hash, const char * key, uint32_t handle, struct message_queue *queue) {
+	uint32_t h = calc_hash(key);
+	struct keyvalue * node = malloc(sizeof(*node));
+	node->next = hash->node[h & (HASH_SIZE-1)];
+	node->hash = h;
+	node->key = strdup(key);
+	node->value = handle;
+	node->queue = queue;
+
+	hash->node[h & (HASH_SIZE-1)] = node;
+}
+
+static void
+send_notice() {
+	if (__sync_lock_test_and_set(&Z->notice_event,1)) {
+		// already send notice
+		return;
+	}
+	static __thread void * queue_notice = NULL;
+	if (queue_notice == NULL) {
+		void * pub = zmq_socket(Z->zmq_context, ZMQ_PUSH);
+		int r = zmq_connect(pub , "inproc://notice");
+		assert(r==0);
+		queue_notice = pub;
+	}
+	zmq_msg_t dummy;
+	zmq_msg_init(&dummy);
+	zmq_send(queue_notice,&dummy,0);
+	zmq_msg_close(&dummy);
+}
+
+void 
+skynet_harbor_send(const char *name, struct skynet_message * message) {
+	if (name == NULL) {
+		int remote_id = message->destination >> HANDLE_REMOTE_SHIFT;
+		assert(remote_id > 0 && remote_id <= REMOTE_MAX);
+		--remote_id;
+		struct remote * r = &Z->remote[remote_id];
+		if (r->socket) {
+			skynet_mq_enter(Z->queue, message);
+			send_notice();
+		} else {
+			skynet_mq_enter(r->queue, message);
+		}
+	} else {
+		_lock();
+		struct keyvalue * node = _hash_search(Z->map, name);
+		if (node) {
+			uint32_t dest = node->value;
+			_unlock();
+			if (dest == 0) {
+				// push message to unknown name service queue
+				skynet_mq_enter(node->queue, message);
+			} else {
+				message->destination = dest;
+				if (!skynet_harbor_message_isremote(dest)) {
+					// local message
+					skynet_mq_push(message);
+					return;
+				}
+				skynet_mq_enter(Z->queue,message);
+				send_notice();
+			}
+		} else {
+			// never seen name before
+			struct message_queue * queue =  skynet_mq_create(DEFAULT_QUEUE_SIZE);
+			skynet_mq_enter(queue, message);
+			_hash_insert(Z->map, name, 0, queue);
+			_unlock();
+			// 0 for query
+			skynet_harbor_register(name,0);
+		}
+	}
+}
+
+//queue a register message (destination = 0)
+void 
+skynet_harbor_register(const char *name, uint32_t handle) {
+	struct skynet_message msg;
+	msg.source = handle;
+	msg.destination = SKYNET_SYSTEM_NAME;
+	msg.data = strdup(name);
+	msg.sz = 0;
+	skynet_mq_enter(Z->queue,&msg);
+	send_notice();
+}
+
+static void
+_register_name(const char *name, uint32_t addr) {
+	_lock();
+	struct keyvalue * node = _hash_search(Z->map, name);
+	if (node) {
+		if (node->value) {
+			node->value = addr;
+			assert(node->queue == NULL);
+		} else {
+			node->value = addr;
+		}
+	} else {
+		_hash_insert(Z->map, name, addr, NULL);
+	}
+
+	struct skynet_message msg;
+	struct message_queue * queue = node ? node->queue : NULL;
+
+	if (queue) {
+		if (skynet_harbor_message_isremote(addr)) {
+			while (skynet_mq_leave(queue, &msg)) {
+				msg.destination = addr;
+				skynet_mq_enter(Z->queue, &msg);
+			}
+		} else {
+			while (skynet_mq_leave(queue, &msg)) {
+				msg.destination = addr;
+				skynet_mq_push(&msg);
+			}
+		}
+
+		node->queue = NULL;
+	}
+
+	_unlock();
+
+	if (queue) {
+		skynet_mq_release(queue);
+	}
+}
+
+static void
+_remote_harbor_update(int harbor_id, const char * addr) {
+	struct remote * r = &Z->remote[harbor_id-1];
+	void *socket = zmq_socket( Z->zmq_context, ZMQ_PUSH);
+	int rc = zmq_connect(socket, addr);
+	if (rc<0) {
+		skynet_error(NULL, "Can't connect to %d %s",harbor_id,addr);
+		zmq_close(socket);
+		socket = NULL;
+	}
+	if (socket) {
+		for (;;) {
+			void *old_socket = r->socket;
+			if (__sync_bool_compare_and_swap(&r->socket, old_socket, socket)) {
+				if (old_socket) {
+					zmq_close(old_socket);
+				}
+				break;
+			}
+		}
+	}
+}
+
+static void
+_report_zmq_error(int rc) {
+	if (rc) {
+		fprintf(stderr, "zmq error : %s\n",zmq_strerror(errno));
+		exit(1);
+	}
+}
+
+static void
+_name_update() {
+	zmq_msg_t content;
+	zmq_msg_init(&content);
+	int rc = zmq_recv(Z->zmq_local,&content,0);
+	_report_zmq_error(rc);
+	int sz = zmq_msg_size(&content);
+	int i;
+	int n = 0;
+	uint8_t * buffer = zmq_msg_data(&content);
+	for (i=0;i<sz;i++) {
+		if (buffer[i] == '=') {
+			buffer[i] = '\0';
+			break;
+		}
+		n = n * 10 + (buffer[i] - '0');
+	}
+	if (i==sz) {
+		char tmp[sz+1];
+		memcpy(tmp,buffer,sz);
+		tmp[sz] = '\0';
+		skynet_error(NULL, "Invalid master update [%s]",tmp);
+		zmq_msg_close(&content);
+		return;
+	}
+
+	char tmp[sz-i];
+	memcpy(tmp,buffer+i+1,sz-i-1);
+	tmp[sz-i-1]='\0';
+
+	if (n>0 && n <= REMOTE_MAX) {
+		_remote_harbor_update(n, tmp);
+	} else {
+		uint32_t source = strtoul(tmp,NULL,16);
+		if (source == 0) {
+			skynet_error(NULL, "Invalid master update [%s=%s]",(const char *)buffer,tmp);
+		} else {
+			_register_name((const char *)buffer, source);
+		}
+	}
+
+	zmq_msg_close(&content);
+}
+
+static void
+remote_query_harbor(int harbor_id) {
+	char tmp[32];
+	int sz = sprintf(tmp,"%d",harbor_id);
+	zmq_msg_t request;
+	zmq_msg_init_size(&request,sz);
+	memcpy(zmq_msg_data(&request),tmp,sz);
+	zmq_send(Z->zmq_master_request, &request, 0);
+	zmq_msg_close(&request);
+	zmq_msg_t reply;
+	zmq_msg_init(&reply);
+	int rc = zmq_recv(Z->zmq_master_request, &reply, 0);
+	_report_zmq_error(rc);
+	sz = zmq_msg_size(&reply);
+	char tmp2[sz+1];
+	memcpy(tmp2,zmq_msg_data(&reply),sz);
+	tmp2[sz] = '\0';
+	_remote_harbor_update(harbor_id, tmp2);
+	zmq_msg_close(&reply);
+}
+
+// remote message has two part
+// when part one is nil (size == 0), part two is name update
+// Or part one is source:destination (8 bytes little endian), part two is a binary block for message
+static void
+_remote_recv() {
+	zmq_msg_t header;
+	zmq_msg_init(&header);
+	int rc = zmq_recv(Z->zmq_local,&header,0);
+	_report_zmq_error(rc);
+	size_t s = zmq_msg_size(&header);
+	if (s!=8) {
+		// s should be 0
+		if (s>0) {
+			char tmp[s+1];
+			memcpy(tmp, zmq_msg_data(&header),s);
+			tmp[s] = '\0';
+			skynet_error(NULL,"Invalid master header [%s]",tmp);
+		}
+		_name_update();
+		return;
+	}
+	uint8_t * buffer = zmq_msg_data(&header);
+	struct remote_header rh;
+	buffer_to_remote_header(buffer, &rh);
+	zmq_close(&header);
+
+	zmq_msg_t * data = malloc(sizeof(zmq_msg_t));
+	zmq_msg_init(data);
+	rc = zmq_recv(Z->zmq_local,data,0);
+	_report_zmq_error(rc);
+	struct skynet_message msg;
+	msg.source = rh.source;
+	msg.destination = rh.destination;
+	msg.data = data;
+	msg.sz = zmq_msg_size(data);
+
+	// push remote message to local message queue
+	skynet_mq_push(&msg);
+}
+
+void * 
+skynet_harbor_message_open(struct skynet_message * message) {
+	return zmq_msg_data(message->data);
+}
+
+void 
+skynet_harbor_message_close(struct skynet_message * message) {
+	zmq_msg_close(message->data);
+}
+
+int 
+skynet_harbor_message_isremote(uint32_t handle) {
+	int harbor_id = handle >> HANDLE_REMOTE_SHIFT;
+	return !(harbor_id == 0 || harbor_id == Z->harbor);
+}
+
+static void
+_remote_register_name(const char *name, uint32_t source) {
+	char tmp[strlen(name) + 20];
+	int sz = sprintf(tmp,"%s=%X",name,source);
+	zmq_msg_t msg;
+	zmq_msg_init_size(&msg,sz);
+	memcpy(zmq_msg_data(&msg), tmp , sz);
+	zmq_send(Z->zmq_master_request, &msg,0);
+	zmq_msg_close(&msg);
+	zmq_msg_init(&msg);
+	int rc = zmq_recv(Z->zmq_master_request, &msg,0);
+	_report_zmq_error(rc);
+	zmq_msg_close(&msg);
+}
+
+static void
+_remote_query_name(const char *name) {
+	int sz = strlen(name);
+	zmq_msg_t msg;
+	zmq_msg_init_size(&msg,sz);
+	memcpy(zmq_msg_data(&msg), name , sz);
+	zmq_send(Z->zmq_master_request, &msg,0);
+	zmq_msg_close(&msg);
+	zmq_msg_init(&msg);
+	int rc = zmq_recv(Z->zmq_master_request, &msg,0);
+	_report_zmq_error(rc);
+	sz = zmq_msg_size(&msg);
+	char tmp[sz+1];
+	memcpy(tmp, zmq_msg_data(&msg),sz);
+	tmp[sz] = '\0';
+
+	uint32_t addr = strtoul(tmp,NULL,16);
+	_register_name(name,addr);
+
+	zmq_msg_close(&msg);
+}
+
+static void 
+free_message (void *data, void *hint) {
+	free(data);
+}
+static void
+remote_socket_send(void * socket, struct skynet_message *msg) {
+	struct remote_header rh;
+	rh.source = msg->source;
+	rh.destination = msg->destination;
+	zmq_msg_t part;
+	zmq_msg_init_size(&part,8);
+	uint8_t * buffer = zmq_msg_data(&part);
+	remote_header_to_buffer(&rh,buffer);
+	zmq_send(socket, &part, ZMQ_SNDMORE);
+	zmq_msg_close(&part);
+
+	zmq_msg_init_data(&part,msg->data,msg->sz,free_message,NULL);
+	zmq_send(socket, &part, 0);
+	zmq_msg_close(&part);
+}
+
+static void
+_remote_send() {
+	struct skynet_message msg;
+	while (skynet_mq_leave(Z->queue,&msg)) {
+_goback:
+		if (msg.destination == SKYNET_SYSTEM_NAME) {
+			// register name
+			const char * name = msg.data;
+			if (msg.source) {
+				_remote_register_name(name, msg.source);
+			} else {
+				_remote_query_name(name);
+			}
+
+			free(msg.data);
+		} else {
+			int harbor_id = (msg.destination >> HANDLE_REMOTE_SHIFT);
+			assert(harbor_id > 0);
+			struct remote * r = &Z->remote[harbor_id-1];
+			if (r->socket == NULL) {
+				if (r->queue == NULL) {
+					r->queue = skynet_mq_create(DEFAULT_QUEUE_SIZE);
+					skynet_mq_enter(r->queue, &msg);
+					remote_query_harbor(harbor_id);
+				} else {
+					skynet_mq_enter(r->queue, &msg);
+				}
+			} else {
+				remote_socket_send(r->socket, &msg);
+			}
+		}
+	}
+	__sync_lock_release(&Z->notice_event);
+	// double check
+	if (skynet_mq_leave(Z->queue,&msg)) {
+		goto _goback;
+	}
+}
+
+void *
+skynet_harbor_dispatch_thread(void *ud) {
+	zmq_pollitem_t items[2];
+
+	items[0].socket = Z->zmq_queue_notice;
+	items[0].events = ZMQ_POLLIN;
+	items[1].socket = Z->zmq_local;
+	items[1].events = ZMQ_POLLIN;
+
+	for (;;) {
+		zmq_poll(items,2,-1);
+		if (items[0].revents) {
+			zmq_msg_t msg;
+			zmq_msg_init(&msg);
+			int rc = zmq_recv(Z->zmq_queue_notice,&msg,0);
+			_report_zmq_error(rc);
+			zmq_msg_close(&msg);
+			_remote_send();
+		}
+		if (items[1].revents) {
+			_remote_recv();
+		}
+	}
+}
+
+static void
+register_harbor(void *request, const char *local, int harbor) {
+	char tmp[1024];
+	sprintf(tmp,"%d",harbor);
+	size_t sz = strlen(tmp);
+
+	zmq_msg_t req;
+	zmq_msg_init_size (&req , sz);
+	memcpy(zmq_msg_data(&req),tmp,sz);
+	zmq_send (request, &req, 0);
+	zmq_msg_close (&req);
+
+	zmq_msg_t reply;
+	zmq_msg_init (&reply);
+	int rc = zmq_recv(request, &reply, 0);
+	_report_zmq_error(rc);
+
+	sz = zmq_msg_size (&reply);
+	if (sz > 0) {
+		memcpy(tmp,zmq_msg_data(&reply),sz);
+		tmp[sz] = '\0';
+		if (strcmp(tmp,local) != 0) {
+			fprintf(stderr, "Harbor %d is already registered by %s [%s]\n", harbor, tmp, local);
+			exit(1);
+		}
+	}
+	zmq_msg_close (&reply);
+
+	sprintf(tmp,"%d=%s",harbor,local);
+	sz = strlen(tmp);
+	zmq_msg_init_size (&req , sz);
+	memcpy(zmq_msg_data(&req),tmp,sz);
+	zmq_send (request, &req, 0);
+	zmq_msg_close (&req);
+
+	zmq_msg_init (&reply);
+	rc = zmq_recv (request, &reply, 0);
+	_report_zmq_error(rc);
+
+	sz = zmq_msg_size (&reply);
+	if (sz > 0) {
+		char * buffer = zmq_msg_data(&reply);
+		memcpy(tmp,buffer,sz);
+		tmp[sz] = '\0';
+
+		if (strcmp(local,tmp) !=0) {
+			fprintf(stderr, "Harbor %d is already registered by %s (%s)\n", harbor,tmp,local);
+			exit(1);
+		}
+	}
+	zmq_msg_close (&reply);
+}
+
+void 
+skynet_harbor_init(const char * master, const char *local, int harbor) {
+	if (harbor <=0 || harbor>255 || strlen(local) > 512) {
+		fprintf(stderr,"Invalid harbor id\n");
+		exit(1);
+	}
+	void *context = zmq_init (1);
+	void *request = zmq_socket (context, ZMQ_REQ);
+	int r = zmq_connect(request, master);
+	if (r<0) {
+		fprintf(stderr, "Can't connect to master: %s\n",master);
+		exit(1);
+	}
+	void *harbor_socket = zmq_socket(context, ZMQ_PULL);
+	r = zmq_bind(harbor_socket, local);
+	if (r<0) {
+		fprintf(stderr, "Can't bind to local : %s\n",local);
+		exit(1);
+	}
+	register_harbor(request,local, harbor);
+	printf("Start harbor on : %s\n",local);
+
+	struct harbor * h = malloc(sizeof(*h));
+	memset(h, 0, sizeof(*h));
+	h->zmq_context = context;
+	h->zmq_master_request = request;
+	h->zmq_local = harbor_socket;
+	h->map = _hash_new();
+	h->harbor = harbor;
+	h->queue = skynet_mq_create(DEFAULT_QUEUE_SIZE);
+	h->zmq_queue_notice = zmq_socket(context, ZMQ_PULL);
+	r = zmq_bind(h->zmq_queue_notice, "inproc://notice");
+	assert(r==0);
+
+	Z = h;
+}

skynet_harbor.h

@@ -0,0 +1,21 @@
+#ifndef SKYNET_HARBOR_H
+#define SKYNET_HARBOR_H
+
+#include <stdint.h>
+
+struct skynet_message;
+
+void skynet_harbor_send(const char *name, struct skynet_message * message);
+void skynet_harbor_register(const char *name, uint32_t handle);
+
+// remote message is diffrent from local message.
+// We must use these api to open and close message , see skynet_server.c
+int skynet_harbor_message_isremote(uint32_t handle);
+void * skynet_harbor_message_open(struct skynet_message * message);
+void skynet_harbor_message_close(struct skynet_message * message);
+
+// harbor worker thread
+void * skynet_harbor_dispatch_thread(void *ud);
+void skynet_harbor_init(const char * master, const char *local, int harbor);
+
+#endif

skynet_imp.h

@@ -4,8 +4,12 @@
 struct skynet_config {
 	int thread;
 	int mqueue_size;
+	int harbor;
 	const char * logger;
 	const char * module_path;
+	const char * master;
+	const char * local;
+	const char * start;
 };
 
 void skynet_start(struct skynet_config * config);

skynet_logger.c

@@ -45,7 +45,7 @@ logger_init(struct logger * inst, struct skynet_context *ctx, const char * parm)
 	}
 	if (inst->handle) {
 		skynet_callback(ctx, inst, _logger);
-		skynet_command(ctx, "REG", "logger");
+		skynet_command(ctx, "REG", ".logger");
 		return 0;
 	}
 	return 1;

skynet_main.c

@@ -57,6 +57,10 @@ main(int argc, char *argv[]) {
 	config.mqueue_size = optint(L,"mqueue",256);
 	config.module_path = optstring(L,"cpath","./");
 	config.logger = optstring(L,"logger",NULL);
+	config.harbor = optint(L, "harbor", 1);
+	config.master = optstring(L,"master","tcp://127.0.0.1:2012");
+	config.start = optstring(L,"start","main.lua");
+	config.local = optstring(L,"address","tcp://127.0.0.1:2525");
 	lua_close(L);
 
 	skynet_start(&config);

skynet_mq.c

@@ -42,9 +42,9 @@ _unlock_queue(struct message_queue *q) {
 	__sync_lock_release(&q->lock);
 }
 
-int 
+uint32_t
 skynet_mq_leave(struct message_queue *q, struct skynet_message *message) {
-	int ret = -1;
+	uint32_t ret = 0;
 	_lock_queue(q);
 
 	if (q->head != q->tail) {
@@ -86,7 +86,7 @@ skynet_mq_enter(struct message_queue *q, struct skynet_message *message) {
 	_unlock_queue(q);
 }
 
-int 
+uint32_t 
 skynet_mq_pop(struct skynet_message *message) {
 	return skynet_mq_leave(Q,message);
 }

skynet_mq.h

@@ -2,22 +2,23 @@
 #define SKYNET_MESSAGE_QUEUE_H
 
 #include <stdlib.h>
+#include <stdint.h>
 
 struct skynet_message {
-	int source;
-	int destination;
+	uint32_t source;
+	uint32_t destination;
 	void * data;
 	size_t sz;
 };
 
 struct message_queue;
 
-int skynet_mq_pop(struct skynet_message *message);
+uint32_t skynet_mq_pop(struct skynet_message *message);
 void skynet_mq_push(struct skynet_message *message);
 
 struct message_queue * skynet_mq_create(int cap);
 void skynet_mq_release(struct message_queue *q);
-int skynet_mq_leave(struct message_queue *q, struct skynet_message *message);
+uint32_t skynet_mq_leave(struct message_queue *q, struct skynet_message *message);
 void skynet_mq_enter(struct message_queue *q, struct skynet_message *message);
 
 void skynet_mq_init(int cap);

skynet_server.c

@@ -2,8 +2,8 @@
 #include "skynet_module.h"
 #include "skynet_handle.h"
 #include "skynet_mq.h"
-#include "skynet_blackhole.h"
 #include "skynet_timer.h"
+#include "skynet_harbor.h"
 #include "skynet.h"
 
 #include <string.h>
@@ -17,7 +17,7 @@
 struct skynet_context {
 	void * instance;
 	struct skynet_module * mod;
-	int handle;
+	uint32_t handle;
 	int calling;
 	int ref;
 	char handle_name[10];
@@ -95,35 +95,21 @@ skynet_context_release(struct skynet_context *ctx) {
 	return ctx;
 }
 
-static void
-_drop_message(int source, const char * addr , void * data, size_t sz) {
-	struct blackhole * b = malloc(sizeof(*b));
-	b->source = source;
-	b->destination = strdup(addr);
-	b->data = data;
-	b->sz = sz;
-
-	int des = skynet_handle_findname(BLACKHOLE);
-	if (des<0)
-		return;
-
-	struct skynet_message msg;
-	msg.source = source;
-	msg.destination = des;
-	msg.data = b;
-	msg.sz = sizeof(*b);
-	skynet_mq_push(&msg);
-}
-
 static void
 _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
-	if (msg->source == -1) {
+	if (msg->source == SKYNET_SYSTEM_TIMER) {
 		ctx->cb(ctx, ctx->cb_ud, NULL, msg->data, msg->sz);
 	} else {
 		char tmp[10];
 		tmp[0] = ':';
 		_id_to_hex(tmp+1, msg->source);
-		ctx->cb(ctx, ctx->cb_ud, tmp, msg->data, msg->sz);
+		if (skynet_harbor_message_isremote(msg->source)) {
+			void * data = skynet_harbor_message_open(msg);
+			ctx->cb(ctx, ctx->cb_ud, tmp, data, msg->sz);
+			skynet_harbor_message_close(msg);
+		} else {
+			ctx->cb(ctx, ctx->cb_ud, tmp, msg->data, msg->sz);
+		}
 
 		free(msg->data);
 	}
@@ -132,16 +118,14 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
 int
 skynet_context_message_dispatch(void) {
 	struct skynet_message msg;
-	int handle = skynet_mq_pop(&msg);
-	if (handle < 0) {
+	uint32_t handle = skynet_mq_pop(&msg);
+	if (handle == 0) {
 		return 1;
 	}
 	struct skynet_context * ctx = skynet_handle_grab(handle);
 	if (ctx == NULL) {
-		char tmp[10];
-		tmp[0] = ':';
-		_id_to_hex(tmp+1, msg.destination);
-		_drop_message(msg.source, tmp, msg.data, msg.sz);
+		free(msg.data);
+		skynet_error(NULL, "Drop message from %u to %u , size = %d",msg.source, msg.destination, (int)msg.sz);
 		return 0;
 	}
 	if (__sync_lock_test_and_set(&ctx->calling, 1)) {
@@ -149,13 +133,14 @@ skynet_context_message_dispatch(void) {
 		skynet_mq_enter(ctx->queue, &msg);
 	} else {
 		if (ctx->cb == NULL) {
-			char tmp[10];
-			tmp[0] = ':';
-			_id_to_hex(tmp+1, msg.destination);
-			_drop_message(msg.source, tmp, msg.data, msg.sz);
+			if (skynet_harbor_message_isremote(msg.source)) {
+				skynet_harbor_message_close(&msg);
+			}
+			free(msg.data);
+			skynet_error(NULL, "Drop message from %u to %u without callback , size = %d",msg.source, msg.destination, (int)msg.sz);
 		} else {
 			_dispatch_message(ctx, &msg);
-			while(skynet_mq_leave(ctx->queue,&msg) >=0) {
+			while(skynet_mq_leave(ctx->queue,&msg)) {
 				_dispatch_message(ctx,&msg);
 			}
 		}
@@ -188,8 +173,12 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
 	if (strcmp(cmd,"REG") == 0) {
 		if (parm == NULL || parm[0] == '\0') {
 			return context->handle_name;
+		} else if (parm[0] == '.') {
+			return skynet_handle_namehandle(context->handle, parm + 1);
 		} else {
-			return skynet_handle_namehandle(context->handle, parm);
+			assert(context->handle!=0);
+			skynet_harbor_register(parm, context->handle);
+			return NULL;
 		}
 	}
 
@@ -220,33 +209,47 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
 
 void 
 skynet_send(struct skynet_context * context, const char * addr , void * msg, size_t sz) {
-	int des = -1;
+	uint32_t des = 0;
 	if (addr[0] == ':') {
 		des = strtol(addr+1, NULL, 16);
 	} else if (addr[0] == '.') {
 		des = skynet_handle_findname(addr + 1);
-		if (des < 0) {
-			_drop_message(context->handle, addr, (void *)msg, sz);
+		if (des == 0) {
+			free(msg);
+			skynet_error(context, "Drop message to %s, size = %d", addr, (int)sz);
 			return;
 		}
+	} else {
+		struct skynet_message smsg;
+		smsg.source = context->handle;
+		smsg.destination = 0;
+		smsg.data = msg;
+		smsg.sz = sz;
+		skynet_harbor_send(addr, &smsg);
+		return;
 	}
 
-	assert(des >= 0);
+	assert(des > 0);
+
 	struct skynet_message smsg;
 	smsg.source = context->handle;
 	smsg.destination = des;
 	smsg.data = msg;
 	smsg.sz = sz;
-	skynet_mq_push(&smsg);
+	if (skynet_harbor_message_isremote(des)) {
+		skynet_harbor_send(NULL, &smsg);
+	} else {
+		skynet_mq_push(&smsg);
+	}
 }
 
-int 
+uint32_t 
 skynet_context_handle(struct skynet_context *ctx) {
 	return ctx->handle;
 }
 
 void 
-skynet_context_init(struct skynet_context *ctx, int handle) {
+skynet_context_init(struct skynet_context *ctx, uint32_t handle) {
 	ctx->handle = handle;
 }
 

skynet_server.h

@@ -1,14 +1,16 @@
 #ifndef SKYNET_SERVER_H
 #define SKYNET_SERVER_H
 
+#include <stdint.h>
+
 struct skynet_context;
 struct skynet_message;
 
 struct skynet_context * skynet_context_new(const char * name, const char * parm);
 void skynet_context_grab(struct skynet_context *);
 struct skynet_context * skynet_context_release(struct skynet_context *);
-int skynet_context_handle(struct skynet_context *);
-void skynet_context_init(struct skynet_context *, int handle);
+uint32_t skynet_context_handle(struct skynet_context *);
+void skynet_context_init(struct skynet_context *, uint32_t handle);
 void skynet_context_push(struct skynet_context *, struct skynet_message *message);
 int skynet_context_pop(struct skynet_context *, struct skynet_message *message);
 int skynet_context_message_dispatch(void);	// return 1 when block

skynet_start.c

@@ -4,6 +4,7 @@
 #include "skynet_handle.h"
 #include "skynet_module.h"
 #include "skynet_timer.h"
+#include "skynet_harbor.h"
 
 #include <pthread.h>
 #include <unistd.h>
@@ -30,32 +31,33 @@ _worker(void *p) {
 
 static void
 _start(int thread) {
-	pthread_t pid[thread+1];
+	pthread_t pid[thread+2];
 
 	pthread_create(&pid[0], NULL, _timer, NULL);
+	pthread_create(&pid[1], NULL, skynet_harbor_dispatch_thread, NULL);
+
 	int i;
-	for (i=1;i<thread;i++) {
+	for (i=2;i<thread+2;i++) {
 		pthread_create(&pid[i], NULL, _worker, NULL);
 	}
 
-	for (i=0;i<thread;i++) {
+	for (i=0;i<thread+2;i++) {
 		pthread_join(pid[i], NULL); 
 	}
 }
 
 void 
 skynet_start(struct skynet_config * config) {
+	// harbor must be init first
+	skynet_harbor_init(config->master , config->local, config->harbor);
+	skynet_handle_init(config->harbor);
 	skynet_mq_init(config->mqueue_size);
 	skynet_module_init(config->module_path);
-	skynet_handle_init();
 	skynet_timer_init();
 	struct skynet_context *ctx;
 	ctx = skynet_context_new("logger", config->logger);
 	assert(ctx);
-	ctx = skynet_context_new("blackhole", NULL);
-	assert(ctx);
-	char parm[] = "main.lua";
-	ctx = skynet_context_new("snlua", parm);
+	ctx = skynet_context_new("snlua", config->start);
 
 	_start(config->thread);
 }

skynet_system.h

@@ -0,0 +1,8 @@
+#ifndef SKYNET_MESSAGE_SYSTEM_H
+#define SKYNET_MESSAGE_SYSTEM_H
+
+#define SKYNET_SYSTEM_TIMER 1
+#define SKYNET_SYSTEM_LOGGER 2
+#define SKYNET_SYSTEM_NAME 3
+
+#endif

skynet_timer.c

@@ -160,7 +160,7 @@ timer_create_timer()
 void 
 skynet_timeout(int handle, int time, int session) {
 	struct skynet_message message;
-	message.source = -1;
+	message.source = SKYNET_SYSTEM_TIMER;
 	message.destination = handle;
 	message.data = NULL;
 	message.sz = (size_t) session;

skynet_timer.h

@@ -1,6 +1,7 @@
 #ifndef SKYNET_TIMER_H
 #define SKYNET_TIMER_H
 
+#include "skynet_system.h"
 #include <stdint.h>
 
 void skynet_timeout(int handle, int time, int session);

watchdog.lua

@@ -5,19 +5,19 @@ local command = {}
 function command:open(parm)
 	local fd,addr = string.match(parm,"(%d+) ([^%s]+)")
 	fd = tonumber(fd)
-	print("[watchdog] open",self,fd,addr)
+	skynet.send("LOG",string.format("%d %d %s",self,fd,addr))
 	local agent = skynet.command("LAUNCH","snlua agent.lua ".. self)
 	if agent then
-		skynet.send(".gate","forward ".. self .. " " .. agent)
+		skynet.send("gate","forward ".. self .. " " .. agent)
 	end
 end
 
 function command:close()
-	print("[watchdog] close",self)
+	skynet.send("LOG",string.format("close %d",self))
 end
 
 function command:data(data)
-	print("[watchdog] data",self,#data,data)
+	skynet.send("LOG",string.format("data %d size=%d",self,#data))
 end
 
 skynet.callback(function(from , message)
@@ -27,8 +27,8 @@ skynet.callback(function(from , message)
 	if f then
 		f(id,parm)
 	else
-		skynet.error(string.format("[watchdog] Unknown command : %s %d %s",cmd,id,parm))
+		skynet.error(string.format("[watchdog] Unknown command : %s",message))
 	end
 end)
 
-skynet.command("REG","watchdog")
+skynet.command("REG",".watchdog")