From 58101c5be42c69cea74962a91e97bde66f925c3d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Fri, 3 Aug 2012 18:20:05 +0800 Subject: [PATCH] skynet harbor --- Makefile | 16 +- README.md | 18 ++ config | 4 + config_log | 8 + globallog.lua | 7 + main.lua | 2 +- main_log.lua | 8 + master/Makefile | 7 + master/master.c | 250 +++++++++++++++++++ master/test.c | 124 +++++++++ skynet_blackhole.c | 22 -- skynet_blackhole.h | 13 - skynet_error.c | 7 +- skynet_handle.c | 36 +-- skynet_handle.h | 19 +- skynet_harbor.c | 609 +++++++++++++++++++++++++++++++++++++++++++++ skynet_harbor.h | 21 ++ skynet_imp.h | 4 + skynet_logger.c | 2 +- skynet_main.c | 4 + skynet_mq.c | 6 +- skynet_mq.h | 9 +- skynet_server.c | 89 +++---- skynet_server.h | 6 +- skynet_start.c | 18 +- skynet_system.h | 8 + skynet_timer.c | 2 +- skynet_timer.h | 1 + watchdog.lua | 12 +- 29 files changed, 1197 insertions(+), 135 deletions(-) create mode 100644 README.md create mode 100644 config_log create mode 100644 globallog.lua create mode 100644 main_log.lua create mode 100644 master/Makefile create mode 100644 master/master.c create mode 100644 master/test.c delete mode 100644 skynet_blackhole.c delete mode 100644 skynet_blackhole.h create mode 100644 skynet_harbor.c create mode 100644 skynet_harbor.h create mode 100644 skynet_system.h diff --git a/Makefile b/Makefile index 678d30c8..aa91dcdb 100644 --- a/Makefile +++ b/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 diff --git a/README.md b/README.md new file mode 100644 index 00000000..68472f04 --- /dev/null +++ b/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. +``` \ No newline at end of file diff --git a/config b/config index 76634b75..ecdef693 100644 --- a/config +++ b/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" diff --git a/config_log b/config_log new file mode 100644 index 00000000..0530c219 --- /dev/null +++ b/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" diff --git a/globallog.lua b/globallog.lua new file mode 100644 index 00000000..71f81cca --- /dev/null +++ b/globallog.lua @@ -0,0 +1,7 @@ +local skynet = require "skynet" + +skynet.callback(function(from , message) + print("[GLOBALLOG]",from,message) +end) + +skynet.command("REG","LOG") \ No newline at end of file diff --git a/main.lua b/main.lua index 3982b503..0fa03f15 100644 --- a/main.lua +++ b/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") diff --git a/main_log.lua b/main_log.lua new file mode 100644 index 00000000..e16754d2 --- /dev/null +++ b/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") diff --git a/master/Makefile b/master/Makefile new file mode 100644 index 00000000..fae7e177 --- /dev/null +++ b/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 diff --git a/master/master.c b/master/master.c new file mode 100644 index 00000000..7c0aeaf8 --- /dev/null +++ b/master/master.c @@ -0,0 +1,250 @@ +#include +#include +#include +#include +#include + +#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;islave[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 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; +} diff --git a/master/test.c b/master/test.c new file mode 100644 index 00000000..6197caa5 --- /dev/null +++ b/master/test.c @@ -0,0 +1,124 @@ +#include +#include +#include +#include +#include +#include +#include + +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; +} diff --git a/skynet_blackhole.c b/skynet_blackhole.c deleted file mode 100644 index 72a97020..00000000 --- a/skynet_blackhole.c +++ /dev/null @@ -1,22 +0,0 @@ -#include "skynet.h" -#include "skynet_blackhole.h" - -#include -#include -#include - -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; -} diff --git a/skynet_blackhole.h b/skynet_blackhole.h deleted file mode 100644 index 3e549839..00000000 --- a/skynet_blackhole.h +++ /dev/null @@ -1,13 +0,0 @@ -#ifndef SKYNET_BLACKHOLE_H -#define SKYNET_BLACKHOLE_H - -#include - -struct blackhole { - int source; - char * destination; - void * data; - size_t sz; -}; - -#endif diff --git a/skynet_error.c b/skynet_error.c index 99f5391d..da1c417b 100644 --- a/skynet_error.c +++ b/skynet_error.c @@ -2,6 +2,7 @@ #include "skynet_handle.h" #include "skynet_mq.h" #include "skynet_server.h" +#include "skynet_system.h" #include #include @@ -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; diff --git a/skynet_handle.c b/skynet_handle.c index 9f26bfee..ad1aa553 100644 --- a/skynet_handle.c +++ b/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)); diff --git a/skynet_handle.h b/skynet_handle.h index 6e38cd34..28bd548c 100644 --- a/skynet_handle.h +++ b/skynet_handle.h @@ -1,15 +1,22 @@ #ifndef SKYNET_CONTEXT_HANDLE_H #define SKYNET_CONTEXT_HANDLE_H +#include + +// 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 diff --git a/skynet_harbor.c b/skynet_harbor.c new file mode 100644 index 00000000..d305c348 --- /dev/null +++ b/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 +#include +#include +#include +#include + +#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;i0 && 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; +} diff --git a/skynet_harbor.h b/skynet_harbor.h new file mode 100644 index 00000000..b09015d4 --- /dev/null +++ b/skynet_harbor.h @@ -0,0 +1,21 @@ +#ifndef SKYNET_HARBOR_H +#define SKYNET_HARBOR_H + +#include + +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 diff --git a/skynet_imp.h b/skynet_imp.h index 0e55bc8d..5a20e545 100644 --- a/skynet_imp.h +++ b/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); diff --git a/skynet_logger.c b/skynet_logger.c index 8cad1ea9..6919054b 100644 --- a/skynet_logger.c +++ b/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; diff --git a/skynet_main.c b/skynet_main.c index c2f19e14..e75e0876 100644 --- a/skynet_main.c +++ b/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); diff --git a/skynet_mq.c b/skynet_mq.c index a7041a8c..cb07380b 100644 --- a/skynet_mq.c +++ b/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); } diff --git a/skynet_mq.h b/skynet_mq.h index b77a12fa..f073e9d5 100644 --- a/skynet_mq.h +++ b/skynet_mq.h @@ -2,22 +2,23 @@ #define SKYNET_MESSAGE_QUEUE_H #include +#include 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); diff --git a/skynet_server.c b/skynet_server.c index 5e008080..29d6b94b 100644 --- a/skynet_server.c +++ b/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 @@ -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; } diff --git a/skynet_server.h b/skynet_server.h index 87b9b76a..afde351c 100644 --- a/skynet_server.h +++ b/skynet_server.h @@ -1,14 +1,16 @@ #ifndef SKYNET_SERVER_H #define SKYNET_SERVER_H +#include + 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 diff --git a/skynet_start.c b/skynet_start.c index b1fa9cab..50b9c714 100644 --- a/skynet_start.c +++ b/skynet_start.c @@ -4,6 +4,7 @@ #include "skynet_handle.h" #include "skynet_module.h" #include "skynet_timer.h" +#include "skynet_harbor.h" #include #include @@ -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;imaster , 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); } diff --git a/skynet_system.h b/skynet_system.h new file mode 100644 index 00000000..9b671977 --- /dev/null +++ b/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 diff --git a/skynet_timer.c b/skynet_timer.c index d62c8b52..b5387e01 100644 --- a/skynet_timer.c +++ b/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; diff --git a/skynet_timer.h b/skynet_timer.h index 65ce817b..f3439fa1 100644 --- a/skynet_timer.h +++ b/skynet_timer.h @@ -1,6 +1,7 @@ #ifndef SKYNET_TIMER_H #define SKYNET_TIMER_H +#include "skynet_system.h" #include void skynet_timeout(int handle, int time, int session); diff --git a/watchdog.lua b/watchdog.lua index 8fa6874d..8d640f6f 100644 --- a/watchdog.lua +++ b/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") \ No newline at end of file +skynet.command("REG",".watchdog") \ No newline at end of file