diff --git a/HISTORY.md b/HISTORY.md index e4b2b7f7..a7ce9057 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -3,6 +3,10 @@ Dev version * Optimize redis driver `compose_message`. * Add module skynet.harbor for monitor harbor connect/disconnect, see test/testharborlink.lua . * cluster.open support cluster name. +* Add new api skynet.packstring , and skynet.unpack support lua string +* socket.listen support put port into address. (address:port) +* Redesign harbor/master/dummy, remove lots of C code and rewite in lua. +* Remove block connect api, queue sending message during connecting now. v0.3.1 (2014-6-16) ----------- diff --git a/Makefile b/Makefile index 3c5fd446..45631905 100644 --- a/Makefile +++ b/Makefile @@ -40,7 +40,7 @@ jemalloc : $(MALLOC_STATICLIB) # skynet -CSERVICE = snlua logger gate master harbor dummy +CSERVICE = snlua logger gate harbor LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack \ cjson clientsocket memory profile multicast \ cluster diff --git a/examples/globallog.lua b/examples/globallog.lua index 9b4dce4c..ffd0a0f3 100644 --- a/examples/globallog.lua +++ b/examples/globallog.lua @@ -1,24 +1,8 @@ local skynet = require "skynet" -skynet.register_protocol { - name = "text", - id = skynet.PTYPE_TEXT, - pack = function (...) - local n = select ("#" , ...) - if n == 0 then - return "" - elseif n == 1 then - return tostring(...) - else - return table.concat({...}," ") - end - end, - unpack = skynet.tostring -} - skynet.start(function() - skynet.dispatch("text", function(session, address, text) - print("[GLOBALLOG]", skynet.address(address),text) + skynet.dispatch("lua", function(session, address, ...) + print("[GLOBALLOG]", skynet.address(address), ...) end) skynet.register "LOG" end) diff --git a/lualib-src/lua-seri.c b/lualib-src/lua-seri.c index 69e1c2af..c6544462 100644 --- a/lualib-src/lua-seri.c +++ b/lualib-src/lua-seri.c @@ -581,8 +581,16 @@ _luaseri_unpack(lua_State *L) { if (lua_isnoneornil(L,1)) { return 0; } - void * buffer = lua_touserdata(L,1); - int len = luaL_checkinteger(L,2); + void * buffer; + int len; + if (lua_type(L,1) == LUA_TSTRING) { + size_t sz; + buffer = (void *)lua_tolstring(L,1,&sz); + len = (int)sz; + } else { + buffer = lua_touserdata(L,1); + len = luaL_checkinteger(L,2); + } if (len == 0) { return 0; } diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c index db49e997..931de581 100644 --- a/lualib-src/lua-skynet.c +++ b/lualib-src/lua-skynet.c @@ -289,6 +289,16 @@ _harbor(lua_State *L) { return 2; } +static int +lpackstring(lua_State *L) { + _luaseri_pack(L); + char * str = (char *)lua_touserdata(L, -2); + int sz = lua_tointeger(L, -1); + lua_pushlstring(L, str, sz); + skynet_free(str); + return 1; +} + int luaopen_skynet_c(lua_State *L) { luaL_checkversion(L); @@ -303,6 +313,7 @@ luaopen_skynet_c(lua_State *L) { { "harbor", _harbor }, { "pack", _luaseri_pack }, { "unpack", _luaseri_unpack }, + { "packstring", lpackstring }, { "callback", _callback }, { NULL, NULL }, }; diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 93b67d65..385a6d38 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -288,6 +288,7 @@ skynet.redirect = function(dest,source,typename,...) end skynet.pack = assert(c.pack) +skynet.packstring = assert(c.packstring) skynet.unpack = assert(c.unpack) skynet.tostring = assert(c.tostring) @@ -339,8 +340,8 @@ function skynet.dispatch(typename, func) p.dispatch = func end -local function unknown_request(session, address, msg, sz) - print("Unknown request :" , c.tostring(msg,sz)) +local function unknown_request(session, address, msg, sz, prototype) + skynet.error(string.format("Unknown request (%s): %s", prototype, c.tostring(msg,sz))) error(string.format("Unknown session : %d from %x", session, address)) end @@ -394,7 +395,7 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...) session_coroutine_address[co] = source suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz, ...))) else - unknown_request(session, source, msg, sz) + unknown_request(session, source, msg, sz, proto[prototype]) end end end diff --git a/lualib/skynet/harbor.lua b/lualib/skynet/harbor.lua index 06f47bb6..5f470be8 100644 --- a/lualib/skynet/harbor.lua +++ b/lualib/skynet/harbor.lua @@ -2,39 +2,17 @@ local skynet = require "skynet" local harbor = {} -local HARBOR = skynet.getenv "harbor_address" - -if HARBOR then - HARBOR = tonumber("0x" .. string.sub(HARBOR , 2)) -end - function harbor.globalname(name, handle) - assert(HARBOR) handle = handle or skynet.self() - skynet.redirect(HARBOR, handle, "system", 0, "R " .. name) -end - -function harbor.init(h) - assert(HARBOR == nil) - HARBOR = h - skynet.setenv("harbor_address", skynet.address(h)) + skynet.send(".slave", "lua", "REGISTER", name, handle) end function harbor.link(id) - assert(HARBOR) - skynet.call(HARBOR, "system", "M " .. tostring(id)) + skynet.call(".slave", "lua", "LINK", id) end function harbor.connect(id) - assert(HARBOR) - skynet.call(HARBOR, "system", "C " .. tostring(id)) + skynet.call(".slave", "lua", "CONNECT", id) end -skynet.register_protocol { - name = "system", - id = skynet.PTYPE_SYSTEM, - pack = function(...) return ... end, - unpack = skynet.tostring, -} - return harbor diff --git a/lualib/socket.lua b/lualib/socket.lua index 93b9bf4c..240d6ed7 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -271,7 +271,13 @@ function socket.invalid(id) return socket_pool[id] == nil end -socket.listen = assert(driver.listen) +function socket.listen(host, port, backlog) + if port == nil then + host, port = string.match(host, "([^:]+):(.+)$") + port = tonumber(port) + end + return driver.listen(host, port, backlog) +end function socket.lock(id) local s = socket_pool[id] diff --git a/service-src/service_dummy.c b/service-src/service_dummy.c deleted file mode 100644 index a1f19d4c..00000000 --- a/service-src/service_dummy.c +++ /dev/null @@ -1,319 +0,0 @@ -#include "skynet.h" -#include "skynet_harbor.h" - -#include -#include - -#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;isize-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;inode[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 void -dummy_command(struct dummy * h, const char * msg, size_t sz, int session, uint32_t source) { - switch(msg[0]) { - case 'R' : { - // register global name - const char * name = msg + 2; - int s = (int)sz; - s -= 2; - if (s <=0 || s>= GLOBALNAME_LENGTH) { - skynet_error(h->ctx, "Invalid global name %s", name); - return; - } - struct remote_name rn; - memset(&rn, 0, sizeof(rn)); - memcpy(rn.name, name, s); - rn.handle = source; - _update_name(h, rn.name, rn.handle); - break; - } - case 'C' : - case 'M' : - skynet_error(h->ctx, "Don't support harbor monitor in cluster dummy mode"); - skynet_send(h->ctx, 0, source, PTYPE_ERROR, session, NULL, 0); - break; - default: - skynet_error(h->ctx, "Unknown command %s", msg); - return; - } -} - -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: { - dummy_command(h, msg, sz, session, source); - 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; -} diff --git a/service-src/service_harbor.c b/service-src/service_harbor.c index a9cfaec4..8dbf7db4 100644 --- a/service-src/service_harbor.c +++ b/service-src/service_harbor.c @@ -2,6 +2,16 @@ #include "skynet_harbor.h" #include "skynet_socket.h" +/* + harbor listen the PTYPE_HARBOR (in text) + N name : update the global name + S fd id: connect to new harbor , we should send self_id to fd first , and then recv a id (check it), and at last send queue. + A fd id: accept new harbor , we should send self_id to fd , and then send queue. + + If the fd is disconnected, send message to slave in PTYPE_TEXT. D id + If we don't known a globalname, send message to slave in PTYPE_TEXT. Q name + */ + #include #include #include @@ -13,8 +23,22 @@ #define HASH_SIZE 4096 #define DEFAULT_QUEUE_SIZE 1024 +// 12 is sizeof(struct remote_message_header) +#define HEADER_COOKIE_LENGTH 12 + +/* + 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; +}; + struct msg { - uint8_t * buffer; + struct remote_message_header header; + void * buffer; size_t size; }; @@ -37,99 +61,34 @@ 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; -}; +#define STATUS_WAIT 0 +#define STATUS_HANDSHAKE 1 +#define STATUS_HEADER 2 +#define STATUS_CONTENT 3 +#define STATUS_DOWN 4 -struct monitor_response { - uint32_t addr; - int session; +struct slave { + int fd; + struct msg_queue *queue; + int status; + int length; + int read; + uint8_t size[4]; + char * recv_buffer; }; -struct monitor_set { - int cap; - int n; - struct monitor_response * resp; -}; - -// 12 is sizeof(struct remote_message_header) -#define HEADER_COOKIE_LENGTH 12 - struct harbor { struct skynet_context *ctx; - char * local_addr; int id; + uint32_t slave; struct hashmap * map; - int master_fd; - char * master_addr; - int remote_fd[REMOTE_MAX]; - bool connected[REMOTE_MAX]; - char * remote_addr[REMOTE_MAX]; - struct monitor_set * monitor[REMOTE_MAX]; + struct slave s[REMOTE_MAX]; }; -static void -monitor_free(struct harbor *h) { - int i; - for (i=0;imonitor[i]; - if (m) { - skynet_free(m->resp); - skynet_free(m); - h->monitor[i] = NULL; - } - } -} - -static void -monitor_add(struct harbor *h, int id, uint32_t addr, int session) { - struct monitor_set * m = h->monitor[id]; - if (m == NULL) { - m = skynet_malloc(sizeof(*m)); - m->cap = 4; - m->n = 0; - m->resp = skynet_malloc(m->cap * sizeof(struct monitor_response)); - h->monitor[id] = m; - } - if (m->n >= m->cap) { - assert(m->n == m->cap); - struct monitor_response * resp = skynet_malloc(m->cap * 2 * sizeof(struct monitor_response)); - int i; - for (i=0;in;i++) { - resp[i] = m->resp[i]; - } - m->cap *= 2; - skynet_free(m->resp); - m->resp = resp; - } - struct monitor_response * resp = &m->resp[m->n++]; - resp->addr = addr; - resp->session = session; -} - -static void -monitor_clear(struct harbor *h, int id) { - struct monitor_set * m = h->monitor[id]; - if (m) { - int i; - for (i=0;in;i++) { - struct monitor_response * resp = &m->resp[i]; - skynet_send(h->ctx, 0, resp->addr, PTYPE_RESPONSE, resp->session, NULL, 0); - } - m->n = 0; - } -} - // hash table static void -_push_queue(struct msg_queue * queue, const void * buffer, size_t sz, struct remote_message_header * header) { +push_queue_msg(struct msg_queue * queue, struct msg * m) { // 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) { @@ -145,16 +104,21 @@ _push_queue(struct msg_queue * queue, const void * buffer, size_t sz, struct rem queue->size *= 2; } struct msg * slot = &queue->data[queue->tail]; + *slot = *m; 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 void +push_queue(struct msg_queue * queue, void * buffer, size_t sz, struct remote_message_header * header) { + struct msg m; + m.header = *header; + m.buffer = buffer; + m.size = sz; + push_queue_msg(queue, &m); } static struct msg * -_pop_queue(struct msg_queue * queue) { +pop_queue(struct msg_queue * queue) { if (queue->head == queue->tail) { return NULL; } @@ -164,7 +128,7 @@ _pop_queue(struct msg_queue * queue) { } static struct msg_queue * -_new_queue() { +new_queue() { struct msg_queue * queue = skynet_malloc(sizeof(*queue)); queue->size = DEFAULT_QUEUE_SIZE; queue->head = 0; @@ -175,20 +139,19 @@ _new_queue() { } static void -_release_queue(struct msg_queue *queue) { +release_queue(struct msg_queue *queue) { if (queue == NULL) return; - struct msg * m = _pop_queue(queue); - while (m) { + struct msg * m; + while ((m=pop_queue(queue)) != NULL) { 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]) { +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]; @@ -206,7 +169,7 @@ _hash_search(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) { // Don't support erase name yet static struct void -_hash_erase(struct hashmap * hash, char name[GLOBALNAME_LENGTH) { +hash_erase(struct hashmap * hash, char name[GLOBALNAME_LENGTH) { uint32_t *ptr = name; uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3]; struct keyvalue ** ptr = &hash->node[h % HASH_SIZE]; @@ -224,7 +187,7 @@ _hash_erase(struct hashmap * hash, char name[GLOBALNAME_LENGTH) { */ static struct keyvalue * -_hash_insert(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) { +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]; @@ -240,20 +203,20 @@ _hash_insert(struct hashmap * hash, const char name[GLOBALNAME_LENGTH]) { } static struct hashmap * -_hash_new() { +hash_new() { struct hashmap * h = skynet_malloc(sizeof(struct hashmap)); memset(h,0,sizeof(*h)); return h; } static void -_hash_delete(struct hashmap *hash) { +hash_delete(struct hashmap *hash) { int i; for (i=0;inode[i]; while (node) { struct keyvalue * next = node->next; - _release_queue(node->queue); + release_queue(node->queue); skynet_free(node); node = next; } @@ -263,62 +226,49 @@ _hash_delete(struct hashmap *hash) { /////////////// +static void +close_harbor(struct harbor *h, int id) { + struct slave *s = &h->s[id]; + s->status = STATUS_DOWN; + if (s->fd) { + skynet_socket_close(h->ctx, s->fd); + } + if (s->queue) { + release_queue(s->queue); + s->queue = NULL; + } +} + +static void +report_harbor_down(struct harbor *h, int id) { + char down[64]; + int n = sprintf(down, "D %d",id); + + skynet_send(h->ctx, 0, h->slave, PTYPE_TEXT, 0, down, n); +} + struct harbor * harbor_create(void) { struct harbor * h = skynet_malloc(sizeof(*h)); - h->ctx = NULL; - h->id = 0; - h->master_fd = -1; - h->master_addr = NULL; - int i; - for (i=0;iremote_fd[i] = -1; - h->connected[i] = false; - h->remote_addr[i] = NULL; - h->monitor[i] = NULL; - } - h->map = _hash_new(); + memset(h,0,sizeof(*h)); + h->map = hash_new(); return h; } void harbor_release(struct harbor *h) { - struct skynet_context *ctx = h->ctx; - if (h->master_fd >= 0) { - skynet_socket_close(ctx, h->master_fd); - } - skynet_free(h->master_addr); - skynet_free(h->local_addr); int i; - for (i=0;iremote_fd[i] >= 0) { - skynet_socket_close(ctx, h->remote_fd[i]); - skynet_free(h->remote_addr[i]); + for (i=1;is[i]; + if (s->fd && s->status != STATUS_DOWN) { + close_harbor(h,i); + report_harbor_down(h,i); } } - _hash_delete(h->map); - monitor_free(h); + hash_delete(h->map); skynet_free(h); } -static int -_connect_to(struct harbor *h, const char *ipaddress) { - char * port = strchr(ipaddress,':'); - if (port==NULL) { - return -1; - } - int sz = port - ipaddress; - char tmp[sz + 1]; - memcpy(tmp,ipaddress,sz); - tmp[sz] = '\0'; - - int portid = strtol(port+1, NULL,10); - - skynet_error(h->ctx, "Harbor(%d) connect to %s:%d", h->id, tmp, portid); - - return skynet_socket_connect(h->ctx, tmp, portid); -} - static inline void to_bigendian(uint8_t *buffer, uint32_t n) { buffer[0] = (n >> 24) & 0xff; @@ -328,7 +278,7 @@ to_bigendian(uint8_t *buffer, uint32_t n) { } static inline void -_header_to_message(const struct remote_message_header * header, uint8_t * message) { +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); @@ -345,122 +295,200 @@ from_bigendian(uint32_t n) { } static inline void -_message_to_header(const uint32_t *message, struct remote_message_header *header) { +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 -_send_package(struct skynet_context *ctx, int fd, const void * buffer, size_t sz) { - uint8_t * sendbuf = skynet_malloc(sz+4); - to_bigendian(sendbuf, sz); - memcpy(sendbuf+4, buffer, sz); +// socket package - if (skynet_socket_send(ctx, fd, sendbuf, sz+4)) { - skynet_error(ctx, "Send to %d error", fd); - } +static void +forward_local_messsage(struct harbor *h, void *msg, int sz) { + const char * cookie = msg; + cookie += sz - HEADER_COOKIE_LENGTH; + struct remote_message_header header; + message_to_header((const uint32_t *)cookie, &header); + + uint32_t destination = header.destination; + int type = (destination >> HANDLE_REMOTE_SHIFT) | PTYPE_TAG_DONTCOPY; + destination = (destination & HANDLE_MASK) | ((uint32_t)h->id << HANDLE_REMOTE_SHIFT); + + skynet_send(h->ctx, header.source, destination, type, (int)header.session, (void *)msg, sz-HEADER_COOKIE_LENGTH); } static void -_send_remote(struct skynet_context * ctx, int fd, const char * buffer, size_t sz, struct remote_message_header * cookie) { +send_remote(struct skynet_context * ctx, int fd, const char * buffer, size_t sz, struct remote_message_header * cookie) { uint32_t sz_header = sz+sizeof(*cookie); uint8_t * sendbuf = skynet_malloc(sz_header+4); to_bigendian(sendbuf, sz_header); memcpy(sendbuf+4, buffer, sz); - _header_to_message(cookie, sendbuf+4+sz); + header_to_message(cookie, sendbuf+4+sz); - if (skynet_socket_send(ctx, fd, sendbuf, sz_header+4)) { - skynet_error(ctx, "Remote send to %d error", fd); - } + // ignore send error, because if the connection is broken, the mainloop will recv a message. + skynet_socket_send(ctx, fd, sendbuf, sz_header+4); } static void -response_close(struct harbor *h, int id) { - if (h->connected[id]) { - monitor_clear(h, id); - } - h->connected[id] = false; -} - -static void -_update_remote_address(struct harbor *h, int harbor_id, const char * ipaddr) { - if (harbor_id == h->id) { - return; - } - assert(harbor_id > 0 && harbor_id< REMOTE_MAX); - struct skynet_context * context = h->ctx; - if (h->remote_fd[harbor_id] >=0) { - skynet_socket_close(context, h->remote_fd[harbor_id]); - skynet_free(h->remote_addr[harbor_id]); - h->remote_addr[harbor_id] = NULL; - } - h->remote_fd[harbor_id] = _connect_to(h, ipaddr); - response_close(h, harbor_id); -} - -static void -_dispatch_queue(struct harbor *h, struct msg_queue * queue, uint32_t handle, const char name[GLOBALNAME_LENGTH] ) { +dispatch_name_queue(struct harbor *h, struct keyvalue * node) { + struct msg_queue * queue = node->queue; + uint32_t handle = node->value; int harbor_id = handle >> HANDLE_REMOTE_SHIFT; assert(harbor_id != 0); struct skynet_context * context = h->ctx; - int fd = h->remote_fd[harbor_id]; - if (fd < 0) { - char tmp [GLOBALNAME_LENGTH+1]; - memcpy(tmp, name , GLOBALNAME_LENGTH); - tmp[GLOBALNAME_LENGTH] = '\0'; - skynet_error(context, "Drop message to %s (in harbor %d)",tmp,harbor_id); + struct slave *s = &h->s[harbor_id]; + int fd = s->fd; + if (fd == 0) { + if (s->status == STATUS_DOWN) { + char tmp [GLOBALNAME_LENGTH+1]; + memcpy(tmp, node->key, GLOBALNAME_LENGTH); + tmp[GLOBALNAME_LENGTH] = '\0'; + skynet_error(context, "Drop message to %s (in harbor %d)",tmp,harbor_id); + } else { + if (s->queue == NULL) { + s->queue = node->queue; + node->queue = NULL; + } else { + struct msg * m; + while ((m = pop_queue(queue))!=NULL) { + push_queue_msg(s->queue, m); + } + } + } return; } - 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)); - cookie.destination |= (handle & HANDLE_MASK); - _header_to_message(&cookie, ptr); - _send_package(context, fd, m->buffer, m->size); - m = _pop_queue(queue); + struct msg * m; + while ((m = pop_queue(queue)) != NULL) { + m->header.destination |= (handle & HANDLE_MASK); + send_remote(context, fd, m->buffer, m->size, &m->header); } } static void -_update_remote_name(struct harbor *h, const char name[GLOBALNAME_LENGTH], uint32_t handle) { - struct keyvalue * node = _hash_search(h->map, name); +dispatch_queue(struct harbor *h, int id) { + struct slave *s = &h->s[id]; + int fd = s->fd; + assert(fd != 0); + + struct msg_queue *queue = s->queue; + if (queue == NULL) + return; + + struct msg * m; + while ((m = pop_queue(queue)) != NULL) { + send_remote(h->ctx, fd, m->buffer, m->size, &m->header); + } + release_queue(queue); + s->queue = NULL; +} + +static void +push_socket_data(struct harbor *h, const struct skynet_socket_message * message) { + assert(message->type == SKYNET_SOCKET_TYPE_DATA); + int fd = message->id; + int i; + int id = 0; + struct slave * s = NULL; + for (i=1;is[i].fd == fd) { + s = &h->s[i]; + id = i; + break; + } + } + if (s == NULL) { + skynet_free(message->buffer); + skynet_error(h->ctx, "Invalid socket fd (%d) data", fd); + return; + } + uint8_t * buffer = (uint8_t *)message->buffer; + int size = message->ud; + + for (;;) { + switch(s->status) { + case STATUS_HANDSHAKE: { + // check id + uint8_t remote_id = buffer[0]; + if (remote_id != id) { + skynet_error(h->ctx, "Invalid shakehand id (%d) from fd = %d , harbor = %d", id, fd, remote_id); + close_harbor(h,id); + return; + } + ++buffer; + --size; + s->status = STATUS_HEADER; + + dispatch_queue(h, id); + + if (size == 0) { + break; + } + // go though + } + case STATUS_HEADER: + if (size < 4) { + s->length = size; + memcpy(s->size, buffer, size); + return; + } else { + // big endian 4 bytes length + if (buffer[0] != 0) { + skynet_error(h->ctx, "Message is too long from harbor %d", id); + close_harbor(h,id); + return; + } + s->length = buffer[1] << 16 | buffer[2] << 8 | buffer[3]; + s->read = 0; + s->recv_buffer = skynet_malloc(s->length); + s->status = STATUS_CONTENT; + buffer += 4; + size -= 4; + if (size == 0) { + return; + } + } + // go though + case STATUS_CONTENT: + if (size < s->length - s->read) { + memcpy(s->recv_buffer + s->read, buffer, size); + s->read += size; + return; + } + int need = s->length - s->read; + memcpy(s->recv_buffer + s->read, buffer, need); + forward_local_messsage(h, s->recv_buffer, s->length); + s->length = 0; + s->read = 0; + s->recv_buffer = NULL; + size -= need; + buffer += need; + s->status = STATUS_HEADER; + if (size == 0) + return; + break; + default: + return; + } + } +} + +static void +update_name(struct harbor *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 = hash_insert(h->map, name); } node->value = handle; if (node->queue) { - _dispatch_queue(h, node->queue, handle, name); - _release_queue(node->queue); + dispatch_name_queue(h, node); + release_queue(node->queue); node->queue = NULL; } } -static void -_request_master(struct harbor *h, const char name[GLOBALNAME_LENGTH], size_t i, uint32_t handle) { - uint8_t buffer[4+i]; - to_bigendian(buffer, handle); - memcpy(buffer+4,name,i); - - if (h->master_fd >= 0) { - _send_package(h->ctx, h->master_fd, buffer, 4+i); - } -} - -/* - update global name to master - - 2 bytes (size) - 4 bytes (handle) (handle == 0 for request) - n bytes string (name) - */ - static int -_remote_send_handle(struct harbor *h, uint32_t source, uint32_t destination, int type, int session, const char * msg, size_t sz) { +remote_send_handle(struct harbor *h, uint32_t source, uint32_t destination, int type, int session, const char * msg, size_t sz) { int harbor_id = destination >> HANDLE_REMOTE_SHIFT; - assert(harbor_id != 0); struct skynet_context * context = h->ctx; if (harbor_id == h->id) { // local message @@ -468,95 +496,63 @@ _remote_send_handle(struct harbor *h, uint32_t source, uint32_t destination, int return 1; } - int fd = h->remote_fd[harbor_id]; - if (fd >= 0 && h->connected[harbor_id]) { + struct slave * s = &h->s[harbor_id]; + if (s->fd == 0 || s->status == STATUS_HANDSHAKE) { + if (s->status == STATUS_DOWN) { + // throw an error return to source + // report the destination is dead + skynet_send(context, destination, source, PTYPE_ERROR, 0 , NULL, 0); + skynet_error(context, "Drop message to harbor %d from %x to %x (session = %d, msgsz = %d)",harbor_id, source, destination,session,(int)sz); + } else { + struct remote_message_header header; + header.source = source; + header.destination = type << HANDLE_REMOTE_SHIFT; + header.session = (uint32_t)session; + push_queue(s->queue, (void *)msg, sz, &header); + return 1; + } + } else { struct remote_message_header cookie; cookie.source = source; cookie.destination = (destination & HANDLE_MASK) | ((uint32_t)type << HANDLE_REMOTE_SHIFT); cookie.session = (uint32_t)session; - _send_remote(context, fd, msg,sz,&cookie); - } else { - // throw an error return to source - // report the destination is dead - skynet_send(context, destination, source, PTYPE_ERROR, 0 , NULL, 0); - skynet_error(context, "Drop message to harbor %d from %x to %x (session = %d, msgsz = %d)",harbor_id, source, destination,session,(int)sz); + send_remote(context, s->fd, msg,sz,&cookie); } + return 0; } -static void -_remote_register_name(struct harbor *h, const char name[GLOBALNAME_LENGTH], uint32_t handle) { - int i; - for (i=0;imap, name); +remote_send_name(struct harbor *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); + node = hash_insert(h->map, name); } if (node->value == 0) { if (node->queue == NULL) { - node->queue = _new_queue(); + 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); - // 0 for request - _remote_register_name(h, name, 0); + push_queue(node->queue, (void *)msg, sz, &header); + char query[2+GLOBALNAME_LENGTH+1] = "Q "; + query[2+GLOBALNAME_LENGTH] = 0; + memcpy(query+2, name, GLOBALNAME_LENGTH); + skynet_send(h->ctx, 0, h->slave, PTYPE_TEXT, 0, query, strlen(query)); return 1; } else { - return _remote_send_handle(h, source, node->value, type, session, msg, sz); + return remote_send_handle(h, source, node->value, type, session, msg, sz); } } -static int -harbor_id(struct harbor *h, int fd) { - int i; - for (i=1;iremote_fd[i] == fd) - return i; - } - return 0; -} - static void -close_harbor(struct harbor *h, int fd) { - if (fd == h->master_fd) { - skynet_socket_close(h->ctx, fd); - skynet_error(h->ctx, "Master disconnected"); - h->master_fd = -1; - return; - } - int id = harbor_id(h,fd); - - if (id == 0) - return; - skynet_error(h->ctx, "Harbor %d closed",id); - skynet_socket_close(h->ctx, fd); - h->remote_fd[id] = -1; - response_close(h,id); -} - -static void -open_harbor(struct harbor *h, int fd) { - int id = harbor_id(h,fd); - if (id == 0) - return; - assert(h->connected[id] == false); - monitor_clear(h, id); - h->connected[id] = true; - skynet_error(h->ctx, "Harbor(%d) connected", id); +handshake(struct harbor *h, int id) { + struct slave *s = &h->s[id]; + uint8_t * handshake = skynet_malloc(1); + handshake[0] = (uint8_t)h->id; + skynet_socket_send(h->ctx, s->fd, handshake, 1); } static void @@ -565,8 +561,7 @@ harbor_command(struct harbor * h, const char * msg, size_t sz, int session, uint int s = (int)sz; s -= 2; switch(msg[0]) { - case 'R' : { - // register global name + case 'N' : { if (s <=0 || s>= GLOBALNAME_LENGTH) { skynet_error(h->ctx, "Invalid global name %s", name); return; @@ -575,35 +570,35 @@ harbor_command(struct harbor * h, const char * msg, size_t sz, int session, uint memset(&rn, 0, sizeof(rn)); memcpy(rn.name, name, s); rn.handle = source; - _remote_register_name(h, rn.name, rn.handle); + update_name(h, rn.name, rn.handle); break; } - case 'C' : - case 'M' : { - if (s <= 0) { - skynet_error(h->ctx, "Invalid harbor montior"); - skynet_send(h->ctx, 0, source, PTYPE_ERROR, session, NULL, 0); + case 'S' : + case 'A' : { + char buffer[s+1]; + memcpy(buffer, name, s); + buffer[s] = 0; + int fd=0, id=0; + sscanf(buffer, "%d %d",&fd,&id); + if (fd == 0 || id <= 0 || id>=REMOTE_MAX) { + skynet_error(h->ctx, "Invalid command %c %s", msg[0], buffer); return; } - int hid = strtol(name, NULL, 10); - if (hid <= 0 || hid >= REMOTE_MAX) { - skynet_error(h->ctx, "Invalid harbor montior id : %s", name); - skynet_send(h->ctx, 0, source, PTYPE_ERROR, session, NULL, 0); + struct slave * slave = &h->s[id]; + if (slave->fd != 0) { + skynet_error(h->ctx, "Harbor %d alreay exist", id); return; } - if (msg[0] == 'M') { - if (!h->connected[hid]) { - skynet_send(h->ctx, 0, source, PTYPE_RESPONSE, session, NULL, 0); - return; - } + slave->fd = fd; + + skynet_socket_start(h->ctx, fd); + handshake(h, id); + if (msg[0] == 'S') { + slave->status = STATUS_HANDSHAKE; } else { - assert(msg[0] == 'C'); - if (h->connected[hid]) { - skynet_send(h->ctx, 0, source, PTYPE_RESPONSE, session, NULL, 0); - return; - } + slave->status = STATUS_HEADER; + dispatch_queue(h,id); } - monitor_add(h, hid, source, session); break; } default: @@ -613,63 +608,48 @@ harbor_command(struct harbor * h, const char * msg, size_t sz, int session, uint } static int -_mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { +harbor_id(struct harbor *h, int fd) { + int i; + for (i=1;is[i]; + if (s->fd == fd) { + return i; + } + } + return 0; +} + +static int +mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { struct harbor * h = ud; switch (type) { case PTYPE_SOCKET: { const struct skynet_socket_message * message = msg; switch(message->type) { case SKYNET_SOCKET_TYPE_DATA: + push_socket_data(h, message); skynet_free(message->buffer); - skynet_error(context, "recv invalid socket message (size=%d)", message->ud); - break; - case SKYNET_SOCKET_TYPE_ACCEPT: - skynet_error(context, "recv invalid socket accept message"); break; case SKYNET_SOCKET_TYPE_ERROR: - case SKYNET_SOCKET_TYPE_CLOSE: - close_harbor(h, message->id); + case SKYNET_SOCKET_TYPE_CLOSE: { + int id = harbor_id(h, message->id); + if (id) { + report_harbor_down(h,id); + } else { + skynet_error(context, "Unkown fd (%d) closed", message->id); + } break; + } case SKYNET_SOCKET_TYPE_CONNECT: - open_harbor(h, message->id); + // fd forward to this service + break; + default: + skynet_error(context, "recv invalid socket message type %d", type); break; } return 0; } case PTYPE_HARBOR: { - // remote message in - const char * cookie = msg; - cookie += sz - HEADER_COOKIE_LENGTH; - struct remote_message_header header; - _message_to_header((const uint32_t *)cookie, &header); - if (header.source == 0) { - if (header.destination < REMOTE_MAX) { - // 1 byte harbor id (0~255) - // update remote harbor address - char ip [sz - HEADER_COOKIE_LENGTH + 1]; - memcpy(ip, msg, sz-HEADER_COOKIE_LENGTH); - ip[sz-HEADER_COOKIE_LENGTH] = '\0'; - _update_remote_address(h, header.destination, ip); - } else { - // update global name - if (sz - HEADER_COOKIE_LENGTH > GLOBALNAME_LENGTH) { - char name[sz-HEADER_COOKIE_LENGTH+1]; - memcpy(name, msg, sz-HEADER_COOKIE_LENGTH); - name[sz-HEADER_COOKIE_LENGTH] = '\0'; - skynet_error(context, "Global name is too long %s", name); - } - _update_remote_name(h, msg, header.destination); - } - } else { - uint32_t destination = header.destination; - int type = (destination >> HANDLE_REMOTE_SHIFT) | PTYPE_TAG_DONTCOPY; - destination = (destination & HANDLE_MASK) | ((uint32_t)h->id << HANDLE_REMOTE_SHIFT); - skynet_send(context, header.source, destination, type, (int)header.session, (void *)msg, sz-HEADER_COOKIE_LENGTH); - return 1; - } - return 0; - } - case PTYPE_SYSTEM: { harbor_command(h, msg,sz,session,source); return 0; } @@ -677,11 +657,11 @@ _mainloop(struct skynet_context * context, void * ud, int type, int session, uin // remote message out const struct remote_message *rmsg = msg; if (rmsg->destination.handle == 0) { - if (_remote_send_name(h, source , rmsg->destination.name, type, session, rmsg->message, rmsg->sz)) { + if (remote_send_name(h, source , rmsg->destination.name, type, session, rmsg->message, rmsg->sz)) { return 0; } } else { - if (_remote_send_handle(h, source , rmsg->destination.handle, type, session, rmsg->message, rmsg->sz)) { + if (remote_send_handle(h, source , rmsg->destination.handle, type, session, rmsg->message, rmsg->sz)) { return 0; } } @@ -691,48 +671,19 @@ _mainloop(struct skynet_context * context, void * ud, int type, int session, uin } } -static void -_launch_gate(struct skynet_context * ctx, const char * local_addr) { - char tmp[128]; - sprintf(tmp,"gate L ! %s %d %d 0",local_addr, PTYPE_HARBOR, REMOTE_MAX); - const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp); - if (gate_addr == NULL) { - fprintf(stderr, "Harbor : launch gate failed\n"); - exit(1); - } - uint32_t gate = strtoul(gate_addr+1 , NULL, 16); - if (gate == 0) { - fprintf(stderr, "Harbor : launch gate invalid %s", gate_addr); - exit(1); - } - const char * self_addr = skynet_command(ctx, "REG", NULL); - int n = sprintf(tmp,"broker %s",self_addr); - skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, tmp, n); - skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, "start", 5); -} - int harbor_init(struct harbor *h, struct skynet_context *ctx, const char * args) { h->ctx = ctx; - int sz = strlen(args)+1; - char master_addr[sz]; - char local_addr[sz]; int harbor_id = 0; - sscanf(args,"%s %s %d",master_addr, local_addr, &harbor_id); - h->master_addr = skynet_strdup(master_addr); - h->id = harbor_id; - h->master_fd = _connect_to(h, master_addr); - if (h->master_fd == -1) { - fprintf(stderr, "Harbor: Connect to master failed\n"); - exit(1); + uint32_t slave = 0; + sscanf(args,"%d %u", &harbor_id, &slave); + if (slave == 0) { + return 1; } + h->id = harbor_id; + h->slave = slave; + skynet_callback(ctx, h, mainloop); skynet_harbor_start(ctx); - h->local_addr = skynet_strdup(local_addr); - - _launch_gate(ctx, local_addr); - skynet_callback(ctx, h, _mainloop); - _request_master(h, local_addr, strlen(local_addr), harbor_id); - return 0; } diff --git a/service-src/service_master.c b/service-src/service_master.c deleted file mode 100644 index a4fb8b1b..00000000 --- a/service-src/service_master.c +++ /dev/null @@ -1,314 +0,0 @@ -#include "skynet.h" -#include "skynet_harbor.h" -#include "skynet_socket.h" - -#include -#include -#include -#include -#include -#include - -#define HASH_SIZE 4096 - -struct name { - struct name * next; - char key[GLOBALNAME_LENGTH]; - uint32_t hash; - uint32_t value; -}; - -struct namemap { - struct name *node[HASH_SIZE]; -}; - -struct master { - struct skynet_context *ctx; - int remote_fd[REMOTE_MAX]; - bool connected[REMOTE_MAX]; - char * remote_addr[REMOTE_MAX]; - struct namemap map; -}; - -struct master * -master_create() { - struct master *m = skynet_malloc(sizeof(*m)); - int i; - for (i=0;iremote_fd[i] = -1; - m->remote_addr[i] = NULL; - m->connected[i] = false; - } - memset(&m->map, 0, sizeof(m->map)); - return m; -} - -void -master_release(struct master * m) { - int i; - struct skynet_context *ctx = m->ctx; - for (i=0;iremote_fd[i]; - if (fd >= 0) { - assert(ctx); - skynet_socket_close(ctx, fd); - } - skynet_free(m->remote_addr[i]); - } - for (i=0;imap.node[i]; - while (node) { - struct name * next = node->next; - skynet_free(node); - node = next; - } - } - skynet_free(m); -} - -static struct name * -_search_name(struct master *m, char name[GLOBALNAME_LENGTH]) { - uint32_t *ptr = (uint32_t *) name; - uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3]; - struct name * node = m->map.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 name * -_insert_name(struct master *m, char name[GLOBALNAME_LENGTH]) { - uint32_t *ptr = (uint32_t *)name; - uint32_t h = ptr[0] ^ ptr[1] ^ ptr[2] ^ ptr[3]; - struct name **pname = &m->map.node[h % HASH_SIZE]; - struct name * node = skynet_malloc(sizeof(*node)); - memcpy(node->key, name, GLOBALNAME_LENGTH); - node->next = *pname; - node->hash = h; - node->value = 0; - *pname = node; - return node; -} - -static void -_copy_name(char *name, const char * buffer, size_t sz) { - if (sz < GLOBALNAME_LENGTH) { - memcpy(name, buffer, sz); - memset(name+sz, 0 , GLOBALNAME_LENGTH - sz); - } else { - memcpy(name, buffer, GLOBALNAME_LENGTH); - } -} - -static void -_connect_to(struct master *m, int id) { - assert(m->connected[id] == false); - struct skynet_context * ctx = m->ctx; - const char *ipaddress = m->remote_addr[id]; - char * portstr = strchr(ipaddress,':'); - if (portstr==NULL) { - skynet_error(ctx, "Harbor %d : address invalid (%s)",id, ipaddress); - return; - } - int sz = portstr - ipaddress; - char tmp[sz + 1]; - memcpy(tmp,ipaddress,sz); - tmp[sz] = '\0'; - int port = strtol(portstr+1,NULL,10); - skynet_error(ctx, "Master connect to harbor(%d) %s:%d", id, tmp, port); - m->remote_fd[id] = skynet_socket_connect(ctx, tmp, port); - m->connected[id] = true; -} - -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 void -_send_to(struct master *m, int id, const void * buf, int sz, uint32_t handle) { - uint8_t * buffer= (uint8_t *)skynet_malloc(4 + sz + 12); - to_bigendian(buffer, sz+12); - memcpy(buffer+4, buf, sz); - to_bigendian(buffer+4+sz, 0); - to_bigendian(buffer+4+sz+4, handle); - to_bigendian(buffer+4+sz+8, 0); - - sz += 4 + 12; - - if (skynet_socket_send(m->ctx, m->remote_fd[id], buffer, sz)) { - skynet_error(m->ctx, "Harbor %d : send error", id); - } -} - -static void -_broadcast(struct master *m, const char *name, size_t sz, uint32_t handle) { - int i; - for (i=1;iremote_fd[i]; - if (fd < 0 || m->connected[i]==false) - continue; - _send_to(m, i , name, sz, handle); - } -} - -static void -_request_name(struct master *m, const char * buffer, size_t sz) { - char name[GLOBALNAME_LENGTH]; - _copy_name(name, buffer, sz); - struct name * n = _search_name(m, name); - if (n == NULL) { - return; - } - _broadcast(m, name, GLOBALNAME_LENGTH, n->value); -} - -static void -_update_name(struct master *m, uint32_t handle, const char * buffer, size_t sz) { - char name[GLOBALNAME_LENGTH]; - _copy_name(name, buffer, sz); - struct name * n = _search_name(m, name); - if (n==NULL) { - n = _insert_name(m,name); - } - n->value = handle; - _broadcast(m,name,GLOBALNAME_LENGTH, handle); -} - -static void -close_harbor(struct master *m, int harbor_id) { - if (m->connected[harbor_id]) { - struct skynet_context * context = m->ctx; - skynet_socket_close(context, m->remote_fd[harbor_id]); - m->remote_fd[harbor_id] = -1; - m->connected[harbor_id] = false; - } -} - -static void -_update_address(struct master *m, int harbor_id, const char * buffer, size_t sz) { - if (m->remote_fd[harbor_id] >= 0) { - close_harbor(m, harbor_id); - } - skynet_free(m->remote_addr[harbor_id]); - char * addr = skynet_malloc(sz+1); - memcpy(addr, buffer, sz); - addr[sz] = '\0'; - m->remote_addr[harbor_id] = addr; - _connect_to(m, harbor_id); -} - -static int -socket_id(struct master *m, int id) { - int i; - for (i=1;iremote_fd[i] == id) - return i; - } - return 0; -} - -static void -on_connected(struct master *m, int id) { - _broadcast(m, m->remote_addr[id], strlen(m->remote_addr[id]), id); -// m->connected[id] = true; - int i; - for (i=1;iremote_addr[i]; - if (addr == NULL || m->connected[i] == false) { - continue; - } - _send_to(m, id , addr, strlen(addr), i); - } -} - -static void -dispatch_socket(struct master *m, const struct skynet_socket_message *msg, int sz) { - int id = socket_id(m, msg->id); - switch(msg->type) { - case SKYNET_SOCKET_TYPE_CONNECT: - assert(id); - on_connected(m, id); - break; - case SKYNET_SOCKET_TYPE_ERROR: - skynet_error(m->ctx, "socket error on harbor %d", id); - // go though, close socket - case SKYNET_SOCKET_TYPE_CLOSE: - close_harbor(m, id); - break; - default: - skynet_error(m->ctx, "Invalid socket message type %d", msg->type); - break; - } -} - - -/* - update global name to master - - 4 bytes (handle) (handle == 0 for request) - n bytes string (name) - */ - -static int -_mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { - if (type == PTYPE_SOCKET) { - dispatch_socket(ud, msg, (int)sz); - return 0; - } - if (type != PTYPE_HARBOR) { - skynet_error(context, "None harbor message recv from %x (type = %d)", source, type); - return 0; - } - assert(sz >= 4); - struct master *m = ud; - const uint8_t *handlen = msg; - uint32_t handle = handlen[0]<<24 | handlen[1]<<16 | handlen[2]<<8 | handlen[3]; - sz -= 4; - const char * name = msg; - name += 4; - - if (handle == 0) { - _request_name(m , name, sz); - } else if (handle < REMOTE_MAX) { - _update_address(m , handle, name, sz); - } else { - _update_name(m , handle, name, sz); - } - - return 0; -} - -int -master_init(struct master *m, struct skynet_context *ctx, const char * args) { - char tmp[strlen(args) + 32]; - sprintf(tmp,"gate L ! %s %d %d 0",args,PTYPE_HARBOR,REMOTE_MAX); - const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp); - if (gate_addr == NULL) { - skynet_error(ctx, "Master : launch gate failed"); - return 1; - } - uint32_t gate = strtoul(gate_addr+1, NULL, 16); - if (gate == 0) { - skynet_error(ctx, "Master : launch gate invalid %s", gate_addr); - return 1; - } - const char * self_addr = skynet_command(ctx, "REG", NULL); - int n = sprintf(tmp,"broker %s",self_addr); - skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, tmp, n); - skynet_send(ctx, 0, gate, PTYPE_TEXT, 0, "start", 5); - - skynet_callback(ctx, m, _mainloop); - - m->ctx = ctx; - return 0; -} diff --git a/service/bootstrap.lua b/service/bootstrap.lua index 78f98930..11303577 100644 --- a/service/bootstrap.lua +++ b/service/bootstrap.lua @@ -3,29 +3,36 @@ local harbor = require "skynet.harbor" skynet.start(function() local standalone = skynet.getenv "standalone" + + local launcher = assert(skynet.launch("snlua","launcher")) + skynet.name(".launcher", launcher) + local harbor_id = tonumber(skynet.getenv "harbor") if harbor_id == 0 then assert(standalone == nil) standalone = true skynet.setenv("standalone", "true") - local dummy = assert(skynet.launch("dummy")) - harbor.init(dummy) - else - local master_addr = skynet.getenv "master" + local slave = skynet.newservice "cdummy" + if slave == nil then + skynet.abort() + end + skynet.name(".slave", slave) + + else if standalone then - assert(skynet.launch("master", standalone)) + if not skynet.newservice "cmaster" then + skynet.abort() + end end - local local_addr = skynet.getenv "address" - - local h = assert(skynet.launch("harbor",master_addr, local_addr, harbor_id)) - harbor.init(h) + local slave = skynet.newservice "cslave" + if slave == nil then + skynet.abort() + end + skynet.name(".slave", slave) end - local launcher = assert(skynet.launch("snlua","launcher")) - skynet.name(".launcher", launcher) - if standalone then local datacenter = assert(skynet.newservice "datacenterd") skynet.name("DATACENTER", datacenter) diff --git a/service/cdummy.lua b/service/cdummy.lua new file mode 100644 index 00000000..fddb2c01 --- /dev/null +++ b/service/cdummy.lua @@ -0,0 +1,47 @@ +local skynet = require "skynet" + +local globalname = {} +local harbor = {} + +skynet.register_protocol { + name = "harbor", + id = skynet.PTYPE_HARBOR, + pack = function(...) return ... end, + unpack = skynet.tostring, +} + +skynet.register_protocol { + name = "text", + id = skynet.PTYPE_TEXT, + pack = function(...) return ... end, + unpack = skynet.tostring, +} + +function harbor.REGISTER(name, handle) + assert(globalname[name] == nil) + globalname[name] = handle + skynet.redirect(harbor_service, handle, "harbor", 0, "N " .. name) +end + +function harbor.LINK(id) + skynet.ret() +end + +function harbor.CONNECT(id) + skynet.error("Can't connect to other harbor in single node mode") +end + +skynet.start(function() + local harbor_id = tonumber(skynet.getenv "harbor") + assert(harbor_id == 0) + + skynet.dispatch("lua", function (session,source,command,...) + local f = assert(harbor[command]) + f(...) + end) + skynet.dispatch("text", function(session,source,command) + -- ignore all the command + end) + + harbor_service = assert(skynet.launch("harbor", harbor_id, skynet.self())) +end) diff --git a/service/cmaster.lua b/service/cmaster.lua new file mode 100644 index 00000000..9c38515b --- /dev/null +++ b/service/cmaster.lua @@ -0,0 +1,123 @@ +local skynet = require "skynet" +local socket = require "socket" + +--[[ + master manage data : + 1. all the slaves address : id -> ipaddr:port + 2. all the global names : name -> address + + master hold connections from slaves . + + protocol slave->master : + package size 1 byte + type 1 byte : + 'H' : HANDSHAKE, report slave id, and address. + 'R' : REGISTER name address + 'Q' : QUERY name + + + protocol master->slave: + package size 1 byte + type 1 byte : + 'W' : WAIT n + 'C' : CONNECT slave_id slave_address + 'N' : NAME globalname address + 'D' : DISCONNECT slave_id +]] + +local slave_node = {} +local global_name = {} + +local function read_package(fd) + local sz = socket.read(fd, 1) + assert(sz, "closed") + sz = string.byte(sz) + local content = socket.read(fd, sz) + return skynet.unpack(content) +end + +local function pack_package(...) + local message = skynet.packstring(...) + local size = #message + assert(size <= 255 , "too long") + return string.char(size) .. message +end + +local function report_slave(fd, slave_id, slave_addr) + local message = pack_package("C", slave_id, slave_addr) + local n = 0 + for k,v in pairs(slave_node) do + socket.write(v.fd, message) + n = n + 1 + end + socket.write(fd, pack_package("W", n)) +end + +local function handshake(fd) + local t, slave_id, slave_addr = read_package(fd) + assert(t=='H', "Invalid handshake type " .. t) + assert(slave_id ~= 0 , "Invalid slave id 0") + if slave_node[slave_id] then + error(string.format("Slave %d already register on %s", slave_id, slave_node[slave_id].addr)) + end + report_slave(fd, slave_id, slave_addr) + slave_node[slave_id] = { + fd = fd, + id = slave_id, + addr = slave_addr, + } + return slave_id , slave_addr +end + +local function dispatch_slave(fd) + local t, name, address = read_package(fd) + if t == 'R' then + -- register name + assert(type(address)=="number", "Invalid request") + if not global_name[name] then + global_name[name] = address + end + local message = pack_package("N", name, address) + for k,v in pairs(slave_node) do + socket.write(v.fd, message) + end + elseif t == 'Q' then + -- query name + local address = global_name[name] + if address then + socket.write(fd, pack_package("N", name, address)) + end + else + skynet.error("Invalid slave message type " .. t) + end +end + +local function monitor_slave(slave_id, slave_address) + local fd = slave_node[slave_id].fd + skynet.error(string.format("Harbor %d (fd=%d) report %s", slave_id, fd, slave_address)) + while pcall(dispatch_slave, fd) do end + skynet.error("slave " ..slave_id .. " is down") + local message = pack_package("D", slave_id) + slave_node[slave_id].fd = 0 + for k,v in pairs(slave_node) do + socket.write(v.fd, message) + end + socket.close(fd) +end + +skynet.start(function() + local master_addr = skynet.getenv "standalone" + skynet.error("master listen socket " .. tostring(master_addr)) + local fd = socket.listen(master_addr) + socket.start(fd , function(id, addr) + skynet.error("connect from " .. addr .. " " .. id) + socket.start(id) + local ok, slave, slave_addr = pcall(handshake, id) + if ok then + skynet.fork(monitor_slave, slave, slave_addr) + else + skynet.error(string.format("disconnect fd = %d, error = %s", id, slave)) + socket.close(id) + end + end) +end) diff --git a/service/cslave.lua b/service/cslave.lua new file mode 100644 index 00000000..e03c1b83 --- /dev/null +++ b/service/cslave.lua @@ -0,0 +1,222 @@ +local skynet = require "skynet" +local socket = require "socket" + +local slaves = {} +local connect_queue = {} +local globalname = {} +local harbor = {} +local harbor_service +local monitor = {} + +local function read_package(fd) + local sz = socket.read(fd, 1) + assert(sz, "closed") + sz = string.byte(sz) + local content = socket.read(fd, sz) + return skynet.unpack(content) +end + +local function pack_package(...) + local message = skynet.packstring(...) + local size = #message + assert(size <= 255 , "too long") + return string.char(size) .. message +end + +local function monitor_clear(id) + local v = monitor[id] + if v then + monitor[id] = nil + for _, v in ipairs(v) do + skynet.redirect(v.address, 0, "response", v.session, "") + end + end +end + +local function connect_slave(slave_id, address) + local ok, err = pcall(function() + if slaves[slave_id] == nil then + local fd = socket.open(address) + skynet.error(string.format("Connect to harbor %d (fd=%d), %s", slave_id, fd, address)) + slaves[slave_id] = fd + monitor_clear(slave_id) + socket.abandon(fd) + skynet.send(harbor_service, "harbor", string.format("S %d %d",fd,slave_id)) + end + end) + if not ok then + skynet.error(err) + end +end + +local function ready() + local queue = connect_queue + connect_queue = nil + for k,v in pairs(queue) do + connect_slave(k,v) + end + for name,address in pairs(globalname) do + skynet.redirect(harbor_service, address, "harbor", "N " .. name) + end +end + +local function monitor_master(master_fd) + while true do + local ok, t, id_name, address = pcall(read_package,master_fd) + if ok then + if t == 'C' then + if connect_queue then + connect_queue[id_name] = address + else + connect_slave(id_name, address) + end + elseif t == 'N' then + globalname[id_name] = address + if connect_queue == nil then + skynet.redirect(harbor_service, address, "harbor", 0, "N " .. id_name) + end + elseif t == 'D' then + local fd = slaves[id_name] + slaves[id_name] = false + if fd then + socket.close(fd) + end + end + else + skynet.error("Master disconnect") + socket.close(master_fd) + break + end + end +end + +local function accept_slave(fd) + socket.start(fd) + local id = socket.read(fd, 1) + if not id then + skynet.error(string.format("Connection (fd =%d) closed", fd)) + socket.close(fd) + return + end + id = string.byte(id) + if slaves[id] ~= nil then + skynet.error(string.format("Slave %d exist (fd =%d)", id, fd)) + socket.close(fd) + return + end + slaves[id] = fd + monitor_clear(id) + socket.abandon(fd) + skynet.error(string.format("Harbor %d connected (fd = %d)", id, fd)) + skynet.send(harbor_service, "harbor", string.format("A %d %d", fd, id)) +end + +skynet.register_protocol { + name = "harbor", + id = skynet.PTYPE_HARBOR, + pack = function(...) return ... end, + unpack = skynet.tostring, +} + +skynet.register_protocol { + name = "text", + id = skynet.PTYPE_TEXT, + pack = function(...) return ... end, + unpack = skynet.tostring, +} + +local function monitor_harbor(master_fd) + return function(session, source, command) + local t = string.sub(command, 1, 1) + local arg = string.sub(command, 3) + if t == "Q" then + -- query name + if globalname[arg] then + skynet.redirect(harbor_service, globalname[arg], "harbor", "N " .. arg) + else + socket.write(master_fd, pack_package("Q", arg)) + end + elseif t == "D" then + -- harbor down + local id = tonumber(arg) + if slaves[id] then + monitor_clear(id) + end + slaves[id] = false + else + skynet.error("Unknown command ", command) + end + end +end + +function harbor.REGISTER(_,_, fd, name, handle) + assert(globalname[name] == nil) + globalname[name] = handle + socket.write(fd, pack_package("R", name, handle)) + skynet.redirect(harbor_service, handle, "harbor", 0, "N " .. name) +end + +function harbor.LINK(session, source, fd, id) + if slaves[id] then + if monitor[id] == nil then + monitor[id] = {} + end + table.insert(monitor[id], { address = source, session = session }) + else + skynet.ret() + end +end + +function harbor.CONNECT(session, source, fd, id) + if not slaves[id] then + if monitor[id] == nil then + monitor[id] = {} + end + table.insert(monitor[id], { address = source, session = session }) + else + skynet.ret() + end +end + +skynet.start(function() + local master_addr = skynet.getenv "master" + local harbor_id = tonumber(skynet.getenv "harbor") + local slave_address = assert(skynet.getenv "address") + local slave_fd = socket.listen(slave_address) + skynet.error("slave connect to master " .. tostring(master_addr)) + local master_fd = socket.open(master_addr) + + skynet.dispatch("lua", function (session,source,command,...) + local f = assert(harbor[command]) + f(session, source, master_fd, ...) + end) + skynet.dispatch("text", monitor_harbor(master_fd)) + + harbor_service = assert(skynet.launch("harbor", harbor_id, skynet.self())) + + local hs_message = pack_package("H", harbor_id, slave_address) + socket.write(master_fd, hs_message) + local t, n = read_package(master_fd) + assert(t == "W" and type(n) == "number", "slave shakehand failed") + skynet.error(string.format("Waiting for %d harbors", n)) + skynet.fork(monitor_master, master_fd) + if n > 0 then + local co = coroutine.running() + socket.start(slave_fd, function(fd, addr) + skynet.error(string.format("New connection (fd = %d, %s)",fd, addr)) + if pcall(accept_slave,fd) then + local s = 0 + for k,v in pairs(slaves) do + s = s + 1 + end + if s >= n then + skynet.wakeup(co) + end + end + end) + skynet.wait() + end + socket.close(slave_fd) + skynet.error("Shakehand ready") + skynet.fork(ready) +end) diff --git a/service/debug_console.lua b/service/debug_console.lua index d49dd809..b15c3889 100644 --- a/service/debug_console.lua +++ b/service/debug_console.lua @@ -88,7 +88,7 @@ end skynet.start(function() local listen_socket = socket.listen ("127.0.0.1", port) - print("Start debug console at 127.0.0.1",port) + skynet.error("Start debug console at 127.0.0.1 " .. port) socket.start(listen_socket , function(id, addr) local function print(...) local t = { ... }