Minecraft/cc-mek-scada
1
0
Fork 0
mirror of https://github.com/MikaylaFischler/cc-mek-scada.git synced 2024-08-30 18:22:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

#41 close session connections

Browse Source
This commit is contained in:
Mikayla Fischler 2022-05-02 11:42:24 -04:00
parent 7ff0e25711
commit 76c81395b7
5 changed files with 149 additions and 80 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
scada-common/comms.lua
View File

@ -27,10 +27,10 @@ RPLC_LINKING = {
SCADA_MGMT_TYPES = {
PING = 0, -- generic ping
SV_HEARTBEAT = 1, -- supervisor heartbeat
CLOSE = 1, -- close a connection
REMOTE_LINKED = 2, -- remote device linked
RTU_ADVERT = 3, -- RTU capability advertisement
RTU_HEARTBEAT = 4, -- RTU heartbeat
RTU_HEARTBEAT = 4 -- RTU heartbeat
}
RTU_ADVERT_TYPES = {

139
supervisor/session/plc.lua
View File

@ -167,6 +167,7 @@ function new_session(id, for_reactor, in_queue, out_queue)
self.sDB.mek_struct.max_burn = mek_data[8]
end
-- get an ACK status
local _get_ack = function (pkt)
if pkt.length == 1 then
return pkt.data[1]
@ -176,36 +177,35 @@ function new_session(id, for_reactor, in_queue, out_queue)
end
end
local _handle_packet = function (rplc_pkt)
local checks_ok = true
-- handle a packet
local _handle_packet = function (pkt)
-- check sequence number
if self.r_seq_num == nil then
self.r_seq_num = rplc_pkt.scada_frame.seq_num()
elseif self.r_seq_num >= rplc_pkt.scada_frame.seq_num() then
log._warning(log_header .. "sequence out-of-order: last = " .. self.r_seq_num .. ", new = " .. rplc_pkt.scada_frame.seq_num())
checks_ok = false
self.r_seq_num = pkt.scada_frame.seq_num()
elseif self.r_seq_num >= pkt.scada_frame.seq_num() then
log._warning(log_header .. "sequence out-of-order: last = " .. self.r_seq_num .. ", new = " .. pkt.scada_frame.seq_num())
return
else
self.r_seq_num = rplc_pkt.scada_frame.seq_num()
end
-- check reactor ID
if rplc_pkt.id ~= for_reactor then
log._warning(log_header .. "RPLC packet with ID not matching reactor ID: reactor " .. self.for_reactor .. " != " .. rplc_pkt.id)
checks_ok = false
self.r_seq_num = pkt.scada_frame.seq_num()
end
-- process packet
if checks_ok then
if pkt.scada_frame.protocol() == PROTOCOLS.RPLC then
-- check reactor ID
if pkt.id ~= for_reactor then
log._warning(log_header .. "RPLC packet with ID not matching reactor ID: reactor " .. self.for_reactor .. " != " .. pkt.id)
return
end
-- feed watchdog
self.plc_conn_watchdog.feed()
-- handle packet by type
if rplc_pkt.type == RPLC_TYPES.KEEP_ALIVE then
if pkt.type == RPLC_TYPES.KEEP_ALIVE then
-- keep alive reply
if rplc_pkt.length == 2 then
local srv_start = rplc_pkt.data[1]
local plc_send = rplc_pkt.data[2]
if pkt.length == 2 then
local srv_start = pkt.data[1]
local plc_send = pkt.data[2]
local srv_now = util.time()
self.last_rtt = srv_now - srv_start
@ -218,18 +218,18 @@ function new_session(id, for_reactor, in_queue, out_queue)
else
log._debug(log_header .. "RPLC keep alive packet length mismatch")
end
elseif rplc_pkt.type == RPLC_TYPES.STATUS then
elseif pkt.type == RPLC_TYPES.STATUS then
-- status packet received, update data
if rplc_pkt.length >= 5 then
self.sDB.last_status_update = rplc_pkt.data[1]
self.sDB.control_state = rplc_pkt.data[2]
self.sDB.overridden = rplc_pkt.data[3]
self.sDB.degraded = rplc_pkt.data[4]
self.sDB.mek_status.heating_rate = rplc_pkt.data[5]
if pkt.length >= 5 then
self.sDB.last_status_update = pkt.data[1]
self.sDB.control_state = pkt.data[2]
self.sDB.overridden = pkt.data[3]
self.sDB.degraded = pkt.data[4]
self.sDB.mek_status.heating_rate = pkt.data[5]
-- attempt to read mek_data table
if rplc_pkt.data[6] ~= nil then
local status = pcall(_copy_status, rplc_pkt.data[6])
if pkt.data[6] ~= nil then
local status = pcall(_copy_status, pkt.data[6])
if status then
-- copied in status data OK
else
@ -240,10 +240,10 @@ function new_session(id, for_reactor, in_queue, out_queue)
else
log._debug(log_header .. "RPLC status packet length mismatch")
end
elseif rplc_pkt.type == RPLC_TYPES.MEK_STRUCT then
elseif pkt.type == RPLC_TYPES.MEK_STRUCT then
-- received reactor structure, record it
if rplc_pkt.length == 8 then
local status = pcall(_copy_struct, rplc_pkt.data)
if pkt.length == 8 then
local status = pcall(_copy_struct, pkt.data)
if status then
-- copied in structure data OK
else
@ -253,35 +253,36 @@ function new_session(id, for_reactor, in_queue, out_queue)
else
log._debug(log_header .. "RPLC struct packet length mismatch")
end
elseif rplc_pkt.type == RPLC_TYPES.MEK_SCRAM then
elseif pkt.type == RPLC_TYPES.MEK_SCRAM then
-- SCRAM acknowledgement
local ack = _get_ack(rplc_pkt)
local ack = _get_ack(pkt)
if ack then
self.acks.scram = true
self.sDB.control_state = false
elseif ack == false then
log._debug(log_header .. "SCRAM failed!")
end
elseif rplc_pkt.type == RPLC_TYPES.MEK_ENABLE then
elseif pkt.type == RPLC_TYPES.MEK_ENABLE then
-- enable acknowledgement
local ack = _get_ack(rplc_pkt)
local ack = _get_ack(pkt)
if ack then
self.acks.enable = true
self.sDB.control_state = true
elseif ack == false then
log._debug(log_header .. "enable failed!")
end
elseif rplc_pkt.type == RPLC_TYPES.MEK_BURN_RATE then
elseif pkt.type == RPLC_TYPES.MEK_BURN_RATE then
-- burn rate acknowledgement
if _get_ack(rplc_pkt) then
local ack = _get_ack(pkt)
if ack then
self.acks.burn_rate = true
else
elseif ack == false then
log._debug(log_header .. "burn rate update failed!")
end
elseif rplc_pkt.type == RPLC_TYPES.ISS_STATUS then
elseif pkt.type == RPLC_TYPES.ISS_STATUS then
-- ISS status packet received, copy data
if rplc_pkt.length == 7 then
local status = pcall(_copy_iss_status, rplc_pkt.data)
if pkt.length == 7 then
local status = pcall(_copy_iss_status, pkt.data)
if status then
-- copied in ISS status data OK
else
@ -291,13 +292,13 @@ function new_session(id, for_reactor, in_queue, out_queue)
else
log._debug(log_header .. "RPLC ISS status packet length mismatch")
end
elseif rplc_pkt.type == RPLC_TYPES.ISS_ALARM then
elseif pkt.type == RPLC_TYPES.ISS_ALARM then
-- ISS alarm
self.sDB.overridden = true
if rplc_pkt.length == 8 then
if pkt.length == 8 then
self.sDB.iss_tripped = true
self.sDB.iss_trip_cause = rplc_pkt.data[1]
local status = pcall(_copy_iss_status, { table.unpack(rplc_pkt.data, 2, #rplc_pkt.length) })
self.sDB.iss_trip_cause = pkt.data[1]
local status = pcall(_copy_iss_status, { table.unpack(pkt.data, 2, #pkt.length) })
if status then
-- copied in ISS status data OK
else
@ -307,21 +308,30 @@ function new_session(id, for_reactor, in_queue, out_queue)
else
log._debug(log_header .. "RPLC ISS alarm packet length mismatch")
end
elseif rplc_pkt.type == RPLC_TYPES.ISS_CLEAR then
elseif pkt.type == RPLC_TYPES.ISS_CLEAR then
-- ISS clear acknowledgement
if _get_ack(rplc_pkt) then
local ack = _get_ack(pkt)
if ack then
self.acks.iss_tripped = true
self.sDB.iss_tripped = false
self.sDB.iss_trip_cause = "ok"
else
elseif ack == false then
log._debug(log_header .. "ISS clear failed")
end
else
log._debug(log_header .. "handler received unsupported RPLC packet type " .. rplc_pkt.type)
log._debug(log_header .. "handler received unsupported RPLC packet type " .. pkt.type)
end
elseif pkt.scada_frame.protocol() == PROTOCOLS.SCADA_MGMT then
if pkt.type == SCADA_MGMT_TYPES.CLOSE then
-- close the session
self.connected = false
else
log._debug(log_header .. "handler received unsupported SCADA_MGMT packet type " .. pkt.type)
end
end
end
-- send an RPLC packet
local _send = function (msg_type, msg)
local s_pkt = comms.scada_packet()
local r_pkt = comms.rplc_packet()
@ -333,18 +343,38 @@ function new_session(id, for_reactor, in_queue, out_queue)
self.seq_num = self.seq_num + 1
end
-- send a SCADA management packet
local _send_mgmt = function (msg_type, msg)
local s_pkt = comms.scada_packet()
local m_pkt = comms.mgmt_packet()
m_pkt.make(msg_type, msg)
s_pkt.make(self.seq_num, PROTOCOLS.SCADA_MGMT, m_pkt.raw_sendable())
self.out_q.push_packet(s_pkt)
self.seq_num = self.seq_num + 1
end
-- PUBLIC FUNCTIONS --
-- get the session ID
local get_id = function () return self.id end
-- get the session database
local get_db = function () return self.sDB end
local close = function () self.connected = false end
-- close the connection
local close = function ()
self.connected = false
_send_mgmt(SCADA_MGMT_TYPES.CLOSE, {})
end
-- check if a timer matches this session's watchdog
local check_wd = function (timer)
return timer == self.plc_conn_watchdog.get_timer()
end
-- get the reactor structure
local get_struct = function ()
if self.received_struct then
return self.sDB.mek_struct
@ -353,6 +383,7 @@ function new_session(id, for_reactor, in_queue, out_queue)
end
end
-- iterate the session
local iterate = function ()
if self.connected then
------------------
@ -361,7 +392,7 @@ function new_session(id, for_reactor, in_queue, out_queue)
local handle_start = util.time()
while self.in_q.ready() do
while self.in_q.ready() and self.connected do
-- get a new message to process
local message = self.in_q.pop()
@ -406,6 +437,12 @@ function new_session(id, for_reactor, in_queue, out_queue)
end
end
-- exit if connection was closed
if not self.connected then
log._info(log_header .. "session closed by remote host")
return false
end
----------------------
-- update periodics --
----------------------

51
supervisor/session/svsessions.lua
View File

@ -24,39 +24,33 @@ function link_modem(modem)
self.modem = modem
end
function alloc_reactor_plcs(num_reactors)
self.num_reactors = num_reactors
for i = 1, num_reactors do
table.insert(self.plc_sessions, false)
end
end
function find_session(stype, remote_port)
if stype == SESSION_TYPE.RTU_SESSION then
-- find a session by the remote port
function find_session(,remote_port)
-- check RTU sessions
for i = 1, #self.rtu_sessions do
if self.rtu_sessions[i].r_port == remote_port then
return self.rtu_sessions[i]
end
end
elseif stype == SESSION_TYPE.PLC_SESSION then
-- check PLC sessions
for i = 1, #self.plc_sessions do
if self.plc_sessions[i].r_port == remote_port then
return self.plc_sessions[i]
end
end
elseif stype == SESSION_TYPE.COORD_SESSION then
-- check coordinator sessions
for i = 1, #self.coord_sessions do
if self.coord_sessions[i].r_port == remote_port then
return self.coord_sessions[i]
end
end
else
log._error("cannot search for unknown session type " .. stype, true)
end
return nil
end
-- get a session by reactor ID
function get_reactor_session(reactor)
local session = nil
@ -69,6 +63,7 @@ function get_reactor_session(reactor)
return session
end
-- establish a new PLC session
function establish_plc_session(local_port, remote_port, for_reactor)
if get_reactor_session(for_reactor) == nil then
local plc_s = {
@ -96,6 +91,7 @@ function establish_plc_session(local_port, remote_port, for_reactor)
end
end
-- check if a watchdog timer event matches that of one of the provided sessions
local function _check_watchdogs(sessions, timer_event)
for i = 1, #sessions do
local session = sessions[i]
@ -110,6 +106,7 @@ local function _check_watchdogs(sessions, timer_event)
end
end
-- attempt to identify which session's watchdog timer fired
function check_all_watchdogs(timer_event)
-- check RTU session watchdogs
_check_watchdogs(self.rtu_sessions, timer_event)
@ -121,6 +118,7 @@ function check_all_watchdogs(timer_event)
_check_watchdogs(self.coord_sessions, timer_event)
end
-- iterate all the given sessions
local function _iterate(sessions)
for i = 1, #sessions do
local session = sessions[i]
@ -142,6 +140,7 @@ local function _iterate(sessions)
end
end
-- iterate all sessions
function iterate_all()
-- iterate RTU sessions
_iterate(self.rtu_sessions)
@ -153,6 +152,7 @@ function iterate_all()
_iterate(self.coord_sessions)
end
-- delete any closed sessions
local function _free_closed(sessions)
local move_to = 1
for i = 1, #sessions do
@ -172,6 +172,7 @@ local function _free_closed(sessions)
end
end
-- delete all closed sessions
function free_all_closed()
-- free closed RTU sessions
_free_closed(self.rtu_sessions)
@ -182,3 +183,25 @@ function free_all_closed()
-- free closed coordinator sessions
_free_closed(self.coord_sessions)
end
-- close connections
local function _close(sessions)
for i = 1, #sessions do
local session = sessions[i]
if session.open then
session.open = false
session.instance.close()
end
end
end
-- close all open connections
function close_all()
-- close sessions
_close(self.rtu_sessions)
_close(self.plc_sessions)
_close(self.coord_sessions)
-- free sessions
free_all_closed()
end

7
supervisor/startup.lua
View File

@ -18,7 +18,7 @@ os.loadAPI("session/svsessions.lua")
os.loadAPI("supervisor.lua")
local SUPERVISOR_VERSION = "alpha-v0.1.10"
local SUPERVISOR_VERSION = "alpha-v0.1.11"
local print = util.print
local println = util.println
@ -102,7 +102,10 @@ while true do
-- check for termination request
if event == "terminate" or ppm.should_terminate() then
log._info("terminate requested, exiting...")
println_ts("closing sessions...")
log._info("terminate requested, closing sessions...")
svsessions.close_all()
log._info("sessions closed")
break
end
end

8
supervisor/supervisor.lua
View File

@ -117,11 +117,13 @@ function superv_comms(num_reactors, modem, dev_listen, coord_listen)
-- device (RTU/PLC) listening channel
if l_port == self.dev_listen then
-- look for an associated session
local session = svsessions.find_session(r_port)
if protocol == PROTOCOLS.MODBUS_TCP then
-- MODBUS response
elseif protocol == PROTOCOLS.RPLC then
-- reactor PLC packet
local session = svsessions.find_session(SESSION_TYPE.PLC_SESSION, r_port)
if session then
if packet.type == RPLC_TYPES.LINK_REQ then
-- new device on this port? that's a collision
@ -158,6 +160,10 @@ function superv_comms(num_reactors, modem, dev_listen, coord_listen)
end
elseif protocol == PROTOCOLS.SCADA_MGMT then
-- SCADA management packet
if session then
-- pass the packet onto the session handler
session.in_queue.push_packet(packet)
end
else
log._debug("illegal packet type " .. protocol .. " on device listening channel")
end