diff --git a/.gitignore b/.gitignore index 9c452a72..2cbccc98 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ *.o *.a skynet +skynet.pid 3rd/lua/lua 3rd/lua/luac cservice diff --git a/3rd/jemalloc b/3rd/jemalloc index 3541a904..46c0af68 160000 --- a/3rd/jemalloc +++ b/3rd/jemalloc @@ -1 +1 @@ -Subproject commit 3541a904d6fb949f3f0aea05418ccce7cbd4b705 +Subproject commit 46c0af68bd248b04df75e4f92d5fb804c3d75340 diff --git a/HISTORY.md b/HISTORY.md index 07a4c501..27927ea5 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -1,6 +1,8 @@ Dev version ----------- * Bugfix: update lua-bson (signed 32bit int bug) +* Add cluster support +* Add daemon mode v0.2.1 (2014-5-19) ----------- diff --git a/Makefile b/Makefile index d7bb53cf..9c84351d 100644 --- a/Makefile +++ b/Makefile @@ -40,13 +40,15 @@ jemalloc : $(MALLOC_STATICLIB) # skynet -CSERVICE = snlua logger gate master harbor -LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile multicast +CSERVICE = snlua logger gate master harbor dummy +LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack \ + cjson clientsocket memory profile multicast \ + cluster SKYNET_SRC = skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c \ skynet_server.c skynet_start.c skynet_timer.c skynet_error.c \ skynet_harbor.c skynet_env.c skynet_monitor.c skynet_socket.c socket_server.c \ - malloc_hook.c + malloc_hook.c skynet_daemon.c all : \ $(SKYNET_BUILD_PATH)/skynet \ @@ -105,6 +107,9 @@ $(LUA_CLIB_PATH)/profile.so : lualib-src/lua-profile.c | $(LUA_CLIB_PATH) $(LUA_CLIB_PATH)/multicast.so : lualib-src/lua-multicast.c | $(LUA_CLIB_PATH) $(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ +$(LUA_CLIB_PATH)/cluster.so : lualib-src/lua-cluster.c | $(LUA_CLIB_PATH) + $(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ + clean : rm -f $(SKYNET_BUILD_PATH)/skynet $(CSERVICE_PATH)/*.so $(LUA_CLIB_PATH)/*.so diff --git a/examples/cluster1.lua b/examples/cluster1.lua new file mode 100644 index 00000000..73fad761 --- /dev/null +++ b/examples/cluster1.lua @@ -0,0 +1,9 @@ +local skynet = require "skynet" +local cluster = require "cluster" + +skynet.start(function() + skynet.newservice("simpledb") + print(skynet.call("SIMPLEDB", "lua", "SET", "a", "foobar")) + print(skynet.call("SIMPLEDB", "lua", "GET", "a")) + cluster.open(2528) +end) diff --git a/examples/cluster2.lua b/examples/cluster2.lua new file mode 100644 index 00000000..a61ff5c4 --- /dev/null +++ b/examples/cluster2.lua @@ -0,0 +1,6 @@ +local skynet = require "skynet" +local cluster = require "cluster" + +skynet.start(function() + print(cluster.call("db", "SIMPLEDB", "GET", "a")) +end) diff --git a/examples/clustername.lua b/examples/clustername.lua new file mode 100644 index 00000000..c5ed5e69 --- /dev/null +++ b/examples/clustername.lua @@ -0,0 +1 @@ +db = "127.0.0.1:2528" diff --git a/examples/config b/examples/config index f89af28b..087a9a14 100644 --- a/examples/config +++ b/examples/config @@ -12,3 +12,4 @@ lualoader = "lualib/loader.lua" -- preload = "./examples/preload.lua" -- run preload.lua before every lua service run snax = root.."examples/?.lua;"..root.."test/?.lua" cpath = root.."cservice/?.so" +-- daemon = "./skynet.pid" diff --git a/examples/config.c1 b/examples/config.c1 new file mode 100644 index 00000000..d1fdf165 --- /dev/null +++ b/examples/config.c1 @@ -0,0 +1,9 @@ +thread = 8 +logger = nil +harbor = 0 +start = "cluster1" +bootstrap = "snlua bootstrap" -- The service for bootstrap +luaservice = "./service/?.lua;./test/?.lua;./examples/?.lua" +lualoader = "lualib/loader.lua" +cpath = "./cservice/?.so" +cluster = "./examples/clustername.lua" \ No newline at end of file diff --git a/examples/config.c2 b/examples/config.c2 new file mode 100644 index 00000000..2e8cbb0b --- /dev/null +++ b/examples/config.c2 @@ -0,0 +1,9 @@ +thread = 8 +logger = nil +harbor = 0 +start = "cluster2" +bootstrap = "snlua bootstrap" -- The service for bootstrap +luaservice = "./service/?.lua;./test/?.lua;./examples/?.lua" +lualoader = "lualib/loader.lua" +cpath = "./cservice/?.so" +cluster = "./examples/clustername.lua" \ No newline at end of file diff --git a/lualib-src/lua-bson.c b/lualib-src/lua-bson.c index 077e538e..ade3cc4e 100644 --- a/lualib-src/lua-bson.c +++ b/lualib-src/lua-bson.c @@ -532,6 +532,9 @@ unpack_dict(lua_State *L, struct bson_reader *br, bool array) { break; case BSON_STRING: { int sz = read_int32(L, &t); + if (sz == 0) { + luaL_error(L, "Invalid bson string , length = 0"); + } lua_pushlstring(L, read_bytes(L, &t, sz), sz-1); break; } @@ -804,6 +807,9 @@ lreplace(lua_State *L) { write_int64(&b, i); break; } + default: + luaL_error(L, "Can't replace type %d", type); + break; } return 0; } diff --git a/lualib-src/lua-cluster.c b/lualib-src/lua-cluster.c new file mode 100644 index 00000000..b3b8a9d1 --- /dev/null +++ b/lualib-src/lua-cluster.c @@ -0,0 +1,205 @@ +#include +#include +#include + +#include "skynet.h" + +/* + uint32_t/string addr + uint32_t/session session + lightuserdata msg + uint32_t sz + + return + string request + uint32_t next_session + */ + +#define TEMP_LENGTH 0x10002 + +static void +fill_uint32(uint8_t * buf, uint32_t n) { + buf[0] = n & 0xff; + buf[1] = (n >> 8) & 0xff; + buf[2] = (n >> 16) & 0xff; + buf[3] = (n >> 24) & 0xff; +} + +static void +fill_header(lua_State *L, uint8_t *buf, int sz, void *msg) { + if (sz >= 0x10000) { + skynet_free(msg); + luaL_error(L, "request message is too long %d", sz); + } + buf[0] = (sz >> 8) & 0xff; + buf[1] = sz & 0xff; +} + +static void +packreq_number(lua_State *L, int session, void * msg, size_t sz) { + uint32_t addr = lua_tounsigned(L,1); + uint8_t buf[TEMP_LENGTH]; + fill_header(L, buf, sz+9, msg); + buf[2] = 0; + fill_uint32(buf+3, addr); + fill_uint32(buf+7, (uint32_t)session); + memcpy(buf+11,msg,sz); + + lua_pushlstring(L, (const char *)buf, sz+11); +} + +static void +packreq_string(lua_State *L, int session, void * msg, size_t sz) { + size_t namelen = 0; + const char *name = lua_tolstring(L, 1, &namelen); + if (name == NULL || namelen < 1 || namelen > 255) { + skynet_free(msg); + luaL_error(L, "name is too long %s", name); + } + + uint8_t buf[TEMP_LENGTH]; + fill_header(L, buf, sz+5+namelen, msg); + buf[2] = (uint8_t)namelen; + memcpy(buf+3, name, namelen); + fill_uint32(buf+3+namelen, (uint32_t)session); + memcpy(buf+7+namelen,msg,sz); + + lua_pushlstring(L, (const char *)buf, sz+7+namelen); +} + +static int +lpackrequest(lua_State *L) { + void *msg = lua_touserdata(L,3); + if (msg == NULL) { + return luaL_error(L, "Invalid request message"); + } + size_t sz = luaL_checkunsigned(L,4); + int session = luaL_checkinteger(L,2); + if (session <= 0) { + return luaL_error(L, "Invalid request session %d", session); + } + int addr_type = lua_type(L,1); + if (addr_type == LUA_TNUMBER) { + packreq_number(L, session, msg, sz); + } else { + packreq_string(L, session, msg, sz); + } + if (++session < 0) { + session = 1; + } + skynet_free(msg); + lua_pushinteger(L, session); + return 2; +} + +/* + string packed message + return + uint32_t or string addr + int session + string msg + */ + +static inline uint32_t +unpack_uint32(const uint8_t * buf) { + return buf[0] | buf[1]<<8 | buf[2]<<16 | buf[3]<<24; +} + +static int +unpackreq_number(lua_State *L, const uint8_t * buf, size_t sz) { + if (sz < 9) { + return luaL_error(L, "Invalid cluster message"); + } + uint32_t address = unpack_uint32(buf+1); + uint32_t session = unpack_uint32(buf+5); + lua_pushunsigned(L, address); + lua_pushunsigned(L, session); + lua_pushlstring(L, (const char *)buf+9, sz-9); + + return 3; +} + +static int +unpackreq_string(lua_State *L, const uint8_t * buf, size_t sz) { + size_t namesz = buf[0]; + if (sz < namesz + 5) { + return luaL_error(L, "Invalid cluster message"); + } + lua_pushlstring(L, (const char *)buf+1, namesz); + uint32_t session = unpack_uint32(buf + namesz + 1); + lua_pushunsigned(L, session); + lua_pushlstring(L, (const char *)buf+1+namesz+4, sz - namesz - 5); + + return 3; +} + +static int +lunpackrequest(lua_State *L) { + size_t sz; + const char *msg = luaL_checklstring(L,1,&sz); + if (msg[0] == 0) { + return unpackreq_number(L, (const uint8_t *)msg, sz); + } else { + return unpackreq_string(L, (const uint8_t *)msg, sz); + } +} + +/* + int session + lightuserdata msg + int sz + return string response + */ +static int +lpackresponse(lua_State *L) { + uint32_t session = luaL_checkunsigned(L,1); + void * msg = lua_touserdata(L,2); + size_t sz = luaL_checkunsigned(L, 3); + + uint8_t buf[TEMP_LENGTH]; + fill_header(L, buf, sz+4, msg); + fill_uint32(buf+2, session); + memcpy(buf+6,msg,sz); + + skynet_free(msg); + + lua_pushlstring(L, (const char *)buf, sz+6); + + return 1; +} + +/* + string packed response + return integer session + boolean ok + string msg + */ +static int +lunpackresponse(lua_State *L) { + size_t sz; + const char * buf = luaL_checklstring(L, 1, &sz); + if (sz < 4) { + return 0; + } + uint32_t session = unpack_uint32((const uint8_t *)buf); + lua_pushunsigned(L, session); + lua_pushboolean(L, 1); + lua_pushlstring(L, buf+4, sz-4); + + return 3; +} + +int +luaopen_cluster_c(lua_State *L) { + luaL_Reg l[] = { + { "packrequest", lpackrequest }, + { "unpackrequest", lunpackrequest }, + { "packresponse", lpackresponse }, + { "unpackresponse", lunpackresponse }, + { NULL, NULL }, + }; + luaL_checkversion(L); + luaL_newlib(L,l); + + return 1; +} diff --git a/lualib-src/lua-socket.c b/lualib-src/lua-socket.c index 3f22956c..2428914e 100644 --- a/lualib-src/lua-socket.c +++ b/lualib-src/lua-socket.c @@ -191,7 +191,7 @@ pop_lstring(lua_State *L, struct socket_buffer *sb, int sz, int skip) { /* userdata send_buffer table pool - integer sz + integer sz or string (big endian) */ static int lpopbuffer(lua_State *L) { @@ -200,7 +200,23 @@ lpopbuffer(lua_State *L) { return luaL_error(L, "Need buffer object at param 1"); } luaL_checktype(L,2,LUA_TTABLE); - int sz = luaL_checkinteger(L,3); + int type = lua_type(L,3); + int sz; + if (type == LUA_TNUMBER) { + sz = lua_tointeger(L,3); + } else { + size_t len; + const uint8_t * s = (const uint8_t *)luaL_checklstring(L, 3, &len); + if (len > 4 || len < 1) { + return luaL_error(L, "Invalid read %s", s); + } + int i; + sz = 0; + for (i=0;i<(int)len;i++) { + sz <<= 8; + sz |= s[i]; + } + } if (sb->size < sz || sz == 0) { lua_pushnil(L); } else { diff --git a/lualib/cluster.lua b/lualib/cluster.lua new file mode 100644 index 00000000..e4f6cfa2 --- /dev/null +++ b/lualib/cluster.lua @@ -0,0 +1,19 @@ +local skynet = require "skynet" + +local clusterd +local cluster = {} + +function cluster.call(node, address, ...) + -- skynet.pack(...) will free by cluster.c.packrequest + return skynet.call(clusterd, "lua", "req", node, address, skynet.pack(...)) +end + +function cluster.open(port) + skynet.call(clusterd, "lua", "listen", "0.0.0.0", port) +end + +skynet.init(function() + clusterd = skynet.uniqueservice("clusterd") +end) + +return cluster diff --git a/service-src/service_dummy.c b/service-src/service_dummy.c new file mode 100644 index 00000000..006a276f --- /dev/null +++ b/service-src/service_dummy.c @@ -0,0 +1,292 @@ +#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 int +_mainloop(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { + struct dummy * h = ud; + switch (type) { + case PTYPE_SYSTEM: { + // register name message + const struct remote_message *rmsg = msg; + assert (sz == sizeof(rmsg->destination)); + _update_name(h, rmsg->destination.name, rmsg->destination.handle); + return 0; + } + default: { + // remote message out + const struct remote_message *rmsg = msg; + if (rmsg->destination.handle == 0) { + _send_name(h, source , rmsg->destination.name, type, session, rmsg->message, rmsg->sz); + } else { + // local message + skynet_send(context, source, rmsg->destination.handle , type | PTYPE_TAG_DONTCOPY, session, (void *)rmsg->message, rmsg->sz); + } + return 0; + } + } +} + +int +dummy_init(struct dummy *d, struct skynet_context *ctx, const char * args) { + d->ctx = ctx; + skynet_harbor_start(ctx); + skynet_callback(ctx, d, _mainloop); + + return 0; +} diff --git a/service-src/service_logger.c b/service-src/service_logger.c index c78cf4d6..122cc77f 100644 --- a/service-src/service_logger.c +++ b/service-src/service_logger.c @@ -28,7 +28,7 @@ logger_release(struct logger * inst) { static int _logger(struct skynet_context * context, void *ud, int type, int session, uint32_t source, const void * msg, size_t sz) { struct logger * inst = ud; - fprintf(inst->handle, "[:%x] ",source); + fprintf(inst->handle, "[:%08x] ",source); fwrite(msg, sz , 1, inst->handle); fprintf(inst->handle, "\n"); fflush(inst->handle); diff --git a/service/bootstrap.lua b/service/bootstrap.lua index 51a57999..7c92b419 100644 --- a/service/bootstrap.lua +++ b/service/bootstrap.lua @@ -4,17 +4,24 @@ skynet.start(function() assert(skynet.launch("logger", skynet.getenv "logger")) local standalone = skynet.getenv "standalone" - local master_addr = skynet.getenv "master" + local harbor_id = tonumber(skynet.getenv "harbor") + if harbor_id == 0 then + assert(standalone == nil) + standalone = true + skynet.setenv("standalone", "true") + assert(skynet.launch("dummy")) + else + local master_addr = skynet.getenv "master" - if standalone then - assert(skynet.launch("master", master_addr)) + if standalone then + assert(skynet.launch("master", master_addr)) + end + + local local_addr = skynet.getenv "address" + + assert(skynet.launch("harbor",master_addr, local_addr, harbor_id)) end - local local_addr = skynet.getenv "address" - local harbor_id = skynet.getenv "harbor" - - assert(skynet.launch("harbor",master_addr, local_addr, harbor_id)) - local launcher = assert(skynet.launch("snlua","launcher")) skynet.name(".launcher", launcher) diff --git a/service/clusterd.lua b/service/clusterd.lua new file mode 100644 index 00000000..ef6bff68 --- /dev/null +++ b/service/clusterd.lua @@ -0,0 +1,69 @@ +local skynet = require "skynet" +local sc = require "socketchannel" +local socket = require "socket" +local cluster = require "cluster.c" + +local config_name = skynet.getenv "cluster" +local node_address = {} +assert(loadfile(config_name, "t", node_address))() +local node_session = {} +local command = {} + +local function read_response(sock) + local sz = sock:read(2) + local msg = sock:read(sz) + return cluster.unpackresponse(msg) -- session, ok, data +end + +local function open_channel(t, key) + local host, port = string.match(node_address[key], "([^:]+):(.*)$") + local c = sc.channel { + host = host, + port = tonumber(port), + response = read_response, + } + assert(c:connect(true)) + t[key] = c + node_session[key] = 1 + return c +end + +local node_channel = setmetatable({}, { __index = open_channel }) + +function command.listen(source, addr, port) + local gate = skynet.newservice("gate") + skynet.call(gate, "lua", "open", { address = addr, port = port }) + skynet.ret(skynet.pack(nil)) +end + +function command.req(source, node, addr, msg, sz) + local request + local c = node_channel[node] + local session = node_session[node] + -- msg is a local pointer, cluster.packrequest will free it + request, node_session[node] = cluster.packrequest(addr, session , msg, sz) + skynet.ret(c:request(request, session)) +end + +local request_fd = {} + +function command.socket(source, subcmd, fd, msg) + if subcmd == "data" then + local addr, session, msg = cluster.unpackrequest(msg) + local msg, sz = skynet.rawcall(addr, "lua", msg) + local response = cluster.packresponse(session, msg, sz) + socket.write(fd, response) + elseif subcmd == "open" then + skynet.error(string.format("socket accept from %s", msg)) + skynet.call(source, "lua", "accept", fd) + else + skynet.error(string.format("socket %s %d : %s", subcmd, fd, msg)) + end +end + +skynet.start(function() + skynet.dispatch("lua", function(_, source, cmd, ...) + local f = assert(command[cmd]) + f(source, ...) + end) +end) diff --git a/skynet-src/skynet_daemon.c b/skynet-src/skynet_daemon.c new file mode 100644 index 00000000..4c5e3882 --- /dev/null +++ b/skynet-src/skynet_daemon.c @@ -0,0 +1,96 @@ +#include +#include +#include +#include +#include +#include +#include + +#include "skynet_daemon.h" + +static int +check_pid(const char *pidfile) { + int pid = 0; + FILE *f = fopen(pidfile,"r"); + if (f == NULL) + return 0; + int n = fscanf(f,"%d", &pid); + fclose(f); + + if (n !=1 || pid == 0 || pid == getpid()) { + return 0; + } + + if (kill(pid, 0) && errno == ESRCH) + return 0; + + return pid; +} + +static int +write_pid(const char *pidfile) { + FILE *f; + int pid = 0; + int fd = open(pidfile, O_RDWR|O_CREAT, 0644); + if (fd == -1) { + fprintf(stderr, "Can't create %s.\n", pidfile); + return 0; + } + f = fdopen(fd, "r+"); + if (f == NULL) { + fprintf(stderr, "Can't open %s.\n", pidfile); + return 0; + } + + if (flock(fd, LOCK_EX|LOCK_NB) == -1) { + int n = fscanf(f, "%d", &pid); + fclose(f); + if (n != 1) { + fprintf(stderr, "Can't lock and read pidfile.\n"); + } else { + fprintf(stderr, "Can't lock pidfile, lock is held by pid %d.\n", pid); + } + return 0; + } + + pid = getpid(); + if (!fprintf(f,"%d\n", pid)) { + fprintf(stderr, "Can't write pid.\n"); + close(fd); + return 0; + } + fflush(f); + + return pid; +} + +int +daemon_init(const char *pidfile) { + int pid = check_pid(pidfile); + + if (pid) { + fprintf(stderr, "Skynet is already running, pid = %d.\n", pid); + return 1; + } + +#ifdef __APPLE__ + fprintf(stderr, "'daemon' is deprecated: first deprecated in OS X 10.5 , use launchd instead.\n"); +#else + if (daemon(1,0)) { + fprintf(stderr, "Can't daemonize.\n"); + return 1; + } +#endif + + pid = write_pid(pidfile); + if (pid == 0) { + return 1; + } + + return 0; +} + +int +daemon_exit(const char *pidfile) { + return unlink(pidfile); +} diff --git a/skynet-src/skynet_daemon.h b/skynet-src/skynet_daemon.h new file mode 100644 index 00000000..0d99562a --- /dev/null +++ b/skynet-src/skynet_daemon.h @@ -0,0 +1,7 @@ +#ifndef skynet_daemon_h +#define skynet_daemon_h + +int daemon_init(const char *pidfile); +int daemon_exit(const char *pidfile); + +#endif diff --git a/skynet-src/skynet_harbor.c b/skynet-src/skynet_harbor.c index ae47cf45..160b1994 100644 --- a/skynet-src/skynet_harbor.c +++ b/skynet-src/skynet_harbor.c @@ -7,18 +7,20 @@ #include static struct skynet_context * REMOTE = 0; -static unsigned int HARBOR = 0; +static unsigned int HARBOR = ~0; void skynet_harbor_send(struct remote_message *rmsg, uint32_t source, int session) { int type = rmsg->sz >> HANDLE_REMOTE_SHIFT; rmsg->sz &= HANDLE_MASK; - assert(type != PTYPE_SYSTEM && type != PTYPE_HARBOR); + assert(type != PTYPE_SYSTEM && type != PTYPE_HARBOR && REMOTE); skynet_context_send(REMOTE, rmsg, sizeof(*rmsg) , source, type , session); } void skynet_harbor_register(struct remote_name *rname) { + if (REMOTE == NULL) + return; int i; int number = 1; for (i=0;iin_global); - skynet_globalmq_push(queue); -} - -void -skynet_mq_pushglobal(struct message_queue *queue) { - LOCK(queue) - assert(queue->in_global); - skynet_globalmq_push(queue); - queue->in_global = MQ_IN_GLOBAL; - UNLOCK(queue) -} - void skynet_mq_mark_release(struct message_queue *q) { LOCK(q) @@ -261,7 +246,7 @@ skynet_mq_release(struct message_queue *q, message_drop drop_func, void *ud) { UNLOCK(q) _drop_queue(q, drop_func, ud); } else { - skynet_mq_force_push(q); + skynet_globalmq_push(q); UNLOCK(q) } } diff --git a/skynet-src/skynet_mq.h b/skynet-src/skynet_mq.h index 5ac582a1..df1a9998 100644 --- a/skynet-src/skynet_mq.h +++ b/skynet-src/skynet_mq.h @@ -13,6 +13,7 @@ struct skynet_message { struct message_queue; +void skynet_globalmq_push(struct message_queue * queue); struct message_queue * skynet_globalmq_pop(void); struct message_queue * skynet_mq_create(uint32_t handle); @@ -30,9 +31,6 @@ void skynet_mq_push(struct message_queue *q, struct skynet_message *message); // return the length of message queue, for debug int skynet_mq_length(struct message_queue *q); -void skynet_mq_force_push(struct message_queue *q); -void skynet_mq_pushglobal(struct message_queue *q); - void skynet_mq_init(); #endif diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index 4bc7b66a..ab7094c8 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -139,7 +139,7 @@ skynet_context_new(const char * name, const char *param) { if (ret) { ctx->init = true; } - skynet_mq_force_push(queue); + skynet_globalmq_push(queue); if (ret) { skynet_error(ret, "LAUNCH %s %s", name, param ? param : ""); } @@ -228,11 +228,13 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) { CHECKCALLING_END(ctx) } -int -skynet_context_message_dispatch(struct skynet_monitor *sm) { - struct message_queue * q = skynet_globalmq_pop(); - if (q==NULL) - return 1; +struct message_queue * +skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q) { + if (q == NULL) { + q = skynet_globalmq_pop(); + if (q==NULL) + return NULL; + } uint32_t handle = skynet_mq_handle(q); @@ -240,13 +242,13 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) { if (ctx == NULL) { struct drop_t d = { handle }; skynet_mq_release(q, drop_message, &d); - return 0; + return skynet_globalmq_pop(); } struct skynet_message msg; if (skynet_mq_pop(q,&msg)) { skynet_context_release(ctx); - return 0; + return skynet_globalmq_pop(); } skynet_monitor_trigger(sm, msg.source , handle); @@ -258,12 +260,18 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) { } assert(q == ctx->queue); - skynet_mq_pushglobal(q); + struct message_queue *nq = skynet_globalmq_pop(); + if (nq) { + // If global mq is not empty , push q back, and return next queue (nq) + // Else (global mq is empty or block, don't push q back, and return q again (for next dispatch) + skynet_globalmq_push(q); + q = nq; + } skynet_context_release(ctx); skynet_monitor_trigger(sm, 0,0); - return 0; + return q; } static void diff --git a/skynet-src/skynet_server.h b/skynet-src/skynet_server.h index 7adc1292..51b8d1b2 100644 --- a/skynet-src/skynet_server.h +++ b/skynet-src/skynet_server.h @@ -16,7 +16,7 @@ void skynet_context_init(struct skynet_context *, uint32_t handle); int skynet_context_push(uint32_t handle, struct skynet_message *message); void skynet_context_send(struct skynet_context * context, void * msg, size_t sz, uint32_t source, int type, int session); int skynet_context_newsession(struct skynet_context *); -int skynet_context_message_dispatch(struct skynet_monitor *); // return 1 when block +struct message_queue * skynet_context_message_dispatch(struct skynet_monitor *, struct message_queue *); // return next queue int skynet_context_total(); void skynet_context_endless(uint32_t handle); // for monitor diff --git a/skynet-src/skynet_start.c b/skynet-src/skynet_start.c index d1f80e10..d45ded49 100644 --- a/skynet-src/skynet_start.c +++ b/skynet-src/skynet_start.c @@ -7,6 +7,7 @@ #include "skynet_timer.h" #include "skynet_monitor.h" #include "skynet_socket.h" +#include "skynet_daemon.h" #include #include @@ -120,8 +121,10 @@ _worker(void *p) { struct monitor *m = wp->m; struct skynet_monitor *sm = m->m[id]; skynet_initthread(THREAD_WORKER); + struct message_queue * q = NULL; for (;;) { - if (skynet_context_message_dispatch(sm)) { + q = skynet_context_message_dispatch(sm, q); + if (q == NULL) { CHECK_ABORT if (pthread_mutex_lock(&m->mutex) == 0) { ++ m->sleep; @@ -195,6 +198,11 @@ bootstrap(const char * cmdline) { void skynet_start(struct skynet_config * config) { + if (config->daemon) { + if (daemon_init(config->daemon)) { + exit(1); + } + } skynet_harbor_init(config->harbor); skynet_handle_init(config->harbor); skynet_mq_init(); @@ -206,4 +214,7 @@ skynet_start(struct skynet_config * config) { _start(config->thread); skynet_socket_free(); + if (config->daemon) { + daemon_exit(config->daemon); + } } diff --git a/skynet-src/skynet_timer.c b/skynet-src/skynet_timer.c index 489f8d09..32b71f77 100644 --- a/skynet-src/skynet_timer.c +++ b/skynet-src/skynet_timer.c @@ -16,6 +16,9 @@ typedef void (*timer_execute_func)(void *ud,void *arg); +#define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {} +#define UNLOCK(q) __sync_lock_release(&(q)->lock); + #define TIME_NEAR_SHIFT 8 #define TIME_NEAR (1 << TIME_NEAR_SHIFT) #define TIME_LEVEL_SHIFT 6 @@ -50,8 +53,7 @@ struct timer { static struct timer * TI = NULL; static inline struct timer_node * -link_clear(struct link_list *list) -{ +link_clear(struct link_list *list) { struct timer_node * ret = list->head.next; list->head.next = 0; list->tail = &(list->head); @@ -60,23 +62,20 @@ link_clear(struct link_list *list) } static inline void -link(struct link_list *list,struct timer_node *node) -{ +link(struct link_list *list,struct timer_node *node) { list->tail->next = node; list->tail = node; node->next=0; } static void -add_node(struct timer *T,struct timer_node *node) -{ +add_node(struct timer *T,struct timer_node *node) { int time=node->expire; int current_time=T->time; if ((time|TIME_NEAR_MASK)==(current_time|TIME_NEAR_MASK)) { link(&T->near[time&TIME_NEAR_MASK],node); - } - else { + } else { int i; int mask=TIME_NEAR << TIME_LEVEL_SHIFT; for (i=0;i<3;i++) { @@ -90,21 +89,21 @@ add_node(struct timer *T,struct timer_node *node) } static void -timer_add(struct timer *T,void *arg,size_t sz,int time) -{ +timer_add(struct timer *T,void *arg,size_t sz,int time) { struct timer_node *node = (struct timer_node *)skynet_malloc(sizeof(*node)+sz); memcpy(node+1,arg,sz); - while (__sync_lock_test_and_set(&T->lock,1)) {}; + LOCK(T); node->expire=time+T->time; add_node(T,node); - __sync_lock_release(&T->lock); + UNLOCK(T); } static void timer_shift(struct timer *T) { + LOCK(T); int mask = TIME_NEAR; int time = (++T->time) >> TIME_NEAR_SHIFT; int i=0; @@ -125,50 +124,57 @@ timer_shift(struct timer *T) { time >>= TIME_LEVEL_SHIFT; ++i; } + UNLOCK(T); +} + +static inline void +dispatch_list(struct timer_node *current) { + do { + struct timer_event * event = (struct timer_event *)(current+1); + struct skynet_message message; + message.source = 0; + message.session = event->session; + message.data = NULL; + message.sz = PTYPE_RESPONSE << HANDLE_REMOTE_SHIFT; + + skynet_context_push(event->handle, &message); + + struct timer_node * temp = current; + current=current->next; + skynet_free(temp); + } while (current); } static inline void timer_execute(struct timer *T) { + LOCK(T); int idx = T->time & TIME_NEAR_MASK; while (T->near[idx].head.next) { struct timer_node *current = link_clear(&T->near[idx]); - - do { - struct timer_event * event = (struct timer_event *)(current+1); - struct skynet_message message; - message.source = 0; - message.session = event->session; - message.data = NULL; - message.sz = PTYPE_RESPONSE << HANDLE_REMOTE_SHIFT; - - skynet_context_push(event->handle, &message); - - struct timer_node * temp = current; - current=current->next; - skynet_free(temp); - } while (current); + UNLOCK(T); + // dispatch_list don't need lock T + dispatch_list(current); + LOCK(T); } + + UNLOCK(T); } static void -timer_update(struct timer *T) -{ - while (__sync_lock_test_and_set(&T->lock,1)) {}; - +timer_update(struct timer *T) { // try to dispatch timeout 0 (rare condition) timer_execute(T); // shift time first, and then dispatch timer message timer_shift(T); + timer_execute(T); - __sync_lock_release(&T->lock); } static struct timer * -timer_create_timer() -{ +timer_create_timer() { struct timer *r=(struct timer *)skynet_malloc(sizeof(struct timer)); memset(r,0,sizeof(*r));