ab_gp_channel_checkpoint.c

Go to the documentation of this file.

#include "ab_gp_channel_checkpoint.h"
 
#include "rte/oal/log/rte_log.h"
#include "rte/oal/timer/rte_timer.h"
 
#include "../../common/common_config.h"
#include "ab_gp_channel_types.h"
#include "ab_gp_channel_util.h"
#include "ab_gp_channel_negotiate.h"
#include "ab_gp_sibling_state.h"


static void ab_gp_channel_checkpoint_note_peer_frame(ab_gp_channel_context_t *ctx)
{
    ctx->last_peer_checkpoint_cycle = ctx->cyc_cycle;
}

static bool ab_gp_channel_checkpoint_peer_recent(const ab_gp_channel_context_t *ctx)
{
    bool recent = false;
 
    if ((ctx->last_peer_checkpoint_cycle != 0U) && (ctx->cyc_cycle >= ctx->last_peer_checkpoint_cycle))
    {
        uint32_t gap = ctx->cyc_cycle - ctx->last_peer_checkpoint_cycle;
 
        recent = (gap <= (uint32_t)RTE_EXAMPLE_AB_CHECKPOINT_LOOKBACK_CYCLES);
    }
    return recent;
}

 
rte_status_t ab_gp_channel_checkpoint_backend_send(void *channel, const void *data, size_t data_size)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)channel;
    uint8_t frame[PEER_WIRE_FRAME_SIZE] = { 0 };
    rte_status_t status;
 
    if (ctx == NULL || data == NULL || data_size != sizeof(rte_vital_message_t))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    /* Retain this cycle's own arrival verbatim: backend_recv() returns it
     * to satisfy the lookback path (it verifies against this cycle's
     * checkpoint_id by construction). */
    rte_mem_copy(ctx->last_sent_checkpoint_arrival, data, sizeof(rte_vital_message_t));
    ctx->has_last_sent_checkpoint_arrival = true;
 
    frame[0] = PEER_MSG_KIND_CHECKPOINT_REQUEST;
    rte_mem_copy(&frame[1], data, data_size);
    status = rte_channel_service_send(&ctx->com.peer_channel, frame, sizeof(frame), RTE_EXAMPLE_LINK_TIMEOUT_MS);
    rte_log_write_event(RTE_LOG_LEVEL_DEBUG, site_config_name(ctx->site), ctx->cyc_cycle, ctx->role_tag,
                          ctx->is_role_a ? "B" : "A", "CHECKPOINT_REQUEST",
                          (status == RTE_STATUS_OK) ? "sent" : rte_status_to_string(status), NULL);
    return status;
}

static rte_status_t poll_for_checkpoint_frame(ab_gp_channel_context_t *ctx, void *data, size_t data_size,
                                              uint32_t timeout_ms)
{
    rte_timestamp_ms_t start_ms = 0U;
    rte_timestamp_ms_t now_ms = 0U;
    uint8_t in_bytes[PEER_WIRE_FRAME_SIZE];
 
    (void)rte_timer_now(&start_ms);
    for (;;)
    {
        rte_duration_ms_t elapsed;
        rte_duration_ms_t remaining;
        rte_duration_ms_t poll_ms;
        rte_status_t status;
 
        (void)rte_timer_now(&now_ms);
        elapsed = (rte_duration_ms_t)(now_ms - start_ms);
        if (elapsed >= (rte_duration_ms_t)timeout_ms)
        {
            return RTE_STATUS_TIMEOUT;
        }
        remaining = (rte_duration_ms_t)timeout_ms - elapsed;
        poll_ms = (remaining > 20U) ? 20U : remaining;
 
        /* This wait can last most of a cycle (the two channels' cycles are not phase-locked). The inter-site
         * negotiation link acknowledges a frame only from inside a receive call, so without servicing it here the
         * other site's 50 ms ack window is missed for as long as this wait lasts, and for some start-up phase
         * relations that happened every cycle: the link was torn down every third cycle for good (Docker). */
        ab_gp_channel_negotiate_service_link(ctx);
        status = rte_channel_service_read(&ctx->com.peer_channel, in_bytes, sizeof(in_bytes), poll_ms);
        if (status == RTE_STATUS_OK)
        {
            if (in_bytes[0] == PEER_MSG_KIND_CHECKPOINT_REQUEST || in_bytes[0] == PEER_MSG_KIND_CHECKPOINT_REPLY)
            {
                ab_gp_channel_checkpoint_note_peer_frame(ctx);
                rte_mem_copy(data, &in_bytes[1], data_size);
                if (in_bytes[0] == PEER_MSG_KIND_CHECKPOINT_REQUEST)
                {
                    uint8_t reply_frame[PEER_WIRE_FRAME_SIZE] = { 0 };
                    reply_frame[0] = PEER_MSG_KIND_CHECKPOINT_REPLY;
                    rte_mem_copy(&reply_frame[1], &in_bytes[1], sizeof(rte_vital_message_t));
                    (void)rte_channel_service_send(&ctx->com.peer_channel, reply_frame, sizeof(reply_frame), RTE_EXAMPLE_LINK_TIMEOUT_MS);
                }
                return RTE_STATUS_OK;
            }
            if (in_bytes[0] == PEER_MSG_KIND_SITE_STATE)
            {
                (void)ab_gp_sibling_state_store(&ctx->peer_site_state, &in_bytes[1], sizeof(in_bytes) - 1U);
            }
            else if (in_bytes[0] == PEER_MSG_KIND_STARTUP_TS)
            {
                /* The sibling restarted and asks for the site's startup timestamp (ADR-040 step 5a): answered
                 * here, since the request would otherwise wait until the next pre_execute. */
                ab_gp_channel_negotiate_on_startup_ts_frame(ctx, &in_bytes[1], sizeof(in_bytes) - 1U);
            }
            else if (((in_bytes[0] == PEER_MSG_KIND_SAF_RB) || (in_bytes[0] == PEER_MSG_KIND_KM114_DIGEST)) &&
                     (ctx->deferred_peer_count < (uint8_t)AB_GP_DEFERRED_PEER_FRAMES))
            {
                /* One-off control frames: dropping one here left the other channel without a train's shared RB
                 * (or the KM114 cross-check without its digest) for good. Replayed by pre_execute. */
                rte_mem_copy(ctx->deferred_peer_frames[ctx->deferred_peer_count], in_bytes, sizeof(in_bytes));
                ctx->deferred_peer_count++;
            }
            else
            {
                /* other kinds are per-cycle data a later cycle re-sends */
            }
        }
        else if (status != RTE_STATUS_TIMEOUT)
        {
            return status;
        }
    }
}
 
rte_status_t ab_gp_channel_checkpoint_backend_recv(void *channel, void *data, size_t data_size, uint32_t timeout_ms)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)channel;
 
    if (ctx == NULL || data == NULL || data_size != sizeof(rte_vital_message_t))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    /* 1. Real matching peer frames the per-cycle pump already drained (see
     *    ab_gp_channel_types.h). The REPLY carries this side's id and
     *    confirms any idle cycle; the peer's REQUEST carries the peer's id
     *    and confirms an active cycle when both channels took the same
     *    path. A stale id is rejected by rte_checkpoint.c's
     *    reply_confirms_checkpoint(), which then retries within budget. */
    if (ctx->has_cached_checkpoint_reply)
    {
        ctx->has_cached_checkpoint_reply = false;
        rte_mem_copy(data, ctx->cached_checkpoint_reply, data_size);
        return RTE_STATUS_OK;
    }
    if (ctx->has_cached_checkpoint_request)
    {
        ctx->has_cached_checkpoint_request = false;
        rte_mem_copy(data, ctx->cached_checkpoint_request, data_size);
        return RTE_STATUS_OK;
    }
 
    /* 2. Lookback: a fresh round trip did not land in-window, but the peer
     *    is demonstrably still running its checkpoint stage (it exchanged a
     *    checkpoint frame with us within the last N cycles). Confirm this
     *    cycle using our own arrival - immune to the A/B cycle-phase drift
     *    that a loaded host causes. Bounded: once the peer goes N+1 cycles
     *    without any checkpoint frame this stops firing and the live poll
     *    below runs to its timeout, so a genuinely dead/partitioned peer
     *    still trips SAFE-state (REQ-CHECKPOINT-003), just one to two
     *    cycles later. */
    if (ctx->has_last_sent_checkpoint_arrival && ab_gp_channel_checkpoint_peer_recent(ctx))
    {
        rte_mem_copy(data, ctx->last_sent_checkpoint_arrival, data_size);
        return RTE_STATUS_OK;
    }
 
    /* 3. Last-ditch live poll for whatever remains of this round's budget - see
     *    poll_for_checkpoint_frame()'s own doc. */
    return poll_for_checkpoint_frame(ctx, data, data_size, timeout_ms);
}