From 56f30e7f7fbc6ebc598afde4d48f899578a2c272 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Mon, 6 Aug 2012 19:58:35 +0800 Subject: [PATCH] service client and a simple DB example --- Makefile | 7 +++++-- agent.lua | 17 +++++++++++++---- client.c | 47 +++++++++++++++++++++++++++++++++++++++++++++++ gate/main.c | 14 ++++++++++---- globallog.lua | 2 +- main.lua | 2 ++ main_log.lua | 3 --- service_client.c | 32 ++++++++++++++++++++++++++++++++ simpledb.lua | 7 ++++++- skynet.lua | 6 +++++- skynet_server.c | 26 +++++++++++++++++++++++--- watchdog.lua | 15 +++++++++++++-- 12 files changed, 157 insertions(+), 21 deletions(-) create mode 100644 service_client.c diff --git a/Makefile b/Makefile index aa91dcdb..a06f46eb 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -all : skynet snlua.so logger.so skynet.so gate.so client skynet-master +all : skynet snlua.so logger.so skynet.so gate.so client.so client skynet-master skynet : skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c skynet_server.c skynet_start.c skynet_timer.c skynet_error.c skynet_harbor.c gcc -Wall -g -Wl,-E -o $@ $^ -lpthread -ldl -lrt -Wl,-E -llua -lm -lzmq @@ -15,8 +15,11 @@ gate.so : gate/mread.c gate/ringbuffer.c gate/map.c gate/main.c skynet.so : lua-skynet.c gcc -Wall -g -fPIC --shared $^ -o $@ +client.so : service_client.c + gcc -Wall -g -fPIC --shared $^ -o $@ + client : client.c - gcc -Wall -g $^ -o $@ + gcc -Wall -g $^ -o $@ -lpthread skynet-master : master/master.c gcc -g -Wall -o $@ $^ -lzmq diff --git a/agent.lua b/agent.lua index 386a06ef..6c1734b6 100644 --- a/agent.lua +++ b/agent.lua @@ -1,7 +1,16 @@ local skynet = require "skynet" +local client = ... -print("agent",...) - -skynet.dispatch(function(msg , addr) - print("[agent]",addr,msg) +skynet.dispatch(function(msg,session) + if session == 0 then + print("client command",msg) + local result = skynet.call("SIMPLEDB",msg) + skynet.send(client,0,result) + else + print("server command",msg) + if msg == "CLOSE" then + skynet.kill(client) + skynet.exit() + end + end end) diff --git a/client.c b/client.c index 1ab9108c..759fd78c 100644 --- a/client.c +++ b/client.c @@ -1,3 +1,4 @@ +#include #include #include #include @@ -7,6 +8,46 @@ #include #include #include +#include + +struct args { + int fd; +}; + +static void +readall(int fd, void * buffer, size_t sz) { + for (;;) { + int err = recv(fd , buffer, sz, MSG_WAITALL); + if (err < 0) { + if (errno == EAGAIN || errno == EINTR) + continue; + break; + } + return; + } + perror("Socket error"); + exit(1); +} + +static void * +_read(void *ud) { + struct args *p = ud; + int fd = p->fd; + fflush(stdout); + for (;;) { + uint8_t header[2]; + fflush(stdout); + readall(fd, header, 2); + size_t len = header[0] | header[1] << 8; + if (len>0) { + char tmp[len+1]; + readall(fd, tmp, len); + tmp[len]='\0'; + printf("%s\n",tmp); + } + } + return NULL; +} static void test(int fd) { @@ -50,8 +91,14 @@ main(int argc, char * argv[]) { return 1; } + struct args arg = { fd }; + pthread_t pid ; + pthread_create(&pid, NULL, _read, &arg); + test(fd); + pthread_join(pid, NULL); + close(fd); return 0; diff --git a/gate/main.c b/gate/main.c index 9a75cdc9..03c7e642 100644 --- a/gate/main.c +++ b/gate/main.c @@ -118,10 +118,16 @@ _report(struct skynet_context * ctx, const char * data, ...) { static void _forward(struct skynet_context * ctx,struct gate *g, int uid, void * data, size_t len) { struct connection * agent = _id_to_agent(g,uid); - char * tmp = malloc(len + 32); - int n = snprintf(tmp,len+32,"%d data ",uid); - memcpy(tmp+n,data,len); - skynet_send(ctx, agent->agent ? agent->agent : WATCHDOG, 0, tmp, len + n); + if (agent->agent) { + char * tmp = malloc(len); + memcpy(tmp,data,len); + skynet_send(ctx, agent->agent, 0, tmp, len); + } else { + char * tmp = malloc(len + 32); + int n = snprintf(tmp,len+32,"%d data ",uid); + memcpy(tmp+n,data,len); + skynet_send(ctx, WATCHDOG, 0, tmp, len + n); + } } static int diff --git a/globallog.lua b/globallog.lua index e2f4bfbb..f6379996 100644 --- a/globallog.lua +++ b/globallog.lua @@ -1,6 +1,6 @@ local skynet = require "skynet" -skynet.dispatch(function(message, from, session) +skynet.dispatch(function(message, session , from) print("[GLOBALLOG]",session, from,message) end) diff --git a/main.lua b/main.lua index 5e026338..5d633b06 100644 --- a/main.lua +++ b/main.lua @@ -8,5 +8,7 @@ local watchdog = skynet.launch("snlua","watchdog.lua") print("watchdog",watchdog) local gate = skynet.launch("gate","8888 4 0") print("gate",gate) +local db = skynet.launch("snlua","simpledb.lua") +print("simpledb",db) skynet.exit() diff --git a/main_log.lua b/main_log.lua index 55e987d3..9e613c51 100644 --- a/main_log.lua +++ b/main_log.lua @@ -5,7 +5,4 @@ print("Log server start") local log = skynet.launch("snlua","globallog.lua") print("log",log) -local db = skynet.launch("snlua","simpledb.lua") -print("simpledb",db) - skynet.exit() diff --git a/service_client.c b/service_client.c new file mode 100644 index 00000000..fec975c7 --- /dev/null +++ b/service_client.c @@ -0,0 +1,32 @@ +#include "skynet.h" + +#include +#include +#include +#include +#include +#include + +static void +_cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { + assert(session == 0); + assert(sz <= 65535); + int fd = (int)(intptr_t)ud; + + struct iovec buffer[2]; + uint8_t head[2] = { sz & 0xff , sz >> 8 & 0xff }; + buffer[0].iov_base = head; + buffer[0].iov_len = 2; + buffer[1].iov_base = (void *)msg; + buffer[1].iov_len = sz; + + writev(fd, buffer, 2); +} + +int +client_init(void * dummy, struct skynet_context *ctx, const char * args) { + int fd = strtol(args, NULL, 10); + skynet_callback(ctx, (void*)(intptr_t)fd, _cb); + + return 0; +} diff --git a/simpledb.lua b/simpledb.lua index 0ca9fb39..3568990b 100644 --- a/simpledb.lua +++ b/simpledb.lua @@ -16,7 +16,12 @@ end skynet.dispatch(function(message, from, session) print("simpledb",message, from, session) local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)") - command[cmd](key,value) + local f = command[cmd] + if f then + f(key,value) + else + skynet.ret("Invalid command : "..message) + end end) skynet.register "SIMPLEDB" diff --git a/skynet.lua b/skynet.lua index e2d4d838..bdb803c0 100644 --- a/skynet.lua +++ b/skynet.lua @@ -65,6 +65,10 @@ function skynet.exit() c.command("EXIT") end +function skynet.kill(name) + c.command("KILL",name) +end + skynet.send = c.send function skynet.call(addr, message) @@ -83,7 +87,7 @@ function skynet.dispatch(f) co = coroutine.create(f) session_coroutine_id[co] = session session_coroutine_address[co] = address - suspend(co, coroutine.resume(co, message, address, session)) + suspend(co, coroutine.resume(co, message, session, address)) else session_id_coroutine[session] = nil suspend(co, coroutine.resume(co, message)) diff --git a/skynet_server.c b/skynet_server.c index ff112e68..a1b4447d 100644 --- a/skynet_server.c +++ b/skynet_server.c @@ -128,17 +128,20 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) { } } -static void +static int _drop_queue(struct message_queue *q) { // todo: send message back to message source struct skynet_message msg; + int s = 0; while(!skynet_mq_pop(q, &msg)) { + ++s; if (skynet_harbor_message_isremote(msg.source)) { skynet_harbor_message_close(&msg); } free(msg.data); } skynet_mq_release(q); + return s; } int @@ -151,8 +154,10 @@ skynet_context_message_dispatch(void) { struct skynet_context * ctx = skynet_handle_grab(handle); if (ctx == NULL) { - skynet_error(NULL, "Drop message queue %x ", handle); - _drop_queue(q); + int s = _drop_queue(q); + if (s>0) { + skynet_error(NULL, "Drop message queue %x (%d messages)", handle,s); + } return 0; } @@ -226,6 +231,21 @@ skynet_command(struct skynet_context * context, const char * cmd , int session, return NULL; } + if (strcmp(cmd,"KILL") == 0) { + uint32_t handle = 0; + if (parm[0] == ':') { + handle = strtoul(parm+1, NULL, 16); + } else if (parm[0] == '.') { + handle = skynet_handle_findname(parm+1); + } else { + // todo : kill global service + } + if (handle) { + skynet_handle_retire(handle); + } + return NULL; + } + if (strcmp(cmd,"LAUNCH") == 0) { size_t sz = strlen(parm); char tmp[sz+1]; diff --git a/watchdog.lua b/watchdog.lua index e45e5b3e..937e8aca 100644 --- a/watchdog.lua +++ b/watchdog.lua @@ -1,23 +1,34 @@ local skynet = require "skynet" local command = {} +local agent_all = {} function command:open(parm) local fd,addr = string.match(parm,"(%d+) ([^%s]+)") fd = tonumber(fd) skynet.send("LOG", 0, string.format("%d %d %s",self,fd,addr)) - local agent = skynet.launch("snlua","agent.lua",self) + local client = skynet.launch("client",fd) + print("client",client) + local agent = skynet.launch("snlua","agent.lua",client) if agent then + agent_all[self] = agent skynet.send("gate",0, "forward ".. self .. " " .. agent) end end function command:close() skynet.send("LOG",0,string.format("close %d",self)) + skynet.send(agent_all[self],1,"CLOSE") + agent_all[self] = nil end function command:data(data) - skynet.send("LOG",0,string.format("data %d size=%d",self,#data)) + local agent = agent_all[self] + if agent then + skynet.send(agent,0,data) + else + skynet.send("LOG",0,string.format("data %d size=%d",self,#data)) + end end skynet.dispatch(function(message)