GitHub/cc-scada
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

#225 network changes for supervisor sessions

Browse Source
This commit is contained in:
Mikayla Fischler
2023-06-05 17:24:00 -04:00
parent 9a5fc92c86
commit 360609df1f
7 changed files with 233 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

206
supervisor/session/svsessions.lua
View File

@@ -27,7 +27,7 @@ local svsessions = {}
local SESSION_TYPE = {
RTU_SESSION = 0, -- RTU gateway
PLC_SESSION = 1, -- reactor PLC
COORD_SESSION = 2, -- coordinator
CRD_SESSION = 2, -- coordinator
PDG_SESSION = 3 -- pocket diagnostics
}
@@ -38,11 +38,11 @@ local self = {
fp_ok = false,
num_reactors = 0,
facility = nil, ---@type facility|nil
sessions = { rtu = {}, plc = {}, coord = {}, pdg = {} },
next_ids = { rtu = 0, plc = 0, coord = 0, pdg = 0 }
sessions = { rtu = {}, plc = {}, crd = {}, pdg = {} },
next_ids = { rtu = 0, plc = 0, crd = 0, pdg = 0 }
}
---@alias sv_session_structs plc_session_struct|rtu_session_struct|coord_session_struct|pdg_session_struct
---@alias sv_session_structs plc_session_struct|rtu_session_struct|crd_session_struct|pdg_session_struct
-- PRIVATE FUNCTIONS --
@@ -60,7 +60,7 @@ local function _sv_handle_outq(session)
if msg ~= nil then
if msg.qtype == mqueue.TYPE.PACKET then
-- handle a packet to be sent
self.modem.transmit(session.r_port, session.l_port, msg.message.raw_sendable())
self.modem.transmit(session.r_chan, config.SVR_CHANNEL, msg.message.raw_sendable())
elseif msg.qtype == mqueue.TYPE.COMMAND then
-- handle instruction/notification
elseif msg.qtype == mqueue.TYPE.DATA then
@@ -81,11 +81,11 @@ local function _sv_handle_outq(session)
elseif cmd.key == SV_Q_DATA.SET_BURN and type(cmd.val) == "table" and #cmd.val == 2 then
plc_s.in_queue.push_data(PLC_S_DATA.BURN_RATE, cmd.val[2])
else
log.debug(util.c("unknown PLC SV queue command ", cmd.key))
log.debug(util.c("[SVS] unknown PLC SV queue command ", cmd.key))
end
end
else
local crd_s = svsessions.get_coord_session()
local crd_s = svsessions.get_crd_session()
if crd_s ~= nil then
if cmd.key == SV_Q_DATA.CRDN_ACK then
-- ack to be sent to coordinator
@@ -104,8 +104,8 @@ local function _sv_handle_outq(session)
-- max 100ms spent processing queue
if util.time() - handle_start > 100 then
log.warning("supervisor out queue handler exceeded 100ms queue process limit")
log.warning(util.c("offending session: port ", session.r_port, " type '", session.s_type, "'"))
log.warning("[SVS] supervisor out queue handler exceeded 100ms queue process limit")
log.warning(util.c("[SVS] offending session: ", session))
break
end
end
@@ -131,15 +131,15 @@ local function _shutdown(session)
session.open = false
session.instance.close()
-- send packets in out queue (namely the close packet)
-- send packets in out queue (for the close packet)
while session.out_queue.ready() do
local msg = session.out_queue.pop()
if msg ~= nil and msg.qtype == mqueue.TYPE.PACKET then
self.modem.transmit(session.r_port, session.l_port, msg.message.raw_sendable())
self.modem.transmit(session.r_chan, config.SVR_CHANNEL, msg.message.raw_sendable())
end
end
log.debug(util.c("closed ", session.s_type, " session ", session.instance.get_id(), " on remote port ", session.r_port))
log.debug(util.c("[SVS] closed session ", session))
end
-- close connections
@@ -160,8 +160,7 @@ local function _check_watchdogs(sessions, timer_event)
if session.open then
local triggered = session.instance.check_wd(timer_event)
if triggered then
log.debug(util.c("watchdog closing ", session.s_type, " session ", session.instance.get_id(),
" on remote port ", session.r_port, "..."))
log.debug(util.c("[SVS] watchdog closing session ", session, "..."))
_shutdown(session)
end
end
@@ -175,21 +174,20 @@ local function _free_closed(sessions)
---@param session sv_session_structs
local on_delete = function (session)
log.debug(util.c("free'ing closed ", session.s_type, " session ", session.instance.get_id(),
" on remote port ", session.r_port))
log.debug(util.c("[SVS] free'ing closed session ", session))
end
util.filter_table(sessions, f, on_delete)
end
-- find a session by remote port
-- find a session by computer ID
---@nodiscard
---@param list table
---@param port integer
---@param s_addr integer
---@return sv_session_structs|nil
local function _find_session(list, port)
local function _find_session(list, s_addr)
for i = 1, #list do
if list[i].r_port == port then return list[i] end
if list[i].s_addr == s_addr then return list[i] end
end
return nil
end
@@ -214,55 +212,55 @@ function svsessions.relink_modem(modem)
self.modem = modem
end
-- find an RTU session by the remote port
-- find an RTU session by the computer ID
---@nodiscard
---@param remote_port integer
---@param source_addr integer
---@return rtu_session_struct|nil
function svsessions.find_rtu_session(remote_port)
function svsessions.find_rtu_session(source_addr)
-- check RTU sessions
local session = _find_session(self.sessions.rtu, remote_port)
local session = _find_session(self.sessions.rtu, source_addr)
---@cast session rtu_session_struct|nil
return session
end
-- find a PLC session by the remote port
-- find a PLC session by the computer ID
---@nodiscard
---@param remote_port integer
---@param source_addr integer
---@return plc_session_struct|nil
function svsessions.find_plc_session(remote_port)
function svsessions.find_plc_session(source_addr)
-- check PLC sessions
local session = _find_session(self.sessions.plc, remote_port)
local session = _find_session(self.sessions.plc, source_addr)
---@cast session plc_session_struct|nil
return session
end
-- find a coordinator session by the remote port
-- find a coordinator session by the computer ID
---@nodiscard
---@param remote_port integer
---@return coord_session_struct|nil
function svsessions.find_coord_session(remote_port)
---@param source_addr integer
---@return crd_session_struct|nil
function svsessions.find_crd_session(source_addr)
-- check coordinator sessions
local session = _find_session(self.sessions.coord, remote_port)
---@cast session coord_session_struct|nil
local session = _find_session(self.sessions.crd, source_addr)
---@cast session crd_session_struct|nil
return session
end
-- find a pocket diagnostics session by the remote port
-- find a pocket diagnostics session by the computer ID
---@nodiscard
---@param remote_port integer
---@param source_addr integer
---@return pdg_session_struct|nil
function svsessions.find_pdg_session(remote_port)
function svsessions.find_pdg_session(source_addr)
-- check diagnostic sessions
local session = _find_session(self.sessions.diag, remote_port)
local session = _find_session(self.sessions.diag, source_addr)
---@cast session pdg_session_struct|nil
return session
end
-- get the a coordinator session if exists
---@nodiscard
---@return coord_session_struct|nil
function svsessions.get_coord_session()
return self.sessions.coord[1]
---@return crd_session_struct|nil
function svsessions.get_crd_session()
return self.sessions.crd[1]
end
-- get a session by reactor ID
@@ -283,12 +281,11 @@ end
-- establish a new PLC session
---@nodiscard
---@param local_port integer
---@param remote_port integer
---@param source_addr integer
---@param for_reactor integer
---@param version string
---@return integer|false session_id
function svsessions.establish_plc_session(local_port, remote_port, for_reactor, version)
function svsessions.establish_plc_session(source_addr, for_reactor, version)
if svsessions.get_reactor_session(for_reactor) == nil and for_reactor >= 1 and for_reactor <= self.num_reactors then
---@class plc_session_struct
local plc_s = {
@@ -296,26 +293,34 @@ function svsessions.establish_plc_session(local_port, remote_port, for_reactor,
open = true,
reactor = for_reactor,
version = version,
l_port = local_port,
r_port = remote_port,
r_chan = config.PLC_CHANNEL,
s_addr = source_addr,
in_queue = mqueue.new(),
out_queue = mqueue.new(),
instance = nil ---@type plc_session
}
plc_s.instance = plc.new_session(self.next_ids.plc, for_reactor, plc_s.in_queue, plc_s.out_queue,
config.PLC_TIMEOUT, self.fp_ok)
local id = self.next_ids.plc
plc_s.instance = plc.new_session(id, source_addr, for_reactor, plc_s.in_queue, plc_s.out_queue,
config.PLC_TIMEOUT, self.fp_ok)
table.insert(self.sessions.plc, plc_s)
local units = self.facility.get_units()
units[for_reactor].link_plc_session(plc_s)
log.debug(util.c("established new PLC session to ", remote_port, " with ID ", self.next_ids.plc,
" for reactor ", for_reactor))
local mt = {
---@param s plc_session_struct
__to_string = function (s) return util.c("PLC [", s.instance.get_id(), "] for reactor #", s.reactor,
" (@", s.s_addr, ")") end
}
databus.tx_plc_connected(for_reactor, version, remote_port)
setmetatable(plc_s, mt)
self.next_ids.plc = self.next_ids.plc + 1
databus.tx_plc_connected(for_reactor, version, source_addr)
log.debug(util.c("[SVS] established new session: ", plc_s))
self.next_ids.plc = id + 1
-- success
return plc_s.instance.get_id()
@@ -327,70 +332,84 @@ end
-- establish a new RTU session
---@nodiscard
---@param local_port integer
---@param remote_port integer
---@param source_addr integer
---@param advertisement table
---@param version string
---@return integer session_id
function svsessions.establish_rtu_session(local_port, remote_port, advertisement, version)
function svsessions.establish_rtu_session(source_addr, advertisement, version)
---@class rtu_session_struct
local rtu_s = {
s_type = "rtu",
open = true,
version = version,
l_port = local_port,
r_port = remote_port,
r_chan = config.RTU_CHANNEL,
s_addr = source_addr,
in_queue = mqueue.new(),
out_queue = mqueue.new(),
instance = nil ---@type rtu_session
}
rtu_s.instance = rtu.new_session(self.next_ids.rtu, rtu_s.in_queue, rtu_s.out_queue, config.RTU_TIMEOUT, advertisement,
self.facility, self.fp_ok)
local id = self.next_ids.rtu
rtu_s.instance = rtu.new_session(id, source_addr, rtu_s.in_queue, rtu_s.out_queue, config.RTU_TIMEOUT,
advertisement, self.facility, self.fp_ok)
table.insert(self.sessions.rtu, rtu_s)
log.debug("established new RTU session to " .. remote_port .. " with ID " .. self.next_ids.rtu)
local mt = {
---@param s rtu_session_struct
__to_string = function (s) return util.c("RTU [", s.instance.get_id(), "] (@", s.s_addr, ")") end
}
databus.tx_rtu_connected(self.next_ids.rtu, version, remote_port)
setmetatable(rtu_s, mt)
self.next_ids.rtu = self.next_ids.rtu + 1
databus.tx_rtu_connected(id, version, source_addr)
log.debug(util.c("[SVS] established new session: ", rtu_s))
self.next_ids.rtu = id + 1
-- success
return rtu_s.instance.get_id()
return id
end
-- establish a new coordinator session
---@nodiscard
---@param local_port integer
---@param remote_port integer
---@param source_addr integer
---@param version string
---@return integer|false session_id
function svsessions.establish_coord_session(local_port, remote_port, version)
if svsessions.get_coord_session() == nil then
---@class coord_session_struct
local coord_s = {
function svsessions.establish_crd_session(source_addr, version)
if svsessions.get_crd_session() == nil then
---@class crd_session_struct
local crd_s = {
s_type = "crd",
open = true,
version = version,
l_port = local_port,
r_port = remote_port,
r_chan = config.CRD_CHANNEL,
s_addr = source_addr,
in_queue = mqueue.new(),
out_queue = mqueue.new(),
instance = nil ---@type coord_session
instance = nil ---@type crd_session
}
coord_s.instance = coordinator.new_session(self.next_ids.coord, coord_s.in_queue, coord_s.out_queue, config.CRD_TIMEOUT,
local id = self.next_ids.crd
crd_s.instance = coordinator.new_session(id, source_addr, crd_s.in_queue, crd_s.out_queue, config.CRD_TIMEOUT,
self.facility, self.fp_ok)
table.insert(self.sessions.coord, coord_s)
table.insert(self.sessions.crd, crd_s)
log.debug("established new coordinator session to " .. remote_port .. " with ID " .. self.next_ids.coord)
local mt = {
---@param s crd_session_struct
__to_string = function (s) return util.c("CRD [", s.instance.get_id(), "] (@", s.s_addr, ")") end
}
databus.tx_crd_connected(version, remote_port)
setmetatable(crd_s, mt)
self.next_ids.coord = self.next_ids.coord + 1
databus.tx_crd_connected(version, source_addr)
log.debug(util.c("[SVS] established new session: ", crd_s))
self.next_ids.crd = id + 1
-- success
return coord_s.instance.get_id()
return id
else
-- we already have a coordinator linked
return false
@@ -399,34 +418,41 @@ end
-- establish a new pocket diagnostics session
---@nodiscard
---@param local_port integer
---@param remote_port integer
---@param source_addr integer
---@param version string
---@return integer|false session_id
function svsessions.establish_pdg_session(local_port, remote_port, version)
function svsessions.establish_pdg_session(source_addr, version)
---@class pdg_session_struct
local pdg_s = {
s_type = "pkt",
open = true,
version = version,
l_port = local_port,
r_port = remote_port,
r_chan = config.PKT_CHANNEL,
s_addr = source_addr,
in_queue = mqueue.new(),
out_queue = mqueue.new(),
instance = nil ---@type pdg_session
}
pdg_s.instance = pocket.new_session(self.next_ids.pdg, pdg_s.in_queue, pdg_s.out_queue, config.PKT_TIMEOUT, self.fp_ok)
local id = self.next_ids.pdg
pdg_s.instance = pocket.new_session(id, source_addr, pdg_s.in_queue, pdg_s.out_queue, config.PKT_TIMEOUT, self.fp_ok)
table.insert(self.sessions.pdg, pdg_s)
log.debug("established new pocket diagnostics session to " .. remote_port .. " with ID " .. self.next_ids.pdg)
local mt = {
---@param s pdg_session_struct
__to_string = function (s) return util.c("PDG [", s.instance.get_id(), "] (@", s.s_addr, ")") end
}
databus.tx_pdg_connected(self.next_ids.pdg, version, remote_port)
setmetatable(pdg_s, mt)
self.next_ids.pdg = self.next_ids.pdg + 1
databus.tx_pdg_connected(id, version, source_addr)
log.debug(util.c("[SVS] established new session: ", pdg_s))
self.next_ids.pdg = id + 1
-- success
return pdg_s.instance.get_id()
return id
end
-- attempt to identify which session's watchdog timer fired
@@ -458,9 +484,7 @@ end
-- close all open connections
function svsessions.close_all()
-- close sessions
for _, list in pairs(self.sessions) do
_close(list)
end
for _, list in pairs(self.sessions) do _close(list) end
-- free sessions
svsessions.free_all_closed()