dual_link.py

Go to the documentation of this file.

"""Dual-homed UDP link helper (ADR-029) - keeps one socket open to EACH
site's C at all times, reconnecting independently, so a Train/IL sim
never visibly drops contact across a site failover: whichever site is
currently ONLINE is the only one that ever answers, but both sockets stay
open so there is nothing to notice or reconnect from the sim's own point
of view when ONLINE moves from one site to the other. See ADR-029's own
"Failover continuity" design note.
 
Transport: UDP with a HELLO/HELLO-ACK handshake (ADR-027), NOT plain TCP -
this project's own sapi_netlink POSIX backend
(safeAPIRBC2oo2/src/posix_backend/sapi_posix_backend_netlink.c) moved off
TCP project-wide; C's LISTEN side for these links speaks the exact same
protocol it already does for A/B. This module is a hand-ported Python
mirror of that CONNECT-side handshake (open_connect_udp() in that C
file) - keep the two in sync if that protocol ever changes:
  - HELLO = 0xA5, HELLO_ACK = 0x5A, one byte each.
  - CONNECT side: connect() a UDP socket (fixes the default peer address
    only - no network I/O, no reachability signal by itself), then send
    HELLO every ~20ms and poll for a 1-byte 0x5A reply until one arrives
    or a bounded budget elapses.
Once handshake completes, every later send()/recv() on that socket is
exactly one RBC envelope datagram - sapi_netlink's own message-per-
datagram framing, matching RBC_ENVELOPE_WIRE_SIZE-sized reads.
 
First simple pass, not a general-purpose framing layer: each recv() is
expected to be exactly one whole datagram (true for UDP, unlike TCP's own
stream semantics) of exactly message_size bytes; a differently-sized
datagram is dropped (logged), not reassembled or truncated-and-kept.
"""
 
import logging
import select
import selectors
import socket
import time
 
SITES = ("WEST", "EAST")
 
HELLO = 0xA5
HELLO_ACK = 0x5A
HELLO_PERIOD_S = 0.02  # matches POSIX_NETLINK_UDP_HELLO_PERIOD_MS (sapi_posix_backend_netlink.c)
HANDSHAKE_BUDGET_S = 0.3  # bounded per ensure_connected() call - see that method's own doc
 
 
class DualHomedLink:
    def __init__(self, log, targets, message_size):
        """@param targets: {"WEST": (host, port), "EAST": (host, port)}"""
        self.log = log
        self.targets = targets
        self.message_size = message_size
        self.sockets = {site: None for site in SITES}
        self.sel = selectors.DefaultSelector()
 
    def ensure_connected(self):
        """Bounded (HANDSHAKE_BUDGET_S) HELLO/HELLO-ACK attempt for each
        not-yet-connected site, per call - short enough not to stall the
        sim's own overall pacing for long, retried forever (call this
        every loop iteration) until it succeeds, matching this project's
        own C-side reconnect-forever convention."""
        for site in SITES:
            if self.sockets[site] is not None:
                continue
            self._handshake(site)
 
    def _handshake(self, site):
        host, port = self.targets[site]
        try:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.setblocking(False)
            s.connect((host, port))  # UDP connect(): fixes the peer locally only, no I/O - see this file's own doc
        except OSError:
            return
 
        deadline = time.monotonic() + HANDSHAKE_BUDGET_S
        next_hello = 0.0
        got_ack = False
        while time.monotonic() < deadline:
            now = time.monotonic()
            if now >= next_hello:
                try:
                    s.send(bytes([HELLO]))
                except OSError:
                    pass  # ECONNREFUSED etc. - peer not up yet, keep retrying within this budget
                next_hello = now + HELLO_PERIOD_S
            remaining = min(next_hello - now, deadline - now)
            if remaining > 0:
                try:
                    r, _w, _x = select.select([s], [], [], max(remaining, 0.0))
                except OSError:
                    break
                if not r:
                    continue
            try:
                data = s.recv(1)
            except OSError:
                continue
            if data == bytes([HELLO_ACK]):
                got_ack = True
                break
            # Not our ACK (or a stray real envelope datagram arriving
            # early, per this file's own header note on why the C side
            # can start sending before it sees our own ACK) - keep
            # waiting within the same budget; a non-ACK single byte is
            # simply dropped here (there is no legitimate 1-byte RBC
            # envelope, so this can't be mistaken for real traffic).
 
        if got_ack:
            # Drain any trailing 1-byte handshake packets in the socket buffer
            while True:
                try:
                    trailing = s.recv(1)
                    if not trailing:
                        break
                except (BlockingIOError, OSError):
                    break
            s.setblocking(False)
            self.sockets[site] = s
            self.sel.register(s, selectors.EVENT_READ, site)
            self.log.info("connected to %s (%s:%d)", site, host, port)
        else:
            try:
                s.close()
            except OSError:
                pass
 
    def connected_sites(self):
        """Returns the list of currently-connected site names - generic
        transport-layer state a caller's own application layer can use
        for its own purposes (e.g. a "which site do I currently believe
        is reachable" heuristic); this module has no opinion on what
        that purpose is."""
        return [site for site in SITES if self.sockets[site] is not None]
 
    def disconnect(self, site):
        """Closes this site's connection right now, if open - the sim's
        own main loop reconnects it automatically on its next
        ensure_connected() call, per this project's reconnect-forever
        convention (this file's own header). Public wrapper around the
        same internal close a real send/recv failure already uses
        (_close) - a test-harness-triggered disconnect and a genuine
        network failure look identical from here on, deliberately: that
        is exactly the condition a disconnect test wants to reproduce."""
        self._close(site)
 
    def send_all(self, payload):
        """Sends @payload to every currently-connected site - both sites
        must see identical inputs (mirrors this project's C-side
        broadcast-to-both-A-and-B convention), since only whichever site
        is ONLINE will actually act on it. Silently skips a not-yet-
        connected site; closes+reconnects on a genuine send failure."""
        for site in SITES:
            s = self.sockets[site]
            if s is None:
                continue
            try:
                s.send(payload)
                self.log.debug("sent %d bytes to %s", len(payload), site)
            except OSError:
                self._close(site)
 
    def poll_recv(self, timeout=0.0):
        """Yields (site, payload) for each ready socket - see this
        module's own header for the "one recv() == one datagram" note.
        Honors @timeout (actually sleeps for it) even while NEITHER site
        is connected yet, rather than returning instantly - a caller that
        relies on this call for its own loop's pacing (as every sim here
        does) would otherwise busy-loop, spamming reconnect attempts as
        fast as the CPU allows, whenever both sites are simultaneously
        down (selectors.BaseSelector.select() itself provides no timeout
        behavior at all when it has nothing registered to select on)."""
        if not self.sel.get_map():
            if timeout > 0.0:
                time.sleep(timeout)
            return
        events = self.sel.select(timeout=timeout)
        for key, _mask in events:
            site = key.data
            s = self.sockets.get(site)
            if s is None:
                continue
            try:
                data = s.recv(self.message_size)
            except OSError:
                self._close(site)
                continue
            if not data:
                self.log.info("%s disconnected - will reconnect", site)
                self._close(site)
                continue
            if len(data) != self.message_size:
                if len(data) == 1 and data[0] in (HELLO, HELLO_ACK):
                    continue  # Silently skip leftover 1-byte handshake datagrams
                self.log.warning("%s: unexpected datagram size (%d/%d bytes) - dropping", site, len(data),
                                  self.message_size)
                continue
            self.log.debug("received %d bytes from %s", len(data), site)
            yield site, data
 
    def _close(self, site):
        s = self.sockets[site]
        if s is not None:
            try:
                self.sel.unregister(s)
            except KeyError:
                pass
            try:
                s.close()
            except OSError:
                pass
            self.sockets[site] = None