#74 supervisor-coordinator comms establish

supervisor/session/coordinator.lua

@@ -1,3 +1,210 @@
+local comms  = require("scada-common.comms")
+local log    = require("scada-common.log")
+local mqueue = require("scada-common.mqueue")
+local util   = require("scada-common.util")
+
 local coordinator = {}
 
+local PROTOCOLS = comms.PROTOCOLS
+local SCADA_MGMT_TYPES = comms.SCADA_MGMT_TYPES
+local SCADA_CRDN_TYPES = comms.SCADA_CRDN_TYPES
+
+local print = util.print
+local println = util.println
+local print_ts = util.print_ts
+local println_ts = util.println_ts
+
+local PERIODICS = {
+    KEEP_ALIVE = 2.0
+}
+
+-- coordinator supervisor session
+---@param id integer
+---@param in_queue mqueue
+---@param out_queue mqueue
+function coordinator.new_session(id, in_queue, out_queue)
+    local log_header = "crdn_session(" .. id .. "): "
+
+    local self = {
+        id = id,
+        in_q = in_queue,
+        out_q = out_queue,
+        -- connection properties
+        seq_num = 0,
+        r_seq_num = nil,
+        connected = true,
+        conn_watchdog = util.new_watchdog(3),
+        last_rtt = 0,
+        -- periodic messages
+        periodics = {
+            last_update = 0,
+            keep_alive = 0
+        }
+    }
+
+    -- mark this coordinator session as closed, stop watchdog
+    local function _close()
+        self.conn_watchdog.cancel()
+        self.connected = false
+    end
+
+    -- send a CRDN packet
+    ---@param msg_type SCADA_CRDN_TYPES
+    ---@param msg table
+    local function _send(msg_type, msg)
+        local s_pkt = comms.scada_packet()
+        local c_pkt = comms.crdn_packet()
+
+        c_pkt.make(msg_type, msg)
+        s_pkt.make(self.seq_num, PROTOCOLS.SCADA_CRDN, c_pkt.raw_sendable())
+
+        self.out_q.push_packet(s_pkt)
+        self.seq_num = self.seq_num + 1
+    end
+
+    -- send a SCADA management packet
+    ---@param msg_type SCADA_MGMT_TYPES
+    ---@param msg table
+    local function _send_mgmt(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
+
+    -- handle a packet
+    ---@param pkt crdn_frame
+    local function _handle_packet(pkt)
+        -- check sequence number
+        if self.r_seq_num == nil then
+            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 = pkt.scada_frame.seq_num()
+        end
+
+        -- feed watchdog
+        self.conn_watchdog.feed()
+
+        -- process packet
+        if pkt.scada_frame.protocol() == PROTOCOLS.SCADA_MGMT then
+            if pkt.type == SCADA_MGMT_TYPES.KEEP_ALIVE then
+                -- keep alive reply
+                if pkt.length == 2 then
+                    local srv_start = pkt.data[1]
+                    local coord_send = pkt.data[2]
+                    local srv_now = util.time()
+                    self.last_rtt = srv_now - srv_start
+
+                    if self.last_rtt > 500 then
+                        log.warning(log_header .. "COORD KEEP_ALIVE round trip time > 500ms (" .. self.last_rtt .. "ms)")
+                    end
+
+                    log.debug(log_header .. "COORD RTT = " .. self.last_rtt .. "ms")
+                    log.debug(log_header .. "COORD TT  = " .. (srv_now - coord_send) .. "ms")
+                else
+                    log.debug(log_header .. "SCADA keep alive packet length mismatch")
+                end
+            elseif pkt.type == SCADA_MGMT_TYPES.CLOSE then
+                -- close the session
+                _close()
+            else
+                log.debug(log_header .. "handler received unsupported SCADA_MGMT packet type " .. pkt.type)
+            end
+        elseif pkt.scada_frame.protocol() == PROTOCOLS.SCADA_CRDN then
+            if pkt.type == SCADA_MGMT_TYPES.KEEP_ALIVE then
+            else
+            end
+        end
+    end
+
+    ---@class coord_session
+    local public = {}
+
+    -- get the session ID
+    function public.get_id() return self.id end
+
+    -- check if a timer matches this session's watchdog
+    function public.check_wd(timer)
+        return self.conn_watchdog.is_timer(timer) and self.connected
+    end
+
+    -- close the connection
+    function public.close()
+        _close()
+        _send_mgmt(SCADA_MGMT_TYPES.CLOSE, {})
+        println("connection to coordinator #" .. self.id .. " closed by server")
+        log.info(log_header .. "session closed by server")
+    end
+
+    -- iterate the session
+    ---@return boolean connected
+    function public.iterate()
+        if self.connected then
+            ------------------
+            -- handle queue --
+            ------------------
+
+            local handle_start = util.time()
+
+            while self.in_q.ready() and self.connected do
+                -- get a new message to process
+                local message = self.in_q.pop()
+
+                if message ~= nil then
+                    if message.qtype == mqueue.TYPE.PACKET then
+                        -- handle a packet
+                        _handle_packet(message.message)
+                    elseif message.qtype == mqueue.TYPE.COMMAND then
+                        -- handle instruction
+                    elseif message.qtype == mqueue.TYPE.DATA then
+                        -- instruction with body
+                    end
+                end
+
+                -- max 100ms spent processing queue
+                if util.time() - handle_start > 100 then
+                    log.warning(log_header .. "exceeded 100ms queue process limit")
+                    break
+                end
+            end
+
+            -- exit if connection was closed
+            if not self.connected then
+                println("connection to coordinator " .. self.id .. " closed by remote host")
+                log.info(log_header .. "session closed by remote host")
+                return self.connected
+            end
+
+            ----------------------
+            -- update periodics --
+            ----------------------
+
+            local elapsed = util.time() - self.periodics.last_update
+
+            local periodics = self.periodics
+
+            -- keep alive
+
+            periodics.keep_alive = periodics.keep_alive + elapsed
+            if periodics.keep_alive >= PERIODICS.KEEP_ALIVE then
+                _send_mgmt(SCADA_MGMT_TYPES.KEEP_ALIVE, { util.time() })
+                periodics.keep_alive = 0
+            end
+
+            self.periodics.last_update = util.time()
+        end
+
+        return self.connected
+    end
+
+    return public
+end
+
 return coordinator

supervisor/session/svsessions.lua

@@ -243,6 +243,34 @@ function svsessions.establish_rtu_session(local_port, remote_port, advertisement
     return rtu_s.instance.get_id()
 end
 
+-- establish a new coordinator session
+---@param local_port integer
+---@param remote_port integer
+---@param version string
+---@return integer|false session_id
+function svsessions.establish_coord_session(local_port, remote_port, version)
+    ---@class coord_session_struct
+    local coord_s = {
+        open = true,
+        version = version,
+        l_port = local_port,
+        r_port = remote_port,
+        in_queue = mqueue.new(),
+        out_queue = mqueue.new(),
+        instance = nil
+    }
+
+    coord_s.instance = coordinator.new_session(self.next_coord_id, coord_s.in_queue, coord_s.out_queue)
+    table.insert(self.coord_sessions, coord_s)
+
+    log.debug("established new coordinator session to " .. remote_port .. " with ID " .. self.next_coord_id)
+
+    self.next_coord_id = self.next_coord_id + 1
+
+    -- success
+    return coord_s.instance.get_id()
+end
+
 -- attempt to identify which session's watchdog timer fired
 ---@param timer_event number
 function svsessions.check_all_watchdogs(timer_event)

supervisor/startup.lua

@@ -13,7 +13,7 @@ local svsessions = require("supervisor.session.svsessions")
 local config     = require("supervisor.config")
 local supervisor = require("supervisor.supervisor")
 
-local SUPERVISOR_VERSION = "beta-v0.4.14"
+local SUPERVISOR_VERSION = "beta-v0.5.1"
 
 local print = util.print
 local println = util.println
@@ -72,7 +72,8 @@ if modem == nil then
 end
 
 -- start comms, open all channels
-local superv_comms = supervisor.comms(SUPERVISOR_VERSION, config.NUM_REACTORS, modem, config.SCADA_DEV_LISTEN, config.SCADA_SV_LISTEN)
+local superv_comms = supervisor.comms(SUPERVISOR_VERSION, config.NUM_REACTORS, config.REACTOR_COOLING, modem,
+                                        config.SCADA_DEV_LISTEN, config.SCADA_SV_LISTEN)
 
 -- base loop clock (6.67Hz, 3 ticks)
 local MAIN_CLOCK = 0.15

supervisor/supervisor.lua

@@ -9,8 +9,9 @@ local supervisor = {}
 local PROTOCOLS = comms.PROTOCOLS
 local RPLC_TYPES = comms.RPLC_TYPES
 local RPLC_LINKING = comms.RPLC_LINKING
-local SCADA_MGMT_TYPES = comms.SCADA_MGMT_TYPES
 local RTU_UNIT_TYPES = comms.RTU_UNIT_TYPES
+local SCADA_MGMT_TYPES = comms.SCADA_MGMT_TYPES
+local SCADA_CRDN_TYPES = comms.SCADA_CRDN_TYPES
 
 local SESSION_TYPE = svsessions.SESSION_TYPE
 
@@ -22,10 +23,11 @@ local println_ts = util.println_ts
 -- supervisory controller communications
 ---@param version string
 ---@param num_reactors integer
+---@param cooling_conf table
 ---@param modem table
 ---@param dev_listen integer
 ---@param coord_listen integer
-function supervisor.comms(version, num_reactors, modem, dev_listen, coord_listen)
+function supervisor.comms(version, num_reactors, cooling_conf, modem, dev_listen, coord_listen)
     local self = {
         version = version,
         num_reactors = num_reactors,
@@ -57,7 +59,7 @@ function supervisor.comms(version, num_reactors, modem, dev_listen, coord_listen
     -- link modem to svsessions
     svsessions.link_modem(self.modem)
 
-    -- send PLC link request responses
+    -- send PLC link request response
     ---@param dest integer
     ---@param msg table
     local function _send_plc_linking(seq_id, dest, msg)
@@ -70,7 +72,7 @@ function supervisor.comms(version, num_reactors, modem, dev_listen, coord_listen
         self.modem.transmit(dest, self.dev_listen, s_pkt.raw_sendable())
     end
 
-    -- send RTU advertisement responses
+    -- send RTU advertisement response
     ---@param seq_id integer
     ---@param dest integer
     local function _send_remote_linked(seq_id, dest)
@@ -83,6 +85,26 @@ function supervisor.comms(version, num_reactors, modem, dev_listen, coord_listen
         self.modem.transmit(dest, self.dev_listen, s_pkt.raw_sendable())
     end
 
+    -- send coordinator connection establish response
+    ---@param seq_id integer
+    ---@param dest integer
+    local function _send_crdn_establish(seq_id, dest)
+        local s_pkt = comms.scada_packet()
+        local c_pkt = comms.crdn_packet()
+
+        local config = { self.num_reactors }
+
+        for i = 1, #cooling_conf do
+            table.insert(config, cooling_conf[i].BOILERS)
+            table.insert(config, cooling_conf[i].TURBINES)
+        end
+
+        c_pkt.make(SCADA_CRDN_TYPES.ESTABLISH, config)
+        s_pkt.make(seq_id, PROTOCOLS.SCADA_CRDN, c_pkt.raw_sendable())
+
+        self.modem.transmit(dest, self.coord_listen, s_pkt.raw_sendable())
+    end
+
     -- PUBLIC FUNCTIONS --
 
     -- reconnect a newly connected modem
@@ -100,7 +122,7 @@ function supervisor.comms(version, num_reactors, modem, dev_listen, coord_listen
     ---@param reply_to integer
     ---@param message any
     ---@param distance integer
-    ---@return modbus_frame|rplc_frame|mgmt_frame|coord_frame|nil packet
+    ---@return modbus_frame|rplc_frame|mgmt_frame|crdn_frame|nil packet
     function public.parse_packet(side, sender, reply_to, message, distance)
         local pkt = nil
         local s_pkt = comms.scada_packet()
@@ -128,10 +150,10 @@ function supervisor.comms(version, num_reactors, modem, dev_listen, coord_listen
                     pkt = mgmt_pkt.get()
                 end
             -- get as coordinator packet
-            elseif s_pkt.protocol() == PROTOCOLS.COORD_DATA then
-                local coord_pkt = comms.coord_packet()
-                if coord_pkt.decode(s_pkt) then
-                    pkt = coord_pkt.get()
+            elseif s_pkt.protocol() == PROTOCOLS.SCADA_CRDN then
+                local crdn_pkt = comms.crdn_packet()
+                if crdn_pkt.decode(s_pkt) then
+                    pkt = crdn_pkt.get()
                 end
             else
                 log.debug("attempted parse of illegal packet type " .. s_pkt.protocol(), true)
@@ -142,7 +164,7 @@ function supervisor.comms(version, num_reactors, modem, dev_listen, coord_listen
     end
 
     -- handle a packet
-    ---@param packet modbus_frame|rplc_frame|mgmt_frame|coord_frame
+    ---@param packet modbus_frame|rplc_frame|mgmt_frame|crdn_frame
     function public.handle_packet(packet)
         if packet ~= nil then
             local l_port = packet.scada_frame.local_port()
@@ -226,7 +248,7 @@ function supervisor.comms(version, num_reactors, modem, dev_listen, coord_listen
                         end
                     else
                         -- any other packet should be session related, discard it
-                        log.debug("discarding SCADA_MGMT packet without a known session")
+                        log.debug(util.c(r_port, "->", l_port, ": discarding SCADA_MGMT packet without a known session"))
                     end
                 else
                     log.debug("illegal packet type " .. protocol .. " on device listening channel")
@@ -238,8 +260,34 @@ function supervisor.comms(version, num_reactors, modem, dev_listen, coord_listen
 
                 if protocol == PROTOCOLS.SCADA_MGMT then
                     -- SCADA management packet
-                elseif protocol == PROTOCOLS.COORD_DATA then
+                    if session ~= nil then
+                        -- pass the packet onto the session handler
+                        session.in_queue.push_packet(packet)
+                    else
+                        -- any other packet should be session related, discard it
+                        log.debug(util.c(r_port, "->", l_port, ": discarding SCADA_MGMT packet without a known session"))
+                    end
+                elseif protocol == PROTOCOLS.SCADA_CRDN then
                     -- coordinator packet
+                    if session ~= nil then
+                        -- pass the packet onto the session handler
+                        session.in_queue.push_packet(packet)
+                    elseif packet.type == SCADA_CRDN_TYPES.ESTABLISH then
+                        if packet.length == 1 then
+                            -- this is an attempt to establish a new session
+                            println(util.c("connected to coordinator [:", r_port, "]"))
+
+                            svsessions.establish_coord_session(l_port, r_port, packet.data[1])
+
+                            log.debug("CRDN_ESTABLISH: connected to " .. r_port)
+                            _send_crdn_establish(packet.scada_frame.seq_num() + 1, r_port)
+                        else
+                            log.debug("CRDN_ESTABLISH: establish packet length mismatch")
+                        end
+                    else
+                        -- any other packet should be session related, discard it
+                        log.debug(util.c(r_port, "->", l_port, ": discarding SCADA_CRDN packet without a known session"))
+                    end
                 else
                     log.debug("illegal packet type " .. protocol .. " on coordinator listening channel")
                 end