gateway_c_train_handler.c

Go to the documentation of this file.

#include "gateway_c_train_handler.h"
 
#include <stdio.h>
#include "rte/oal/memory/rte_mem_util.h"
#include "rte/oal/log/rte_log.h"
#include "common/common_config.h"
#include "common/rbc_util_frame.h"
#include "gateway_c_common.h"
#include "gateway_c_selfstatus.h"
 
static gateway_c_train_handler_state_t g_train;
 
static const char *peer_name(gateway_c_peer_t peer)
{
    return (peer == GATEWAY_C_PEER_A) ? "A" : "B";
}
 
rte_status_t gateway_c_train_handler_init(gateway_c_context_t *ctx)
{
    rte_status_t status;
 
    g_train.last_known_tag = "SUPPRESSED";
    g_train.relay_consecutive_timeout = 0U;
    g_train.relay_down_since_ms[GATEWAY_C_PEER_A] = 0U;
    g_train.relay_down_since_ms[GATEWAY_C_PEER_B] = 0U;
    g_train.send_down_since_ms[GATEWAY_C_PEER_A] = 0U;
    g_train.send_down_since_ms[GATEWAY_C_PEER_B] = 0U;
    g_train.a_seen_this_cycle = 0U;
    g_train.a_stale_cycles = 0U;
    gateway_c_dedup_init(&g_train.dedup);
 
    RTE_EXAMPLE_LOG("[GW %s] GP: opening channel 'c-relay-train-a' to RTE\n", site_config_name(ctx->site));
    status = rte_channel_service_setup(&g_train.relay_channel[GATEWAY_C_PEER_A], "c-relay-train-a");
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[GW %s] TrainHandler: relay A channel setup failed (%s)\n",
                             site_config_name(ctx->site), rte_status_to_string(status));
        return status;
    }
    RTE_EXAMPLE_LOG("[GW %s] GP: opening channel 'c-relay-train-b' to RTE\n", site_config_name(ctx->site));
    status = rte_channel_service_setup(&g_train.relay_channel[GATEWAY_C_PEER_B], "c-relay-train-b");
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[GW %s] TrainHandler: relay B channel setup failed (%s)\n",
                             site_config_name(ctx->site), rte_status_to_string(status));
        return status;
    }
    /* REQ-RBC-009: c-sim-train is listen-role (etc/config/c_<site>.json), so this setup only binds the local port
     * and never waits for the Train sim, which may start later. The peer is the most recent sender, so a sim that
     * restarts on a new source port is reached again; a send before the first datagram returns NOT_CONNECTED,
     * which send_to_sim() treats like OK. */
    RTE_EXAMPLE_LOG("[GW %s] GP: opening channel 'c-sim-train' to RTE\n", site_config_name(ctx->site));
    status = rte_channel_service_setup(&g_train.sim_channel, "c-sim-train");
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[GW %s] TrainHandler: sim channel setup failed (%s)\n",
                             site_config_name(ctx->site), rte_status_to_string(status));
        return status;
    }
    return RTE_STATUS_OK;
}
 
/* @p len is caller-supplied (not hardcoded) since this relay now carries
 * two different frame sizes on the same link: RBC_ENVELOPE_WIRE_SIZE
 * (96, the flat scheme's own kinds, keepalives included) and
 * RBC_TRAIN_ENVELOPE_WIRE_SIZE (512, RBC_MSG_ERTMS_ENVELOPE) - a caller
 * sending the smaller shape must never have this function pad it out to
 * the bigger one from whatever happens to follow its buffer. */
static void send_to_relay(gateway_c_context_t *ctx, gateway_c_peer_t peer,
                          const uint8_t *bytes, size_t len)
{
    rte_status_t status = rte_channel_service_send(&g_train.relay_channel[peer], bytes, len, 0U);
    /* NOT_CONNECTED (REQ-CHANSVC-001): this listen-role relay has not had a datagram from A/B yet (still starting),
     * so nothing was sent. That is not a failed send: handled exactly as OK, as before the status existed (the
     * channel service then returned OK and dropped the frame), so no send-failure window and no log. */
    if ((status == RTE_STATUS_OK) || (status == RTE_STATUS_NOT_CONNECTED))
    {
        /* A send that goes out again ends any send-failure window recorded below. Receive-side
         * silence alone cannot detect a dead link on UDP (a read just times out), so this and the
         * failure case below are what feed gateway_c_train_handler_check_link_down(). */
        g_train.send_down_since_ms[peer] = 0U;
    }
    else
    {
        /* TIMEOUT counts too: the POSIX channel backend maps EHOSTUNREACH/ECONNREFUSED/ENOTCONN (what a partitioned
         * peer produces) to TIMEOUT, so ignoring it left a partitioned Gateway running for good (Docker, C West
         * partition test). Any successful send resets the window, so a busy queue cannot trip it. */
        if (g_train.send_down_since_ms[peer] == 0U)
        {
            (void)rte_timer_now(&g_train.send_down_since_ms[peer]);
        }
        if (status == RTE_STATUS_TIMEOUT)
        {
            return;
        }
        RTE_EXAMPLE_LOG("[GW %s] TrainHandler: forward to %s failed (%s)\n", site_config_name(ctx->site),
                             peer_name(peer), rte_status_to_string(status));
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, "GW", peer_name(peer),
                              "ENVELOPE", rte_status_to_string(status), NULL);
    }
}
 
static void send_to_sim(gateway_c_context_t *ctx, const uint8_t *bytes, size_t len)
{
    rte_status_t status = rte_channel_service_send(&g_train.sim_channel, bytes, len, 0U);
    /* NOT_CONNECTED: no Train sim has sent on this listen-role link yet; silent, as OK was before REQ-CHANSVC-001. */
    if ((status != RTE_STATUS_OK) && (status != RTE_STATUS_TIMEOUT) && (status != RTE_STATUS_NOT_CONNECTED))
    {
        RTE_EXAMPLE_LOG("[GW %s] TrainHandler: forward to TrainSim failed (%s)\n",
                             site_config_name(ctx->site), rte_status_to_string(status));
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), 0U, "GW", "Train", "ENVELOPE",
                              rte_status_to_string(status), NULL);
    }
}
 
void gateway_c_train_handler_execute(gateway_c_context_t *ctx, uint32_t cycle)
{
    /* Sized for the widest frame this relay carries now
     * (RBC_MSG_ERTMS_ENVELOPE, RBC_TRAIN_ENVELOPE_WIRE_SIZE) - the flat
     * scheme's own smaller kinds (still RBC_ENVELOPE_WIRE_SIZE), and any
     * future SafeComm SaF/SAI frame smaller than that, all ride the same
     * buffer. Read via rte_channel_service_read_ex(): the underlying
     * transport is a datagram (rte_netlink UDP) that already preserves
     * message boundaries, and _ex reports the true received length instead
     * of always claiming the full buffer - this function forwards exactly
     * that many bytes, never the whole buffer padded with stale/zeroed
     * tail. Zeroed before every read below (not just once here) so the
     * unused tail, when a frame IS shorter, is always deterministic zero,
     * never a previous frame's leftover bytes, should anything downstream
     * ever look past actual_len. */
    uint8_t       bytes[RBC_TRAIN_ENVELOPE_WIRE_SIZE];
    size_t        actual_len;
    char          sim_status[RBC_UTIL_PAYLOAD_MAX + 1];
    rte_status_t status;
    bool          forwarded_this_cycle = false;
    (void)cycle;
 
    /* TrainSim -> Gateway -> A and B (verbatim broadcast, no train_id
     * inspection - A/B route each envelope by its own train_id). */
    rte_mem_set(bytes, 0, sizeof(bytes));
    while (rte_channel_service_read_ex(&g_train.sim_channel, bytes, sizeof(bytes), 0U, &actual_len) ==
           RTE_STATUS_OK)
    {
        /* ADR-038: the Train sim's own-perspective status frame - republish
         * it to the C status broker, never forward it to A/B (they never
         * see this kind). Not counted as relay traffic (the idle-cycle
         * keepalive below still fires if this was the only frame). */
        if (rbc_envelope_is_sim_status(bytes, sim_status, (uint16_t)sizeof(sim_status)))
        {
            gateway_c_selfstatus_publish((uint8_t)RBC_UTIL_SRC_SIM_TRAIN, sim_status);
            rte_mem_set(bytes, 0, sizeof(bytes));
            continue;
        }
        send_to_relay(ctx, GATEWAY_C_PEER_A, bytes, actual_len);
        send_to_relay(ctx, GATEWAY_C_PEER_B, bytes, actual_len);
        forwarded_this_cycle = true;
        rte_mem_set(bytes, 0, sizeof(bytes));
    }
 
    /* ADR-036: with ONE multiplexed link, an idle Train sim (M136 every
     * simPeriodSeconds, ~2s) leaves the C->A/B relay direction silent for
     * longer than RTE_EXAMPLE_LINK_STALE_TIMEOUT_COUNT *
     * _LINK_TIMEOUT_MS (~1.2s) - A/B's relay-RX staleness detector then
     * closes the link for reconnect, cross-compare skips, and the A<->B
     * checkpoint rendezvous times out -> reboot. Keepalive the relay
     * every idle cycle, exactly as A/B already do C-ward for IL/CTC
     * (post_execute.c / rbc_wire_types.h RELAY_KEEPALIVE). A/B recognise
     * and drop it before any train decode (pre_execute.c). */
    if (!forwarded_this_cycle)
    {
        uint8_t keepalive[RBC_ENVELOPE_WIRE_SIZE];
 
        rbc_envelope_write_keepalive(keepalive);
        send_to_relay(ctx, GATEWAY_C_PEER_A, keepalive, sizeof(keepalive));
        send_to_relay(ctx, GATEWAY_C_PEER_B, keepalive, sizeof(keepalive));
    }
 
    /* A/B -> Gateway -> TrainSim.
     * RCA/OCORA PI-API: Gateway acts as external Service Unit / Adapter.
     * 2oo2 arbitration is not needed here because channels A and B already
     * compared data before emission. Gateway forwards whichever envelope arrives
     * first and deduplicates identical frames arriving from the peer replica. */
    {
        uint32_t a_count = 0U;
        uint32_t b_count = 0U;
 
        rte_mem_set(bytes, 0, sizeof(bytes));
        while ((status = rte_channel_service_read_ex(&g_train.relay_channel[GATEWAY_C_PEER_A],
                                                  bytes, sizeof(bytes), 0U, &actual_len)) == RTE_STATUS_OK)
        {
            if (!gateway_c_dedup_check_and_record(&g_train.dedup, bytes, actual_len))
            {
                send_to_sim(ctx, bytes, actual_len);
            }
            a_count++;
            rte_mem_set(bytes, 0, sizeof(bytes));
        }
        if ((status != RTE_STATUS_OK) && (status != RTE_STATUS_TIMEOUT))
        {
            if (g_train.relay_down_since_ms[GATEWAY_C_PEER_A] == 0U)
            {
                (void)rte_timer_now(&g_train.relay_down_since_ms[GATEWAY_C_PEER_A]);
            }
        }
        else if (a_count > 0U)
        {
            g_train.relay_down_since_ms[GATEWAY_C_PEER_A] = 0U;
            g_train.relay_consecutive_timeout = 0U;
            g_train.a_stale_cycles = 0U;
        }
        else
        {
            g_train.a_stale_cycles++;
        }
        g_train.a_seen_this_cycle = a_count;
 
        rte_mem_set(bytes, 0, sizeof(bytes));
        while ((status = rte_channel_service_read_ex(&g_train.relay_channel[GATEWAY_C_PEER_B],
                                                  bytes, sizeof(bytes), 0U, &actual_len)) == RTE_STATUS_OK)
        {
            if (!gateway_c_dedup_check_and_record(&g_train.dedup, bytes, actual_len))
            {
                send_to_sim(ctx, bytes, actual_len);
            }
            b_count++;
            rte_mem_set(bytes, 0, sizeof(bytes));
        }
        if ((status != RTE_STATUS_OK) && (status != RTE_STATUS_TIMEOUT))
        {
            if (g_train.relay_down_since_ms[GATEWAY_C_PEER_B] == 0U)
            {
                (void)rte_timer_now(&g_train.relay_down_since_ms[GATEWAY_C_PEER_B]);
            }
        }
        else if (b_count > 0U)
        {
            g_train.relay_down_since_ms[GATEWAY_C_PEER_B] = 0U;
        }
 
        g_train.last_known_tag = ((a_count > 0U) || (b_count > 0U)) ? "ACTIVE" : "SUPPRESSED";
    }
}
 
void gateway_c_train_handler_shutdown(void)
{
    (void)rte_channel_service_close(&g_train.sim_channel);
    (void)rte_channel_service_close(&g_train.relay_channel[GATEWAY_C_PEER_A]);
    (void)rte_channel_service_close(&g_train.relay_channel[GATEWAY_C_PEER_B]);
}
 
bool gateway_c_train_handler_check_link_down(rte_timestamp_ms_t threshold_ms, rte_timestamp_ms_t now_ms,
                                             char *out_which, size_t which_size)
{
    gateway_c_peer_t peer;
 
    for (peer = GATEWAY_C_PEER_A; peer <= GATEWAY_C_PEER_B; peer = (gateway_c_peer_t)((int)peer + 1))
    {
        rte_timestamp_ms_t down_since = g_train.relay_down_since_ms[peer];
        rte_timestamp_ms_t send_down_since = g_train.send_down_since_ms[peer];
 
        if (((down_since != 0U) && ((now_ms - down_since) >= threshold_ms)) ||
            ((send_down_since != 0U) && ((now_ms - send_down_since) >= threshold_ms)))
        {
            (void)snprintf(out_which, which_size, "Train relay link to %s", peer_name(peer));
            return true;
        }
    }
    return false;
}