gateway_c_ctc_handler.c

Go to the documentation of this file.

#include "gateway_c_ctc_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_ctc_handler_state_t g_ctc;
 
static const char *peer_name(gateway_c_peer_t peer)
{
    return (peer == GATEWAY_C_PEER_A) ? "A" : "B";
}
 
rte_status_t gateway_c_ctc_handler_init(gateway_c_context_t *ctx)
{
    rte_status_t status;
 
    g_ctc.last_known_tag = "SUPPRESSED";
    g_ctc.relay_consecutive_timeout = 0U;
    g_ctc.relay_down_since_ms[GATEWAY_C_PEER_A] = 0U;
    g_ctc.relay_down_since_ms[GATEWAY_C_PEER_B] = 0U;
    g_ctc.send_down_since_ms[GATEWAY_C_PEER_A] = 0U;
    g_ctc.send_down_since_ms[GATEWAY_C_PEER_B] = 0U;
    g_ctc.a_seen_this_cycle = 0U;
    g_ctc.a_stale_cycles = 0U;
    gateway_c_dedup_init(&g_ctc.dedup);
 
    /* Relay links to/from A and B */
    status = rte_channel_service_setup(&g_ctc.relay_channel[GATEWAY_C_PEER_A], "c-relay-ctc-a");
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[GW %s] CTCHandler: relay A channel setup failed (%s)\n",
                             site_config_name(ctx->site), rte_status_to_string(status));
        return status;
    }
 
    status = rte_channel_service_setup(&g_ctc.relay_channel[GATEWAY_C_PEER_B], "c-relay-ctc-b");
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[GW %s] CTCHandler: relay B channel setup failed (%s)\n",
                             site_config_name(ctx->site), rte_status_to_string(status));
        return status;
    }
 
    /* CTCSim-facing link */
    status = rte_channel_service_setup(&g_ctc.sim_channel, "c-sim-ctc");
    if (status != RTE_STATUS_OK)
    {
        RTE_EXAMPLE_LOG("[GW %s] CTCHandler: sim channel setup failed (%s)\n",
                             site_config_name(ctx->site), rte_status_to_string(status));
        return status;
    }
 
    return RTE_STATUS_OK;
}
 
static void send_to_relay(gateway_c_context_t *ctx, gateway_c_peer_t peer, const uint8_t bytes[RBC_ENVELOPE_WIRE_SIZE])
{
    rte_status_t status = rte_channel_service_send(&g_ctc.relay_channel[peer],
                                                     bytes, RBC_ENVELOPE_WIRE_SIZE, 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))
    {
        /* UDP gives no read error when the peer is partitioned, so a failed send is the only link-down
         * signal; a successful send ends the window (see gateway_c_train_handler.c in RBC_GP). */
        g_ctc.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_ctc.send_down_since_ms[peer] == 0U)
        {
            (void)rte_timer_now(&g_ctc.send_down_since_ms[peer]);
        }
        if (status == RTE_STATUS_TIMEOUT)
        {
            return;
        }
        RTE_EXAMPLE_LOG("[GW %s] CTCHandler: keepalive 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[RBC_ENVELOPE_WIRE_SIZE])
{
    rte_status_t status = rte_channel_service_send(&g_ctc.sim_channel,
                                                     bytes, RBC_ENVELOPE_WIRE_SIZE, 0U);
    /* NOT_CONNECTED: no CTC 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] CTCHandler: forward to CTCSim 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", "CTC", "ENVELOPE",
                              rte_status_to_string(status), NULL);
    }
}
 
void gateway_c_ctc_handler_execute(gateway_c_context_t *ctx, uint32_t cycle)
{
    uint8_t       bytes[RBC_ENVELOPE_WIRE_SIZE];
    char          sim_status[RBC_UTIL_PAYLOAD_MAX + 1];
    rte_status_t status;
    (void)cycle;
 
    /* CTCSim -> Gateway -> A/B */
    while (rte_channel_service_read(&g_ctc.sim_channel, bytes, sizeof(bytes), 0U) == RTE_STATUS_OK)
    {
        /* ADR-038: the CTC sim's own-perspective status frame - republish to
         * the C status broker, never forward to A/B (they never see this
         * kind). */
        if (rbc_envelope_is_sim_status(bytes, sim_status, (uint16_t)sizeof(sim_status)))
        {
            gateway_c_selfstatus_publish((uint8_t)RBC_UTIL_SRC_SIM_CTC, sim_status);
            continue;
        }
        send_to_relay(ctx, GATEWAY_C_PEER_A, bytes);
        send_to_relay(ctx, GATEWAY_C_PEER_B, bytes);
    }
 
    /* A/B -> Gateway -> CTCSim.
     * 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;
 
        while ((status = rte_channel_service_read(&g_ctc.relay_channel[GATEWAY_C_PEER_A],
                                                  bytes, sizeof(bytes), 0U)) == RTE_STATUS_OK)
        {
            if (rbc_envelope_is_keepalive(bytes))
            {
                a_count++;
                continue;
            }
            if (!gateway_c_dedup_check_and_record(&g_ctc.dedup, bytes, sizeof(bytes)))
            {
                send_to_sim(ctx, bytes);
            }
            a_count++;
        }
        if ((status != RTE_STATUS_OK) && (status != RTE_STATUS_TIMEOUT))
        {
            if (g_ctc.relay_down_since_ms[GATEWAY_C_PEER_A] == 0U)
            {
                (void)rte_timer_now(&g_ctc.relay_down_since_ms[GATEWAY_C_PEER_A]);
            }
        }
        else if (a_count > 0U)
        {
            g_ctc.relay_down_since_ms[GATEWAY_C_PEER_A] = 0U;
            g_ctc.relay_consecutive_timeout = 0U;
            g_ctc.a_stale_cycles = 0U;
        }
        else
        {
            g_ctc.a_stale_cycles++;
        }
        g_ctc.a_seen_this_cycle = a_count;
 
        while ((status = rte_channel_service_read(&g_ctc.relay_channel[GATEWAY_C_PEER_B],
                                                  bytes, sizeof(bytes), 0U)) == RTE_STATUS_OK)
        {
            if (rbc_envelope_is_keepalive(bytes))
            {
                b_count++;
                continue;
            }
            if (!gateway_c_dedup_check_and_record(&g_ctc.dedup, bytes, sizeof(bytes)))
            {
                send_to_sim(ctx, bytes);
            }
            b_count++;
        }
        if ((status != RTE_STATUS_OK) && (status != RTE_STATUS_TIMEOUT))
        {
            if (g_ctc.relay_down_since_ms[GATEWAY_C_PEER_B] == 0U)
            {
                (void)rte_timer_now(&g_ctc.relay_down_since_ms[GATEWAY_C_PEER_B]);
            }
        }
        else if (b_count > 0U)
        {
            g_ctc.relay_down_since_ms[GATEWAY_C_PEER_B] = 0U;
        }
 
        g_ctc.last_known_tag = ((a_count > 0U) || (b_count > 0U)) ? "ACTIVE" : "SUPPRESSED";
    }
}
 
void gateway_c_ctc_handler_shutdown(void)
{
    (void)rte_channel_service_close(&g_ctc.sim_channel);
    (void)rte_channel_service_close(&g_ctc.relay_channel[GATEWAY_C_PEER_A]);
    (void)rte_channel_service_close(&g_ctc.relay_channel[GATEWAY_C_PEER_B]);
}
 
bool gateway_c_ctc_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_ctc.relay_down_since_ms[peer];
        rte_timestamp_ms_t send_down_since = g_ctc.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, "CTC relay link to %s", peer_name(peer));
            return true;
        }
    }
    return false;
}