ab_gp_channel_crosscompare.c

Go to the documentation of this file.

#include "ab_gp_channel_crosscompare.h"
#include "ab_gp_standby_policy.h"
 
#include <stdio.h>
#include <stdlib.h>
 
#include "rte/oal/log/rte_log.h"
#include "rte/oal/memory/rte_mem_util.h"
#include "rte/utils/safestate/rte_safestate.h"
 
#include "../../common/common_config.h"
#include "ab_gp_channel.h"
#include "ab_gp_channel_negotiate.h"
#include "ab_gp_channel_payload.h"

#define XCOMPARE_PAYLOAD_SIZE ((size_t)RTE_EXAMPLE_MAX_TRAINS * RTE_EXAMPLE_TRAIN_SESSION_WIRE_SIZE)

#define RTE_EXAMPLE_XCOMPARE_FAULT_CYCLE_ENV "RTE_EXAMPLE_XCOMPARE_FAULT_CYCLE"

static uint32_t xcompare_fault_cycle(void)
{
    static bool     parsed = false;
    static uint32_t cycle  = UINT32_MAX;
 
    if (!parsed)
    {
        const char *env = getenv(RTE_EXAMPLE_XCOMPARE_FAULT_CYCLE_ENV);
 
        if (env != NULL)
        {
            char *end = NULL;
            long  val = strtol(env, &end, 10);
 
            if ((end != env) && (val >= 0))
            {
                cycle = (uint32_t)val;
            }
        }
        parsed = true;
    }
    return cycle;
}

static void encode_local_payload(const ab_gp_channel_context_t *ctx, uint8_t out[XCOMPARE_PAYLOAD_SIZE])
{
    ab_gp_channel_payload_encode_collector(ab_gp_train_payload_collector_id(), out, ctx->sessions);
    if (ctx->is_role_a && (ctx->cyc_cycle >= xcompare_fault_cycle()))
    {
        out[0] ^= 0xFFU;
        RTE_EXAMPLE_LOG("[%s] cycle %u: RTE_EXAMPLE_XCOMPARE_FAULT_CYCLE reached - "
                             "corrupting own cross-compare payload (test-only fault injection)\n",
                             ctx->role_tag, ctx->cyc_cycle);
    }
}

static void encode_peer_payload(const ab_gp_channel_context_t *ctx, uint8_t out[XCOMPARE_PAYLOAD_SIZE])
{
    ab_gp_channel_payload_encode_collector(ab_gp_train_payload_collector_id(), out, ctx->cyc_peer_sessions);
}

static void fault_cleanup(ab_gp_channel_context_t *ctx, rte_safestate_reason_t reason)
{
    ctx->own_faulted = true;
    /* Tell this channel's own inter-site counterpart about the fault
     * BEFORE the link closes below - see
     * ab_gp_channel_negotiate_flush_faulted()'s own doc for why this is
     * required (not just a nice-to-have) for the counterpart's own
     * promotion rule to ever fire. */
    ab_gp_channel_negotiate_flush_faulted(ctx);
    /* Same "before the link closes" requirement applies to CTC: a
     * dispatcher needs to see this fault too, not just A/B's own inter-
     * site counterpart - see ab_gp_channel_raise_ctc_alarm()'s own doc. */
    ab_gp_channel_raise_ctc_alarm(ctx, RBC_ALARM_SEVERITY_ERROR, (uint32_t)reason, "reboot fault detected");
    /* Tear this process's own links/tasks/timer/watchdog down cleanly
     * BEFORE re-execing: RTE_SAFESTATE_LEVEL_REBOOT (entered either by
     * react_to_fault() below or by the cross-comparator itself) hands
     * off to main.c's handler, which re-execs via execv() - POSIX
     * execv() does NOT close inherited file descriptors on its own, so
     * skipping this would leak sockets across the reboot AND, worse,
     * mean the peer/C side never observes a real disconnect.
     * ab_gp_channel_shutdown() is safe to call from here (main thread,
     * not one of the background tasks - no self-join deadlock). */
    (void)ab_gp_channel_shutdown(ctx);
}

static void react_to_fault(ab_gp_channel_context_t *ctx, rte_safestate_reason_t reason)
{
    fault_cleanup(ctx, reason);
    RTE_SAFESTATE(RTE_SAFESTATE_LEVEL_REBOOT, reason);
}

static int first_disagreeing_train_index(const ab_gp_channel_context_t *ctx)
{
    uint8_t local_bytes[RTE_EXAMPLE_MAX_TRAINS][RTE_EXAMPLE_TRAIN_SESSION_WIRE_SIZE];
    uint8_t peer_bytes[RTE_EXAMPLE_MAX_TRAINS][RTE_EXAMPLE_TRAIN_SESSION_WIRE_SIZE];
    uint8_t i;
 
    ab_gp_channel_payload_encode_collector(ab_gp_train_payload_collector_id(), &local_bytes[0][0], ctx->sessions);
    ab_gp_channel_payload_encode_collector(ab_gp_train_payload_collector_id(), &peer_bytes[0][0], ctx->cyc_peer_sessions);
    for (i = 0U; i < (uint8_t)RTE_EXAMPLE_MAX_TRAINS; i++)
    {
        if (rte_mem_compare(local_bytes[i], peer_bytes[i], RTE_EXAMPLE_TRAIN_SESSION_WIRE_SIZE) != 0)
        {
            return (int)i;
        }
    }
    return -1;
}
 
static void xcompare_on_disagreement(void *context, rte_voting_result_t result)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)context;
    uint32_t cycle = ctx->cyc_cycle;
 
    if (result != RTE_VOTING_DISAGREED)
    {
        /* Defensive only: both xcompare_recv_local()/xcompare_recv_peer()
         * above always succeed immediately (no real I/O, no timeout), so
         * TIMEOUT/INSUFFICIENT_QUORUM can never actually occur through
         * this pair of channels - but on_disagreement() itself is
         * invoked for every non-AGREED result, so this guard keeps that
         * contract explicit rather than assuming it. */
        RTE_EXAMPLE_LOG("[%s] cycle %u: unexpected cross-compare result %d (not DISAGREED) - ignoring\n", ctx->role_tag,
               cycle, (int)result);
        rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), cycle, ctx->role_tag,
                              ctx->is_role_a ? "B" : "A", "XCOMPARE", "unexpected non-DISAGREED result from comparator",
                              NULL);
        return;
    }
 
    {
        int bad_index = first_disagreeing_train_index(ctx);
 
        RTE_EXAMPLE_LOG("[%s] cycle %u: DISAGREE (first differing train index=%d) - this site's role -> FAULTED, entering REBOOT\n",
               ctx->role_tag, cycle, bad_index);
        {
            char extra[32];
            (void)snprintf(extra, sizeof(extra), "train_index=%d", bad_index);
            rte_log_write_event(RTE_LOG_LEVEL_ERROR, site_config_name(ctx->site), cycle, ctx->role_tag,
                                  ctx->is_role_a ? "B" : "A", "DISAGREE", "FAULTED, entering REBOOT", extra);
        }
    }
    /* Cleanup only - the cross-comparator itself enters REBOOT right
     * after this callback returns, see this function's own doc above. */
    fault_cleanup(ctx, RTE_EXAMPLE_REASON_AB_MTYPE_DISAGREE);
}
 
rte_status_t ab_gp_channel_crosscompare_init(ab_gp_channel_context_t *ctx)
{
    rte_cross_comparator_config_t xcompare_cfg;
 
    rte_mem_set(&xcompare_cfg, 0, sizeof(xcompare_cfg));
    xcompare_cfg.log_disagreements = false; /* xcompare_on_disagreement() logs its own, richer detail */
    /* REBOOT, not SAFE - see this file's own header. The comparator
     * itself now unconditionally enters this level on DISAGREE, right
     * after calling xcompare_on_disagreement() below for this
     * application's own pre-transition cleanup (RCA/OCORA PI-API
     * compatibility - see rte_cross_comparator.h's own note). */
    xcompare_cfg.safestate_level = RTE_SAFESTATE_LEVEL_REBOOT;
    xcompare_cfg.safestate_reason = RTE_EXAMPLE_REASON_AB_MTYPE_DISAGREE;
    xcompare_cfg.on_disagreement = xcompare_on_disagreement;
    xcompare_cfg.disagreement_context = ctx;
 
    /* RCA/OCORA Phase 2b: no rte_channel_t/rte_cross_comparator_register_channel()
     * calls needed any more - ab_gp_channel_crosscompare_execute() below
     * calls rte_cross_comparator_execute_buffers() directly with the
     * local/peer bytes already in hand. */
    return rte_cross_comparator_init(&ctx->xcompare_cmp, &xcompare_cfg);
}

static bool per_train_readiness_gate(ab_gp_channel_context_t *ctx, uint32_t cycle)
{
    uint8_t i;
 
    for (i = 0U; i < (uint8_t)RTE_EXAMPLE_MAX_TRAINS; i++)
    {
        const ab_gp_train_session_t *local_session = &ctx->sessions[i];
        const ab_gp_train_session_t *peer_session = &ctx->cyc_peer_sessions[i];
        bool fields_match;
 
        if ((!local_session->in_use) && (!peer_session->in_use))
        {
            ctx->xcompare_skip_count[i] = 0U;
            continue;
        }
        /* The sibling's sample is one cycle old, so a train connected or disconnected in this very cycle shows up as "held
         * by the sibling only" for a moment. Adopt only when the same picture has lasted two cycles in a row, otherwise a
         * session the sibling is just dropping is resurrected here and the two tables diverge (ISS-016). */
        if ((!local_session->in_use) && peer_session->in_use)
        {
            if (ctx->adopt_pending[i] < UINT8_MAX)
            {
                ctx->adopt_pending[i]++;
            }
        }
        else
        {
            ctx->adopt_pending[i] = 0U;
        }
        if ((ctx->adopt_pending[i] >= 2U) &&
            ab_gp_session_adopt_allowed(ctx->join_window_cycles_left, local_session->in_use, peer_session->in_use))
        {
            /* This channel started receiving later than its A/B sibling and missed this train's connect: take the
             * sibling's session once (see ab_gp_session_adopt_allowed()); later cycles are computed independently
             * and cross-compared as usual. */
            rte_mem_copy(&ctx->sessions[i], peer_session, sizeof(ctx->sessions[i]));
            ctx->xcompare_skip_count[i] = 0U;
            RTE_EXAMPLE_LOG("[%s] cycle %u: train index %u missing here but held by the A/B sibling - adopted its session "
                            "(joined late)\n", ctx->role_tag, cycle, (unsigned int)i);
            rte_log_write_event(RTE_LOG_LEVEL_WARNING, site_config_name(ctx->site), cycle, ctx->role_tag,
                                  ctx->is_role_a ? "B" : "A", "XCOMPARE", "adopted a missed train session from the sibling", NULL);
            continue;
        }
        /* Direct field comparison (not the raw struct via
         * rte_mem_compare() - ab_gp_train_session_t mixes bool/uint8_t/
         * uint32_t members, so the compiler is free to insert padding;
         * see this file's own doc above for the same reasoning applied
         * to the real comparator below). Compares every wire-relevant
         * field (matches ab_gp_train.c's own registered payload-collector
         * field list exactly) - ma_pending_send/ctc_*_sent are
         * deliberately excluded, same as the wire encoding itself. */
        fields_match = (local_session->in_use == peer_session->in_use)
                        && (local_session->train_id == peer_session->train_id)
                        && (local_session->cycle == peer_session->cycle)
                        && (local_session->d_lrbg == peer_session->d_lrbg)
                        && (local_session->granted_length == peer_session->granted_length)
                        && (local_session->ma_seq == peer_session->ma_seq)
                        && (local_session->ma_acked == peer_session->ma_acked);
 
        RTE_EXAMPLE_LOG("[%s] cycle %u: xcompare train index %u: local(id=%u,pos=%u,ma_len=%u,seq=%u,ack=%d) "
                            "vs peer(id=%u,pos=%u,ma_len=%u,seq=%u,ack=%d) -> %s\n",
                            ctx->role_tag, cycle, (unsigned int)i,
                            (unsigned int)local_session->train_id, (unsigned int)local_session->d_lrbg,
                            (unsigned int)local_session->granted_length, (unsigned int)local_session->ma_seq, (int)local_session->ma_acked,
                            (unsigned int)peer_session->train_id, (unsigned int)peer_session->d_lrbg,
                            (unsigned int)peer_session->granted_length, (unsigned int)peer_session->ma_seq, (int)peer_session->ma_acked,
                            fields_match ? "MATCH" : "DIFF");
 
        if (fields_match)
        {
            ctx->xcompare_skip_count[i] = 0U;
            continue;
        }
        if (ctx->xcompare_skip_count[i] < (uint8_t)RTE_EXAMPLE_XCOMPARE_SYNC_SKIP_LIMIT)
        {
            ctx->xcompare_skip_count[i]++;
            RTE_EXAMPLE_LOG("[%s] cycle %u: train index %u not yet synchronized with peer (skip %u/%u) - "
                                 "skipping cross-compare this cycle\n",
                                 ctx->role_tag, cycle, (unsigned int)i, (unsigned int)ctx->xcompare_skip_count[i],
                                 (unsigned int)RTE_EXAMPLE_XCOMPARE_SYNC_SKIP_LIMIT);
            rte_log_write_event(RTE_LOG_LEVEL_DEBUG, site_config_name(ctx->site), cycle, ctx->role_tag,
                                  ctx->is_role_a ? "B" : "A", "XCOMPARE",
                                  "train not yet synchronized with peer - skipping cross-compare", NULL);
            return false;
        }
        RTE_EXAMPLE_LOG("[%s] cycle %u: train index %u still not synchronized with peer after %u skips - "
                             "treating as a genuine divergence, not a timing race\n",
                             ctx->role_tag, cycle, (unsigned int)i,
                             (unsigned int)RTE_EXAMPLE_XCOMPARE_SYNC_SKIP_LIMIT);
        /* fall through - the real comparator below will correctly DISAGREE on this */
    }
    return true;
}

static void execute_cross_comparator(ab_gp_channel_context_t *ctx, uint32_t cycle)
{
    uint8_t local_bytes[XCOMPARE_PAYLOAD_SIZE];
    uint8_t peer_bytes[XCOMPARE_PAYLOAD_SIZE];
    uint8_t agreed_bytes[XCOMPARE_PAYLOAD_SIZE] = { 0 };
    size_t agreed_size = 0U;
    rte_voting_result_t result = RTE_VOTING_AGREED;
    rte_status_t status;
 
    encode_local_payload(ctx, local_bytes);
    encode_peer_payload(ctx, peer_bytes);
    status = rte_cross_comparator_execute_buffers(&ctx->xcompare_cmp, local_bytes, peer_bytes,
                                                    XCOMPARE_PAYLOAD_SIZE, &result, agreed_bytes, &agreed_size);
 
    if (status == RTE_STATUS_OK)
    {
        ctx->cyc_agree = true;
        RTE_EXAMPLE_LOG("[%s] cycle %u: cross-compare consensus result: AGREE (%u train session(s))\n", ctx->role_tag, cycle,
                             (unsigned int)RTE_EXAMPLE_MAX_TRAINS);
        rte_log_write_event(RTE_LOG_LEVEL_INFO, site_config_name(ctx->site), cycle, ctx->role_tag,
                              ctx->is_role_a ? "B" : "A", "AGREE", "sessions match", NULL);
    }
}
 
rte_status_t ab_gp_channel_crosscompare_execute(ab_gp_channel_context_t *ctx)
{
    uint32_t cycle = ctx->cyc_cycle;
    bool have_peer_data = ctx->cyc_have_peer_data;
    uint32_t peer_cycle = ctx->cyc_peer_cycle;
    bool crc_ok = ctx->cyc_crc_ok;
 
    ctx->cyc_agree = false;
 
    /* ADR-029: gated on ctx->is_online (a STANDBY pair has nothing
     * meaningful in its own empty session table to compare), not
     * ctx->cyc_have_channel_data - AB_SAMPLE is now sent every cycle
     * while ONLINE regardless of whether fresh train/IL data arrived
     * THIS specific cycle (see ab_gp_channel_send_peer_sample_if_online()'s own doc), so
     * cross-compare runs every ONLINE cycle too, continuously
     * reconfirming the current session table rather than only reacting
     * to fresh arrivals - same "always every cycle" philosophy the
     * PEER_MSG_KIND_SITE_STATE cross-check already used. */
    if ((!ctx->is_online) || (!have_peer_data))
    {
        RTE_EXAMPLE_LOG("[%s] cycle %u: not online and/or no peer data yet - skipping cross-compare this cycle\n",
                             ctx->role_tag, cycle);
        rte_log_write_event(RTE_LOG_LEVEL_DEBUG, site_config_name(ctx->site), cycle, ctx->role_tag,
                              ctx->is_role_a ? "B" : "A", "XCOMPARE", "not online/no peer data yet this cycle - skipping cross-compare",
                              NULL);
        return RTE_STATUS_OK;
    }
    if (!crc_ok)
    {
        /* Malformed (not CRC-invalid - ab_gp_channel_payload_decode_ab_sample()
         * already returned INVALID_PARAM for a length/field problem, not
         * a corruption fault, which would have already diverted via
         * RTE_SAFESTATE in ab_gp_channel.c's
         * ab_gp_channel_read_and_stash_peer_sample()) - nothing trustworthy
         * to compare. */
        return RTE_STATUS_OK;
    }
    if (((peer_cycle > cycle) ? (peer_cycle - cycle) : (cycle - peer_cycle)) > RTE_EXAMPLE_AB_CYCLE_SKEW_TOLERANCE)
    {
        /* Peer's report cycle number is too far from this cycle's own to
         * be considered comparable - independently free-running 500ms
         * timers can legitimately be a few cycles apart; beyond the
         * tolerance, though, this is stale enough not to compare, not a
         * fault. See common_config.h's RTE_EXAMPLE_AB_CYCLE_SKEW_TOLERANCE
         * comment. */
        RTE_EXAMPLE_LOG("[%s] cycle %u: peer answered for a too-different cycle (%u) - skipping cross-compare this cycle\n",
               ctx->role_tag, cycle, peer_cycle);
        {
            char extra[24];
            (void)snprintf(extra, sizeof(extra), "PeerCycle=%u", peer_cycle);
            rte_log_write_event(RTE_LOG_LEVEL_DEBUG, site_config_name(ctx->site), cycle, ctx->role_tag,
                                  ctx->is_role_a ? "B" : "A", "XCOMPARE",
                                  "peer answered for a too-different cycle - skipping cross-compare", extra);
        }
        /* A persistent skew is not a timing race: the two channels no longer run in step (found live after a
         * long run: A and B differed by ~700 cycles, so the 2oo2 comparison was silently off for good). Without a
         * comparison the pair is not 2oo2, so treat it as a fault; the restart of both channels realigns them. */
        ctx->xcompare_skew_skips++;
        if (ctx->xcompare_skew_skips >= 60U)
        {
            RTE_EXAMPLE_LOG("[%s] cycle %u: peer cycle number off by more than the tolerance for %u cycles - "
                                 "cross-compare impossible, entering REBOOT\n",
                                 ctx->role_tag, cycle, (unsigned int)ctx->xcompare_skew_skips);
            react_to_fault(ctx, RTE_EXAMPLE_REASON_AB_PEER_LINK_DOWN);
        }
        return RTE_STATUS_OK;
    }
    ctx->xcompare_skew_skips = 0U;
 
    /* ADR-029: per-train readiness gate - see per_train_readiness_gate()'s own doc. */
    if (!per_train_readiness_gate(ctx, cycle))
    {
        return RTE_STATUS_OK;
    }
 
    /* Comparable this cycle - see execute_cross_comparator()'s own doc. */
    execute_cross_comparator(ctx, cycle);
 
    return RTE_STATUS_OK;
}
 
void ab_gp_channel_crosscompare_check_site_state(ab_gp_channel_context_t *ctx)
{
    /* The agreement rule, its count and the NEGOTIATION_MISMATCH log live in rte_site_role (REQ-SITEROLE-014/016,
     * ADR-040 step 5b): the service took the sibling's frame through the sibling source in this cycle's
     * rte_site_role_execute() and ran the check; GP keeps the fault reaction. */
    if (rte_site_role_sibling_verdict(&ctx->site_role) == RTE_SITE_ROLE_SIBLING_FAULT)
    {
        react_to_fault(ctx, RTE_EXAMPLE_REASON_AB_SITE_STATE_MISMATCH);
    }
}