ab_gp_channel_stage.c

Go to the documentation of this file.

#include "ab_gp_channel_stage.h"
 
#include <string.h>
 
#include "rte/oal/memory/rte_mem_util.h"
#include "rte/utils/safestate/rte_safestate.h"





static rte_status_t stage_entry(ab_gp_channel_context_t *ctx, ab_gp_stage_dest_kind_t dest_kind,
                                  rte_example_relay_kind_t relay_kind, uint8_t relay_index,
                                  const uint8_t *bytes, size_t size);

void ab_gp_channel_stage_reset(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    if (ctx == NULL)
    {
        return;
    }
    ctx->staged_send_count = 0U;
}
 
rte_status_t ab_gp_channel_stage_peer(ab_gp_channel_handle_t handle, const uint8_t *bytes, size_t size)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    if (ctx == NULL || bytes == NULL || size == 0U)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    return stage_entry(ctx, AB_GP_STAGE_DEST_PEER, RTE_EXAMPLE_RELAY_KIND_TRAIN, 0U, bytes, size);
}
 
rte_status_t ab_gp_channel_stage_crosscompare(ab_gp_channel_handle_t handle, uint8_t kind,
                                               const uint8_t *payload, size_t size)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    uint8_t frame[PEER_WIRE_FRAME_SIZE];
 
    if ((ctx == NULL) || (payload == NULL) || (size == 0U) || (size > (sizeof(frame) - 1U)))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    rte_mem_set(frame, 0, sizeof(frame));
    frame[0] = kind;
    rte_mem_copy(&frame[1], payload, size);
    return stage_entry(ctx, AB_GP_STAGE_DEST_CROSSCOMPARE, RTE_EXAMPLE_RELAY_KIND_TRAIN, 0U, frame, sizeof(frame));
}
 
rte_status_t ab_gp_channel_stage_relay(ab_gp_channel_handle_t handle, rte_example_relay_kind_t kind,
                                         uint8_t index, const uint8_t *bytes, size_t size)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    if (ctx == NULL || bytes == NULL || size == 0U || kind >= RTE_EXAMPLE_RELAY_KIND_COUNT)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    return stage_entry(ctx, AB_GP_STAGE_DEST_RELAY, kind, index, bytes, size);
}
 
rte_status_t ab_gp_channel_commit_pending_sends(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
    rte_status_t first_error = RTE_STATUS_OK;
    uint32_t i;
 
    if (ctx == NULL)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    /* Internal invariant, not a runtime condition a caller can trigger:
     * ctx->staged_send_count is only ever incremented by stage_entry()'s
     * own already-checked-bounds logic (it refuses to grow the count past
     * AB_GP_MAX_STAGED_SENDS_PER_CYCLE, returning RTE_STATUS_RESOURCE_EXHAUSTED
     * instead). Violating this here would mean memory corruption or a
     * logic bug elsewhere wrote past this file's own control-flow
     * guarantees - safeAPIFreamwork's RTE_ASSERT() (unlike a plain
     * return-code check) exists exactly for this class of "should be
     * structurally impossible" invariant. */
    RTE_ASSERT(ctx->staged_send_count <= (uint32_t)AB_GP_MAX_STAGED_SENDS_PER_CYCLE);
 
    for (i = 0U; i < ctx->staged_send_count; i++)
    {
        const ab_gp_staged_send_t *entry = &ctx->staged_sends[i];
        rte_status_t status;
 
        if (entry->dest_kind == AB_GP_STAGE_DEST_PEER)
        {
            status = ab_gp_channel_send_peer(handle, entry->bytes, entry->size);
        }
        else if (entry->dest_kind == AB_GP_STAGE_DEST_CROSSCOMPARE)
        {
            status = ab_gp_channel_send_crosscompare(handle, entry->bytes[0], &entry->bytes[1], entry->size - 1U);
        }
        else
        {
            status = ab_gp_channel_send_relay(handle, entry->relay_kind, entry->relay_index, entry->bytes,
                                               entry->size);
        }
        if ((status != RTE_STATUS_OK) && (first_error == RTE_STATUS_OK))
        {
            first_error = status;
        }
    }
    ctx->staged_send_count = 0U;
    return first_error;
}
 
void ab_gp_channel_discard_pending_sends(ab_gp_channel_handle_t handle)
{
    ab_gp_channel_context_t *ctx = (ab_gp_channel_context_t *)handle;
 
    if (ctx == NULL)
    {
        return;
    }
    ctx->staged_send_count = 0U;
}
 
static rte_status_t stage_entry(ab_gp_channel_context_t *ctx, ab_gp_stage_dest_kind_t dest_kind,
                                  rte_example_relay_kind_t relay_kind, uint8_t relay_index,
                                  const uint8_t *bytes, size_t size)
{
    ab_gp_staged_send_t *entry;
 
    if (ctx->staged_send_count >= (uint32_t)AB_GP_MAX_STAGED_SENDS_PER_CYCLE)
    {
        return RTE_STATUS_RESOURCE_EXHAUSTED;
    }
    if (size > sizeof(entry->bytes))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    entry = &ctx->staged_sends[ctx->staged_send_count];
    entry->dest_kind = dest_kind;
    entry->relay_kind = relay_kind;
    entry->relay_index = relay_index;
    rte_mem_copy(entry->bytes, bytes, size);
    entry->size = size;
    ctx->staged_send_count++;
    return RTE_STATUS_OK;
}