lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-24 20:23:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

clustersender tracetag bug

Browse Source
This commit is contained in:
hong
2019-05-06 21:57:10 +08:00
committed by 云风
parent 431feb6f16
commit db61ec376a
2 changed files with 4 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
service/clusterd.lua
View File

@@ -1,23 +1,14 @@
local skynet = require "skynet" local skynet = require "skynet"
require "skynet.manager" require "skynet.manager"
local sc = require "skynet.socketchannel"
local socket = require "skynet.socket"
local cluster = require "skynet.cluster.core" local cluster = require "skynet.cluster.core"
local config_name = skynet.getenv "cluster" local config_name = skynet.getenv "cluster"
local node_address = {} local node_address = {}
local node_session = {}
local node_sender = {} local node_sender = {}
local command = {} local command = {}
local config = {} local config = {}
local nodename = cluster.nodename() local nodename = cluster.nodename()
local function read_response(sock)
local sz = socket.header(sock:read(2))
local msg = sock:read(sz)
return cluster.unpackresponse(msg) -- session, ok, data, padding
end
local connecting = {} local connecting = {}
local function open_channel(t, key) local function open_channel(t, key)
@@ -44,7 +35,7 @@ local function open_channel(t, key)
local host, port = string.match(address, "([^:]+):(.*)$") local host, port = string.match(address, "([^:]+):(.*)$")
c = node_sender[key] c = node_sender[key]
if c == nil then if c == nil then
c = skynet.newservice "clustersender" c = skynet.newservice("clustersender", key, nodename)
if node_sender[key] then if node_sender[key] then
-- double check -- double check
skynet.kill(c) skynet.kill(c)

10
service/clustersender.lua
View File

@@ -2,21 +2,21 @@ local skynet = require "skynet"
local sc = require "skynet.socketchannel" local sc = require "skynet.socketchannel"
local socket = require "skynet.socket" local socket = require "skynet.socket"
local cluster = require "skynet.cluster.core" local cluster = require "skynet.cluster.core"
local ignoreret = skynet.ignoreret
local channel local channel
local session = 1 local session = 1
local node, nodename = ...
local command = {} local command = {}
local waiting = {} local waiting = {}
local function send_request(addr, msg, sz) local function send_request(addr, msg, sz)
-- msg is a local pointer, cluster.packrequest will free it -- msg is a local pointer, cluster.packrequest will free it
local current_session = session local current_session = session
local request, new_session, padding = cluster.packrequest(addr, session, msg, sz) local request, new_session, padding = cluster.packrequest(addr, session, msg, sz)
session = new_session session = new_session
-- node_channel[node] may yield or throw error
local tracetag = skynet.tracetag() local tracetag = skynet.tracetag()
if tracetag then if tracetag then
if tracetag:sub(1,1) ~= "(" then if tracetag:sub(1,1) ~= "(" then
@@ -63,7 +63,6 @@ function command.push(addr, msg, sz)
session = new_session session = new_session
end end
-- node_channel[node] may yield or throw error
channel:request(request, nil, padding) channel:request(request, nil, padding)
end end
@@ -80,7 +79,7 @@ function command.changenode(host, port)
response = read_response, response = read_response,
nodelay = true, nodelay = true,
} }
succ, err = pcall(c.connect, c, true) local succ, err = pcall(c.connect, c, true)
if channel then if channel then
channel:close() channel:close()
end end
@@ -103,6 +102,3 @@ skynet.start(function()
f(...) f(...)
end) end)
end) end)