From 1e9a27232ba43da10d3a1e1d59e5ee14b052e2d5 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Mon, 14 Apr 2014 14:21:31 +0800 Subject: [PATCH] rewrite watchdog/agent, use new gate --- examples/agent.lua | 21 ++++++- examples/main.lua | 9 ++- examples/watchdog.lua | 73 +++++++++++------------- service-src/service_client.c | 1 - service/gate.lua | 105 +++++++++++++++++++++++++++++++---- 5 files changed, 151 insertions(+), 58 deletions(-) diff --git a/examples/agent.lua b/examples/agent.lua index 2416da5e..69962315 100644 --- a/examples/agent.lua +++ b/examples/agent.lua @@ -1,5 +1,7 @@ local skynet = require "skynet" -local client = ... +local client + +local CMD = {} skynet.register_protocol { name = "client", @@ -8,12 +10,25 @@ skynet.register_protocol { unpack = skynet.tostring, dispatch = function (session, address, text) -- It's client, there is no session - -- skynet.send("LOG", "text", "client message :" .. text) local result = skynet.call("SIMPLEDB", "text", text) skynet.ret(result) end } -skynet.start(function() +function CMD.start(gate , fd) + client = skynet.launch("client", fd) + skynet.call(gate, "lua", "forward", fd, client) skynet.send(client,"text","Welcome to skynet") +end + +function CMD.exit() + skynet.kill(client) + client = nil +end + +skynet.start(function() + skynet.dispatch("lua", function(_,_, command, ...) + local f = CMD[command] + skynet.ret(skynet.pack(f(...))) + end) end) diff --git a/examples/main.lua b/examples/main.lua index 1e935af6..e95f05c2 100644 --- a/examples/main.lua +++ b/examples/main.lua @@ -9,9 +9,12 @@ skynet.start(function() local lualog = skynet.newservice("lualog") local console = skynet.newservice("console") -- skynet.newservice("debug_console",8000) - local watchdog = skynet.newservice("watchdog","8888", max_client, 0) - local db = skynet.newservice("simpledb") --- skynet.launch("snlua","testgroup") + skynet.newservice("simpledb") + local watchdog = skynet.newservice("watchdog") + skynet.call(watchdog, "lua", "start", { + port = 8888, + maxclient = max_client, + }) skynet.exit() end) diff --git a/examples/watchdog.lua b/examples/watchdog.lua index c15f3299..a6d5339a 100644 --- a/examples/watchdog.lua +++ b/examples/watchdog.lua @@ -1,58 +1,49 @@ local skynet = require "skynet" -local port, max_agent, buffer = ... -local command = {} -local agent_all = {} +local CMD = {} +local SOCKET = {} local gate +local agent = {} -function command:open(parm) - local fd,addr = string.match(parm,"(%d+) ([^%s]+)") - fd = tonumber(fd) - print("agent open",self,string.format("%d %d %s",self,fd,addr)) - local client = skynet.launch("client",fd) - local agent = skynet.launch("snlua","agent",skynet.address(client)) - if agent then - agent_all[self] = { agent , client } - skynet.send(gate, "text", "forward" , self, skynet.address(agent) , skynet.address(client)) +function SOCKET.open(fd, addr) + agent[fd] = skynet.newservice("agent") + skynet.call(agent[fd], "lua", "start", gate, fd) +end + +local function close_agent(fd) + local a = agent[fd] + if a then + skynet.call(a, "lua", "exit") + skynet.kill(a) + agent[fd] = nil end end -function command:close() - print("agent close",self,string.format("close %d",self)) - local agent = agent_all[self] - agent_all[self] = nil - skynet.kill(agent[1]) - skynet.kill(agent[2]) +function SOCKET.close(fd) + print("socket close",fd) + close_agent(fd) end -function command:data(data, session) - local agent = agent_all[self] - if agent then - -- PTYPE_CLIENT = 3 , read skynet.h - skynet.redirect(agent[1], agent[2], "client", 0, data) - else - skynet.error(string.format("agent data drop %d size=%d",self,#data)) - end +function SOCKET.error(fd, msg) + print("socket error",fd, msg) + close_agent(fd) end -skynet.register_protocol { - name = "client", - id = skynet.PTYPE_CLIENT, -} +function CMD.start(conf) + skynet.call(gate, "lua", "open" , conf) +end skynet.start(function() - skynet.dispatch("text", function(session, from, message) - local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)") - id = tonumber(id) - local f = command[cmd] - if f then - f(id,parm,session) + skynet.dispatch("lua", function(session, source, cmd, subcmd, ...) + if cmd == "socket" then + local f = SOCKET[subcmd] + f(...) + -- socket api don't need return else - error(string.format("[watchdog] Unknown command : %s",message)) + local f = assert(CMD[cmd]) + skynet.ret(skynet.pack(f(subcmd, ...))) end end) - -- 0 for default client tag - gate = skynet.launch("gate" , "S" , skynet.address(skynet.self()), port, 0, max_agent, buffer) - skynet.send(gate,"text", "start") - skynet.register(".watchdog") + + gate = skynet.newservice("gate") end) diff --git a/service-src/service_client.c b/service-src/service_client.c index e31f9705..754cf962 100644 --- a/service-src/service_client.c +++ b/service-src/service_client.c @@ -22,7 +22,6 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t if (type == PTYPE_RESPONSE) { // todo: response to client with session, session may be packed into package } else { - printf("client %d\n",type); assert(type == PTYPE_TEXT); // todo: support other protocol } diff --git a/service/gate.lua b/service/gate.lua index 61871a7e..86d00209 100644 --- a/service/gate.lua +++ b/service/gate.lua @@ -4,13 +4,21 @@ local socketdriver = require "socketdriver" local socket local queue +local watchdog +local maxclient +local client_number = 0 local CMD = setmetatable({}, { __gc = function() netpack.clear(queue) end }) +local forwarding_address = {} +local forwarding_client = {} +local forwarding_map = {} +local openning = {} -function CMD.open( conf ) +function CMD.open( source , conf ) assert(not socket) local address = conf.address or "0.0.0.0" local port = assert(conf.port) - local maxclient = conf.maxclient or 1024 + maxclient = conf.maxclient or 1024 + watchdog = conf.watchdog or source socket = socketdriver.listen(address, port) socketdriver.start(socket) end @@ -21,29 +29,101 @@ function CMD.close() socket = nil end +local function unforward_fd(fd) + local address = forwarding_address[fd] + if address then + local client = forwarding_client[fd] + local fm = forwarding_map[address] + if type(fm) == "table" then + fm[address] = nil + if next(fm) == nil then + forwarding_map[address] = nil + end + else + forwarding_map[address] = nil + end + forwarding_address[fd] = nil + forwarding_client[fd] = nil + return address, client + end +end + + +function CMD.forward(source, fd, client, address) + local addr = address or source + if not unforward_fd(fd) then + socketdriver.start(fd) + end + + forwarding_address[fd] = addr + forwarding_client[fd] = client or 0 + local fm = forwarding_map[addr] + if fm then + if type(fm) == "table" then + fm[addr] = true + else + fm[addr] = { [fm] = true, [addr] = true } + end + else + forwarding_map[addr] = fd + end +end + +function CMD.kick(source, fd) + if not fd then + fd = forwarding_map[source] + if type(fd) == "table" then + for k in pairs(fd) do + socketdriver.close(k) + end + return + end + end + socketdriver.close(fd) +end + local MSG = {} function MSG.data(fd, msg, sz) - print("Data:", fd, netpack.tostring(msg, sz)) + local address = forwarding_address[fd] + local client = forwarding_client[fd] + skynet.redirect(address, client, "client", 0, msg, sz) end function MSG.more() for fd, msg, sz in netpack.pop, queue do - print("More:", fd, netpack.tostring(msg, sz)) + local address = forwarding_address[fd] + local client = forwarding_client[fd] + skynet.redirect(address, client, "client", 0, msg, sz) end end function MSG.open(fd, msg) - socketdriver.start(fd) - print("Open:", fd, msg) + if client_number >= maxclient then + socketdriver.close(fd) + return + end + openning[fd] = true + client_number = client_number + 1 + skynet.send(watchdog, "lua", "socket", "open", fd, msg) +end + +local function close_fd(fd) + local address, client = unforward_fd(fd) + if openning[fd] then + openning[fd] = nil + client_number = client_number - 1 + end end function MSG.close(fd) - print("Close:", fd) + close_fd(fd) + skynet.send(watchdog, "lua", "socket", "close", fd) end function MSG.error(fd, msg) - print("Error:", fd, msg) + close_fd(fd) + skynet.send(watchdog, "lua", "socket", "error", fd) end skynet.register_protocol { @@ -60,9 +140,14 @@ skynet.register_protocol { end } +skynet.register_protocol { + name = "client", + id = skynet.PTYPE_CLIENT, +} + skynet.start(function() - skynet.dispatch("lua", function (_,_, cmd, ...) + skynet.dispatch("lua", function (_, address, cmd, ...) local f = assert(CMD[cmd]) - skynet.ret(skynet.pack(f(...))) + skynet.ret(skynet.pack(f(address, ...))) end) end)