ab_gp_channel_payload.c

Go to the documentation of this file.

#include "ab_gp_channel_payload.h"
 
#include "rte/utils/buffer/rte_buffer.h"
#include "rte/redundancy/checksum/rte_checksum.h"
#include "rte/utils/lifecycle/rte_lifecycle.h"
 
#include "../../common/common_config.h"


typedef struct
{
    uint32_t                     wire_size; /* this collector's own total contribution, bytes */
    uint32_t                     offset;    /* byte offset of this collector's own slice within
                                              * the combined payload */
    ab_gp_channel_payload_encode_fn encode_fn;
    ab_gp_channel_payload_decode_fn decode_fn;
    void                         *context;
} ab_gp_channel_payload_collector_t;

/* Module-private registry - fan-out list, not a single slot: more than
 * one module can each contribute its own slice (see this file's own
 * header doc). Relies on C's static zero-initialization, same convention
 * this codebase already uses for every other module-private registry. */
static ab_gp_channel_payload_collector_t g_collectors[AB_GP_CHANNEL_PAYLOAD_MAX_COLLECTORS];
static uint32_t                       g_collector_count = 0U;
static uint32_t                       g_payload_size = 0U; /* running total of every registered
                                                              * collector's own wire_size */


 

rte_status_t ab_gp_channel_payload_register_collector(uint32_t wire_size, ab_gp_channel_payload_encode_fn encode_fn,
                                                      ab_gp_channel_payload_decode_fn decode_fn, void *context,
                                                      ab_gp_channel_payload_collector_id_t *out_id)
{
    rte_status_t lifecycle_status;
    ab_gp_channel_payload_collector_t *slot;
 
    if ((wire_size == 0U) || (encode_fn == NULL) || (decode_fn == NULL) || (out_id == NULL))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    lifecycle_status = rte_lifecycle_check_setup_allowed();
    if (lifecycle_status != RTE_STATUS_OK)
    {
        return lifecycle_status;
    }
    if (g_collector_count >= (uint32_t)AB_GP_CHANNEL_PAYLOAD_MAX_COLLECTORS)
    {
        return RTE_STATUS_RESOURCE_EXHAUSTED;
    }
 
    slot = &g_collectors[g_collector_count];
    slot->wire_size = wire_size;
    slot->offset = g_payload_size;
    slot->encode_fn = encode_fn;
    slot->decode_fn = decode_fn;
    slot->context = context;
 
    *out_id = g_collector_count;
    g_payload_size += wire_size;
    g_collector_count++;
    return RTE_STATUS_OK;
}
 
void ab_gp_channel_payload_encode_collector(ab_gp_channel_payload_collector_id_t id, uint8_t *out, const void *items)
{
    if (id >= g_collector_count)
    {
        RTE_EXAMPLE_LOG("ab_gp_channel_payload: encode_collector() called with unregistered id %u\n",
                             (unsigned int)id);
        return;
    }
    g_collectors[id].encode_fn(g_collectors[id].context, items, out);
}
 
void ab_gp_channel_payload_decode_collector(ab_gp_channel_payload_collector_id_t id, const uint8_t *in, void *out_items)
{
    if (id >= g_collector_count)
    {
        RTE_EXAMPLE_LOG("ab_gp_channel_payload: decode_collector() called with unregistered id %u\n",
                             (unsigned int)id);
        return;
    }
    g_collectors[id].decode_fn(g_collectors[id].context, in, out_items);
}
 
void ab_gp_channel_payload_encode_ab_sample(uint8_t out[AB_SAMPLE_WIRE_SIZE], uint32_t cycle,
                                          const void *const sources[AB_GP_CHANNEL_PAYLOAD_MAX_COLLECTORS])
{
    rte_buffer_t buf;
    rte_crc64_t crc;
    uint32_t i;
 
    (void)rte_buffer_init(&buf, out, AB_SAMPLE_WIRE_SIZE);
    (void)rte_buffer_write_u32_le(&buf, cycle);
    for (i = 0U; i < g_collector_count; i++)
    {
        ab_gp_channel_payload_encode_collector(i, &out[4U + g_collectors[i].offset], sources[i]);
    }
    crc = rte_checksum_crc64(out, 4U + g_payload_size);
    (void)rte_buffer_set_length(&buf, 4U + g_payload_size);
    (void)rte_buffer_write_u64_le(&buf, crc);
}
 
rte_status_t ab_gp_channel_payload_decode_ab_sample(uint8_t in[AB_SAMPLE_WIRE_SIZE], uint32_t *out_cycle,
                                                   void *const dests[AB_GP_CHANNEL_PAYLOAD_MAX_COLLECTORS])
{
    rte_buffer_t buf;
    uint32_t cycle;
    rte_crc64_t received_crc;
    rte_checksum_result_t crc_result;
    rte_status_t status;
    const uint32_t crc_offset = 4U + g_payload_size;
    uint32_t i;
 
    if (rte_buffer_init(&buf, in, AB_SAMPLE_WIRE_SIZE) != RTE_STATUS_OK)
    {
        return RTE_STATUS_INVALID_PARAM;
    }
    (void)rte_buffer_set_length(&buf, AB_SAMPLE_WIRE_SIZE);
 
    if ((rte_buffer_read_u32_le(&buf, 0U, &cycle) != RTE_STATUS_OK)
        || (rte_buffer_read_u64_le(&buf, crc_offset, &received_crc) != RTE_STATUS_OK))
    {
        return RTE_STATUS_INVALID_PARAM;
    }
 
    status = rte_checksum_crc64_verify(in, crc_offset, received_crc, &crc_result);
    if (status != RTE_STATUS_OK)
    {
        return status;
    }
 
    *out_cycle = cycle;
    for (i = 0U; i < g_collector_count; i++)
    {
        ab_gp_channel_payload_decode_collector(i, &in[4U + g_collectors[i].offset], dests[i]);
    }
    return RTE_STATUS_OK;
}