safecomm.py

Go to the documentation of this file.

"""SafeCommLink - the CommServer-facing client link for a sim (train).
 
This is the Python client side of the safe-communication path introduced
with the C ``safeCommFreamwork`` / ``CommServer`` product:
 
    TrainSim --(this module, plain TCP)--> CommServer --> RBC role-C gateway
 
CommServer owns the WEST/EAST dual-homing, so from the sim's point of view
there is a single logical link. It is a drop-in alternative to
``simcore.dual_link.DualHomedLink`` - same ``ensure_connected()`` /
``send_all()`` / ``poll_recv()`` surface - selected by ``train_sim`` when
the config carries a ``commServer`` block.
 
Stub, matching the C side: the Safety Functional Module (authentication,
per-message MAC, sequence numbering - ``safecomm::safety``) is
NOT_IMPLEMENTED, so this link is a transparent byte pipe with fixed
``message_size`` framing. When the SaF PDU layout lands in
``safeCommFreamwork`` this module gets the matching wrapper/unwrapper,
kept byte-identical the same way ``simcore.rbc_wire`` mirrors the C codec.
"""
 
import socket
import time
 
# CommServer hides which real site (WEST/EAST) answered - it owns the
# dual-homing. The sim's per-site bookkeeping (was_connected, log tags,
# `for site in SITES`) still expects a site name, so this link presents
# everything under the WEST slot and leaves EAST permanently unused.
_SITE = "WEST"
_UNUSED_SITE = "EAST"
 
# Reconnect policy: this is a CLIENT link - if CommServer restarts, its
# host goes away, or the stream errors, the sim must recover on its own.
# ensure_connected() is called once per sim cycle; it makes at most ONE
# non-blocking connect() attempt per call and never spins/sleeps inside a
# single call (a blocking retry loop here stalled the whole sim - control
# port, status, other trains' M136 - whenever CommServer was down).
_CONNECT_TIMEOUT_S = 0.75          # bound one connect() attempt
_RECONNECT_MIN_INTERVAL_S = 1.0    # don't hammer connect() every ~0.3 s cycle
 
 
class SafeCommLink:
    """Single TCP link to a CommServer, framed as fixed-size messages.
    Auto-reconnecting: any send/recv error drops the socket, and the next
    ensure_connected() re-dials (rate-limited)."""
 
    def __init__(self, log, host, port, message_size):
        self._log = log
        self._host = host
        self._port = int(port)
        self.message_size = int(message_size)
        self._sock = None
        self._rx = bytearray()
        self._last_attempt = 0.0
        self._ever_connected = False
 
    # -- lifecycle --------------------------------------------------------
    def ensure_connected(self):
        """One non-blocking (re)connect attempt, rate-limited. Safe to call
        every cycle - a no-op once connected. Returns True if connected."""
        if self._sock is not None:
            return True
        now = time.monotonic()
        if (now - self._last_attempt) < _RECONNECT_MIN_INTERVAL_S:
            return False
        self._last_attempt = now
        try:
            s = socket.create_connection((self._host, self._port), timeout=_CONNECT_TIMEOUT_S)
            s.setblocking(False)
            try:
                s.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
                s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            except OSError:
                pass
            self._sock = s
            self._rx.clear()
            level = self._log.info if not self._ever_connected else self._log.warning
            level("SafeCommLink %s to CommServer %s:%d",
                  "connected" if not self._ever_connected else "RECONNECTED",
                  self._host, self._port)
            self._ever_connected = True
            return True
        except OSError as exc:
            # DEBUG, not WARNING: this fires every retry interval while
            # CommServer is down and would otherwise flood the log.
            self._log.debug("SafeCommLink connect %s:%d failed (%s) - will retry",
                            self._host, self._port, exc)
            return False
 
    @property
    def sockets(self):
        """DualHomedLink-compatible view: WEST slot carries the one TCP
        socket, EAST slot is always None."""
        return {_SITE: self._sock, _UNUSED_SITE: None}
 
    def connected_sites(self):
        return [_SITE] if self._sock is not None else []
 
    def disconnect(self, site=None):  # noqa: ARG002 - parity with DualHomedLink
        self._close()
 
    def _close(self, site=None):  # noqa: ARG002
        if self._sock is not None:
            try:
                self._sock.close()
            except OSError:
                pass
            self._sock = None
        self._rx.clear()
 
    # -- I/O ------------------------------------------------------------
    def _frame(self, payload):
        b = bytes(payload)
        if len(b) < self.message_size:
            b = b + b"\x00" * (self.message_size - len(b))
        elif len(b) > self.message_size:
            b = b[: self.message_size]
        return b
 
    def send_all(self, payload):
        """Send one message to the CommServer (padded/truncated to
        ``message_size``). Silently drops if not connected - the caller's
        own ``ensure_connected()`` loop re-establishes."""
        if self._sock is None:
            return
        try:
            self._sock.sendall(self._frame(payload))
        except OSError as exc:
            self._log.warning("SafeCommLink send failed (%s) - dropping link", exc)
            self._close()
 
    def poll_recv(self, timeout=0.0):
        """Yield ``(site, message_bytes)`` for every whole framed message
        available within ``timeout`` seconds."""
        if self._sock is None:
            return
        end = time.monotonic() + max(0.0, timeout)
        first = True
        while first or time.monotonic() < end:
            first = False
            try:
                chunk = self._sock.recv(4096)
            except BlockingIOError:
                if timeout <= 0.0:
                    break
                time.sleep(0.005)
                continue
            except OSError as exc:
                self._log.warning("SafeCommLink recv failed (%s) - dropping link", exc)
                self._close()
                return
            if not chunk:
                self._log.warning("SafeCommLink: CommServer closed the connection")
                self._close()
                return
            self._rx.extend(chunk)
            while len(self._rx) >= self.message_size:
                msg = bytes(self._rx[: self.message_size])
                del self._rx[: self.message_size]
                yield _SITE, msg
 
 
def link_from_config(log, config, message_size, sim_config):
    """Return a SafeCommLink if the config has a ``commServer`` block,
    else None (caller falls back to DualHomedLink)."""
    block = config.get("commServer") if isinstance(config, dict) else None
    if not block:
        return None
    host = sim_config.require(config, "commServer", "host")
    port = sim_config.require(config, "commServer", "port")
    return SafeCommLink(log, host, port, message_size)