post_execute.c

Go to the documentation of this file.

 
#include "ab_gp_channel_cycle_hooks.h"
 
#include <stdbool.h>
#include <stdint.h>
 
#include "rte/app/appmanager/rte_appmanager.h"
#include "rte/oal/log/rte_log.h"
#include "rte/redundancy/watchdog/rte_watchdog.h"
 
#include "../../common/common_config.h"
#include "../../common/site_config.h"
#include "../com/ab_gp_channel.h"
#include "../com/ab_gp_channel_negotiate.h"
#include "../com/ab_gp_site_status.h"
#include "../com/ab_gp_channel_stage.h"
#include "../com/ab_gp_channel_types.h"
#include "../com/ab_gp_channel_payload.h"
#include "../com/ab_gp_status.h"
#include "../train/ab_gp_train.h"
#include "ab_gp_ga_hooks.h" /* ab_gp_drain_ctc_outputs()/_il_outputs() - see that file's own doc */

static void stage_peer_sample_if_online(ab_gp_channel_context_t *ctx, ab_gp_channel_handle_t handle, uint32_t cycle)
{
    if (ctx->is_online)
    {
        uint8_t out_bytes[PEER_WIRE_FRAME_SIZE] = { 0 };
        const void *sources[AB_GP_CHANNEL_PAYLOAD_MAX_COLLECTORS] = { NULL };
        rte_status_t stage_status;
 
        sources[ab_gp_train_payload_collector_id()] = ctx->sessions;
        out_bytes[0] = PEER_MSG_KIND_AB_SAMPLE;
        ab_gp_channel_payload_encode_ab_sample(&out_bytes[1], cycle, sources);
 
        RTE_CHECKPOINT_MARK();
        stage_status = ab_gp_channel_stage_crosscompare(handle, PEER_MSG_KIND_AB_SAMPLE, &out_bytes[1], sizeof(out_bytes) - 1U);
        if (stage_status == RTE_STATUS_OK)
        {
            RTE_EXAMPLE_LOG("[%s] cycle %u: dual transfer staged sample (%zu bytes) for peer\n",
                                ctx->role_tag, cycle, sizeof(out_bytes));
            rte_log_write_event(RTE_LOG_LEVEL_DEBUG, site_config_name(ctx->site), cycle, ctx->role_tag,
                                  ctx->is_role_a ? "B" : "A", "DUAL_TRANSFER", "sample staged for peer", "size=343");
        }
    }
}

static void send_il_keepalive_if_online(ab_gp_channel_context_t *ctx, ab_gp_channel_handle_t handle)
{
    if (ctx->is_online)
    {
        uint8_t keepalive[RBC_ENVELOPE_WIRE_SIZE];
 
        /* ADR-036: one IL relay link now - one keepalive, not one per train. */
        rbc_envelope_write_keepalive(keepalive);
        (void)ab_gp_channel_send_relay(handle, RTE_EXAMPLE_RELAY_KIND_IL, 0U, keepalive, RBC_ENVELOPE_WIRE_SIZE);
    }
}

static void build_and_stage_outputs_to_c(ab_gp_channel_context_t *ctx, ab_gp_channel_handle_t handle, uint32_t cycle)
{
    ab_gp_train_pending_output_t train_out[AB_GP_TRAIN_MAX_PENDING_OUTPUTS];
    ab_gp_ctc_pending_output_t   ctc_out[AB_GP_CTC_MAX_PENDING_OUTPUTS];
    ab_gp_il_pending_output_t    il_out[AB_GP_IL_MAX_PENDING_OUTPUTS];
    uint8_t                   train_count;
    uint8_t                   ctc_count;
    uint8_t                   il_count;
    uint8_t                   i;
    uint8_t                   k;
    bool                       should_forward_to_c;
 
    train_count = ab_gp_train_build_outputs(ctx->sessions, ctx->cyc_msg_type_due, ctx->cyc_position_reply_due,
                                          train_out);
    ctc_count = ab_gp_drain_ctc_outputs(ctc_out);
    il_count = ab_gp_drain_il_outputs(il_out);
 
    should_forward_to_c = (ctx->cyc_agree) || (ctx->single_mode && ctx->cyc_have_channel_data);
    if (!should_forward_to_c)
    {
        return;
    }
 
    /* Pending-output flags are only consumed HERE, once this cycle's
     * built outputs are actually being staged for C - not
     * unconditionally after building them above. A real, found-live
     * bug: cyc_agree reflects the PEER's own confirmation of this
     * channel's decision, which inherently lags by at least one
     * cycle (the dual-transfer sample staged this cycle is only
     * received/compared by the peer on a LATER cycle - see
     * xcompare_recv_peer()'s own doc, ab_gp_channel_crosscompare.c).
     * So the very first cycle a brand-new decision is computed (a
     * fresh MA grant or extend, ma_pending_send just set true this
     * same cycle by ab_ga_il_on_route_context_changed()), cyc_agree
     * can legitimately still be false - the built M3 was discarded
     * below (should_forward_to_c false), yet the flag that would
     * have triggered a REBUILD next cycle was already cleared,
     * silently losing that Movement Authority forever. Resetting
     * only on the cycle that actually forwards means an unforwarded
     * cycle's decision simply gets rebuilt (from whatever the
     * session's current state is - always correct, never stale) and
     * retried next cycle until it genuinely reaches C. */
    for (i = 0U; i < (uint8_t)RTE_EXAMPLE_MAX_TRAINS; i++)
    {
        ctx->sessions[i].ma_pending_send = false;
        ctx->sessions[i].position_reply_pending = false;
        /* ctc_ma_notify_pending: ab_ga_ctc_build_outputs() (just
         * staged as part of ctc_out[] above, via ctc_count) already
         * consumed this cycle's real Subset-026/ERTMS MA-grant
         * notification if it was set - see ab_gp_train_session_t's
         * own doc on why this is a separate flag from
         * ma_pending_send. Reset it here in the same place/manner
         * ma_pending_send itself is reset, or it would stay true
         * forever and CTC would never see ANOTHER grant/extend for
         * this train (ctc_ma_granted_sent latches true after the
         * first one, so nothing else would re-arm this path). */
        ctx->sessions[i].ctc_ma_notify_pending = false;
    }
    /* Same reasoning as the flag resets just above, for IL's/CTC's
     * own pending-output queues - those live in GA's private module
     * statics, unreachable from here directly, hence these confirm
     * callbacks instead of a direct reset. See
     * ab_ga_il_confirm_outputs_sent()/ab_ga_ctc_confirm_outputs_sent()'s
     * own doc for the full history (a real, previously-diagnosed-but-
     * not-fully-fixed instance of this exact bug class). */
    ab_gp_confirm_il_outputs_sent();
    ab_gp_confirm_ctc_outputs_sent();
 
    for (k = 0U; k < train_count; k++)
    {
        RTE_CHECKPOINT_MARK();
        (void)ab_gp_channel_stage_relay(handle, RTE_EXAMPLE_RELAY_KIND_TRAIN, train_out[k].index,
                                       train_out[k].bytes, RBC_ENVELOPE_WIRE_SIZE);
    }
    for (k = 0U; k < ctc_count; k++)
    {
        RTE_CHECKPOINT_MARK();
        (void)ab_gp_channel_stage_relay(handle, RTE_EXAMPLE_RELAY_KIND_CTC, 0U, ctc_out[k].bytes,
                                       RBC_ENVELOPE_WIRE_SIZE);
    }
    for (k = 0U; k < il_count; k++)
    {
        RTE_CHECKPOINT_MARK();
        (void)ab_gp_channel_stage_relay(handle, RTE_EXAMPLE_RELAY_KIND_IL, il_out[k].index, il_out[k].bytes,
                                       RBC_ENVELOPE_WIRE_SIZE);
    }
    if (ctc_count == 0U)
    {
        uint8_t keepalive[RBC_ENVELOPE_WIRE_SIZE];
 
        rbc_envelope_write_keepalive(keepalive);
        (void)ab_gp_channel_stage_relay(handle, RTE_EXAMPLE_RELAY_KIND_CTC, 0U, keepalive, RBC_ENVELOPE_WIRE_SIZE);
    }
 
    if (train_count > 0U || ctc_count > 0U || il_count > 0U)
    {
        char extra_payload[96];
        (void)snprintf(extra_payload, sizeof(extra_payload), "train_count=%u ctc_count=%u il_count=%u",
                       (unsigned int)train_count, (unsigned int)ctc_count, (unsigned int)il_count);
        RTE_EXAMPLE_LOG("[%s] cycle %u: staged outputs for C (train=%u, ctc=%u, il=%u)\n",
                             ctx->role_tag, cycle, (unsigned int)train_count, (unsigned int)ctc_count, (unsigned int)il_count);
        rte_log_write_event(RTE_LOG_LEVEL_DEBUG, site_config_name(ctx->site), cycle, ctx->role_tag, "C",
                             "OUTPUT_FORWARD", "staged outputs for C", extra_payload);
    }
}

static void send_site_state(ab_gp_channel_context_t *ctx, ab_gp_channel_handle_t handle)
{
    uint8_t frame[PEER_WIRE_FRAME_SIZE] = { 0 };
 
    frame[0] = PEER_MSG_KIND_SITE_STATE;
    frame[1] = ctx->is_online ? 1U : 0U;
    frame[2] = ctx->single_mode ? (uint8_t)RTE_EXAMPLE_SINGLE_MODE_FLAG_SINGLE : (uint8_t)RTE_EXAMPLE_SINGLE_MODE_FLAG_DUAL;
    frame[3] = ab_gp_site_status_ready_for_online(ctx) ? 1U : 0U;
    (void)ab_gp_channel_send_peer(handle, frame, sizeof(frame));
}
 
rte_status_t ab_gp_channel_post_execute(void *context)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)context;
    ab_gp_channel_handle_t handle = (ab_gp_channel_handle_t)ctx;
    uint32_t cycle;
 
    if (ctx == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    cycle = ctx->cyc_cycle;
 
    stage_peer_sample_if_online(ctx, handle, cycle);
    send_il_keepalive_if_online(ctx, handle);
    build_and_stage_outputs_to_c(ctx, handle, cycle);
    send_site_state(ctx, handle);
 
    /* Warm/hot standby: bring STANDBY up to date with THIS cycle's post-compute state before the staged
     * outputs above are committed at the checkpoint stage. Never withholds output - a STANDBY that does
     * not acknowledge is marked not working instead. No-op in cold mode. */
    ab_gp_channel_negotiate_sync_standby(ctx, cycle);
 
    /* 5. Heartbeat kick for the cyclic executive watchdog */
    if (ctx->watchdog != NULL)
    {
        (void)rte_watchdog_kick(ctx->watchdog);
    }
 
    /* 6. ADR-038: periodic status self-line + PUBLISH on the UTIL relay
     * (non-safety monitoring - after everything else this cycle). */
    ab_gp_status_tick(handle, ctx, cycle);
 
    return RTE_STATUS_OK;
}

rte_status_t ab_gp_channel_on_checkpoint_result(void *context, bool committed)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)context;
    ab_gp_channel_handle_t handle = (ab_gp_channel_handle_t)ctx;
 
    if (ctx == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    if (committed)
    {
        return ab_gp_channel_commit_pending_sends(handle);
    }
 
    ab_gp_channel_discard_pending_sends(handle);
    return RTE_STATUS_OK;
}