stellar-stellar/src/packet_io/marsio_io.cpp

#include <sched.h>
#include <stdlib.h>
#include <assert.h>
#include <string.h>
#include <netinet/ether.h>

#include "log.h"
#include "marsio.h"
#include "marsio_io.h"
#include "packet_private.h"
#include "packet_parser.h"

#define PACKET_IO_LOG_ERROR(format, ...) LOG_ERROR("marsio", format, ##__VA_ARGS__)

struct marsio_io
{
    struct mr_instance *mr_ins;
    struct mr_vdev *mr_dev;
    struct mr_sendpath *mr_path;

    struct packet_io_stat stat[MAX_THREAD_NUM];
};

/******************************************************************************
 * Private API
 ******************************************************************************/

static void metadata_from_mbuff_to_packet(marsio_buff_t *mbuff, struct packet *pkt)
{
    struct route_ctx route_ctx = {};
    struct sids sids = {};
    uint64_t session_id = {0};
    uint16_t link_id = {0};
    int is_ctrl = {0};
    enum packet_direction direction = PACKET_DIRECTION_OUTGOING;

    route_ctx.used = marsio_buff_get_metadata(mbuff, MR_BUFF_ROUTE_CTX, &route_ctx.data, sizeof(route_ctx.data));
    if (route_ctx.used > 0)
    {
        packet_set_route_ctx(pkt, &route_ctx);
    }
    else
    {
        PACKET_IO_LOG_ERROR("failed to get route ctx");
    }

    sids.used = marsio_buff_get_sid_list(mbuff, sids.sid, sizeof(sids.sid) / sizeof(sids.sid[0]));
    if (sids.used > 0)
    {
        packet_set_sids(pkt, &sids);
    }
    else
    {
        PACKET_IO_LOG_ERROR("failed to get sids");
    }

    if (marsio_buff_get_metadata(mbuff, MR_BUFF_SESSION_ID, &session_id, sizeof(session_id)) == sizeof(session_id))
    {
        packet_set_session_id(pkt, session_id);
    }
    else
    {
        PACKET_IO_LOG_ERROR("failed to get session id");
    }

// TODO
#if 0
    if (marsio_buff_get_metadata(mbuff, MR_BUFF_DOMAIN, &domain, sizeof(domain)) == sizeof(domain))
    {
        packet_set_domain(pkt, domain);
    }
    else
    {
        PACKET_IO_LOG_ERROR("failed to get domain id");
    }
#endif

    if (marsio_buff_get_metadata(mbuff, MR_BUFF_LINK_ID, &link_id, sizeof(link_id)) == sizeof(link_id))
    {
        packet_set_link_id(pkt, link_id);
    }
    else
    {
        PACKET_IO_LOG_ERROR("failed to get link id");
    }

    is_ctrl = marsio_buff_is_ctrlbuf(mbuff);
    packet_set_ctrl(pkt, is_ctrl);

    if (marsio_buff_get_metadata(mbuff, MR_BUFF_DIR, &direction, sizeof(direction)) == sizeof(direction))
    {
        packet_set_direction(pkt, direction);
    }
    else
    {
        PACKET_IO_LOG_ERROR("failed to get direction");
    }

    packet_set_action(pkt, PACKET_ACTION_FORWARD);
    packet_set_origin_ctx(pkt, mbuff);
}

static void metadata_from_packet_to_mbuff(struct packet *pkt, marsio_buff_t *mbuff)
{
    const struct route_ctx *route_ctx = packet_get_route_ctx(pkt);
    const struct sids *sids = packet_get_sids(pkt);
    uint64_t session_id = packet_get_session_id(pkt);
    // uint64_t domain = packet_get_domain(pkt);
    uint16_t link_id = packet_get_link_id(pkt);
    int is_ctrl = packet_is_ctrl(pkt);
    enum packet_direction direction = packet_get_direction(pkt);

    if (marsio_buff_set_metadata(mbuff, MR_BUFF_ROUTE_CTX, (void *)route_ctx->data, route_ctx->used) != 0)
    {
        PACKET_IO_LOG_ERROR("failed to set route ctx");
    }

    if (marsio_buff_set_sid_list(mbuff, (sid_t *)sids->sid, sids->used) != 0)
    {
        PACKET_IO_LOG_ERROR("failed to set sids");
    }

    if (marsio_buff_set_metadata(mbuff, MR_BUFF_SESSION_ID, &session_id, sizeof(session_id)) != 0)
    {
        PACKET_IO_LOG_ERROR("failed to set session id");
    }

// TODO
#if 0
    if (marsio_buff_set_metadata(mbuff, MR_BUFF_DOMAIN, &domain, sizeof(domain)) != 0)
    {
        PACKET_IO_LOG_ERROR("failed to set domain id");
    }
#endif

    if (marsio_buff_set_metadata(mbuff, MR_BUFF_LINK_ID, &link_id, sizeof(link_id)) != 0)
    {
        PACKET_IO_LOG_ERROR("failed to set link id");
    }

    if (is_ctrl)
    {
        marsio_buff_set_ctrlbuf(mbuff);
    }

    if (marsio_buff_set_metadata(mbuff, MR_BUFF_DIR, &direction, sizeof(direction)) != 0)
    {
        PACKET_IO_LOG_ERROR("failed to set direction");
    }
}

static inline int is_keepalive_packet(const char *data, int len)
{
    if (data == NULL || len < (int)(sizeof(struct ethhdr)))
    {
        return 0;
    }

    struct ethhdr *eth_hdr = (struct ethhdr *)data;
    if (eth_hdr->h_proto == 0xAAAA)
    {
        return 1;
    }
    else
    {
        return 0;
    }
}

/******************************************************************************
 * Public API
 ******************************************************************************/

struct marsio_io *marsio_io_new(const char *app_symbol, const char *dev_symbol, uint16_t *cpu_mask, uint16_t nr_threads)
{
    int opt = 1;
    cpu_set_t coremask;
    CPU_ZERO(&coremask);
    for (uint16_t i = 0; i < nr_threads; i++)
    {
        CPU_SET(cpu_mask[i], &coremask);
    }

    struct marsio_io *handle = (struct marsio_io *)calloc(1, sizeof(struct marsio_io));
    if (handle == NULL)
    {
        PACKET_IO_LOG_ERROR("unable to allocate memory for marsio_io");
        return NULL;
    }

    handle->mr_ins = marsio_create();
    if (handle->mr_ins == NULL)
    {
        PACKET_IO_LOG_ERROR("unable to create marsio instance");
        goto error_out;
    }
    marsio_option_set(handle->mr_ins, MARSIO_OPT_THREAD_MASK_IN_CPUSET, &coremask, sizeof(cpu_set_t));
    marsio_option_set(handle->mr_ins, MARSIO_OPT_EXIT_WHEN_ERR, &opt, sizeof(opt));
    if (marsio_init(handle->mr_ins, app_symbol) != 0)
    {
        PACKET_IO_LOG_ERROR("unable to init marsio instance");
        goto error_out;
    }
    handle->mr_dev = marsio_open_device(handle->mr_ins, dev_symbol, nr_threads, nr_threads);
    if (handle->mr_dev == NULL)
    {
        PACKET_IO_LOG_ERROR("unable to open marsio device");
        goto error_out;
    }
    handle->mr_path = marsio_sendpath_create_by_vdev(handle->mr_dev);
    if (handle->mr_path == NULL)
    {
        PACKET_IO_LOG_ERROR("unable to create marsio sendpath");
        goto error_out;
    }

    return handle;

error_out:
    marsio_io_free(handle);
    return NULL;
}

void marsio_io_free(struct marsio_io *handle)
{
    if (handle)
    {
        if (handle->mr_path)
        {
            marsio_sendpath_destory(handle->mr_path);
            handle->mr_path = NULL;
        }
        if (handle->mr_dev)
        {
            marsio_close_device(handle->mr_dev);
            handle->mr_dev = NULL;
        }
        if (handle->mr_ins)
        {
            marsio_destory(handle->mr_ins);
            handle->mr_ins = NULL;
        }
        free(handle);
        handle = NULL;
    }
}

int marsio_io_init(struct marsio_io *handle, uint16_t thr_idx __attribute__((unused)))
{
    if (marsio_thread_init(handle->mr_ins) != 0)
    {
        PACKET_IO_LOG_ERROR("unable to init marsio thread");
        return -1;
    }

    return 0;
}

int marsio_io_ingress(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts)
{
    struct packet *pkt;
    marsio_buff_t *mbuff;
    marsio_buff_t *rx_buffs[RX_BURST_MAX];
    struct packet_io_stat *stat = &handle->stat[thr_idx];
    int nr_recv;
    int nr_parsed = 0;
    int len;
    char *data;

    nr_recv = marsio_recv_burst(handle->mr_dev, thr_idx, rx_buffs, MIN(RX_BURST_MAX, nr_pkts));
    if (nr_recv <= 0)
    {
        return 0;
    }

    for (int i = 0; i < nr_recv; i++)
    {
        mbuff = rx_buffs[i];
        data = marsio_buff_mtod(mbuff);
        len = marsio_buff_datalen(mbuff);

        stat->dev_rx_pkts++;
        stat->dev_rx_bytes += len;

        if (is_keepalive_packet(data, len))
        {
            stat->keep_alive_pkts++;
            stat->keep_alive_bytes += len;

            stat->dev_tx_pkts++;
            stat->dev_tx_bytes += len;

            marsio_send_burst(handle->mr_path, thr_idx, &mbuff, 1);
            continue;
        }

        pkt = &pkts[nr_parsed];
        packet_parse(pkt, data, len);
        metadata_from_mbuff_to_packet(mbuff, pkt);
        nr_parsed++;

        if (marsio_buff_is_ctrlbuf(mbuff))
        {
            stat->ctrl_rx_pkts++;
            stat->ctrl_rx_bytes += len;
        }
        else
        {
            stat->raw_rx_pkts++;
            stat->raw_rx_bytes += len;
        }
    }

    return nr_parsed;
}

void marsio_io_egress(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts)
{
    struct packet *pkt;
    marsio_buff_t *mbuff;
    struct packet_io_stat *stat = &handle->stat[thr_idx];
    int len;

    for (int i = 0; i < nr_pkts; i++)
    {
        pkt = &pkts[i];
        len = packet_get_raw_len(pkt);

        stat->dev_tx_pkts++;
        stat->dev_tx_bytes += len;

        mbuff = (marsio_buff_t *)packet_get_origin_ctx(pkt);
        assert(mbuff != NULL);
        metadata_from_packet_to_mbuff(pkt, mbuff);

        if (marsio_buff_is_ctrlbuf(mbuff))
        {
            stat->ctrl_tx_pkts++;
            stat->ctrl_tx_bytes += len;
        }
        else
        {
            stat->raw_tx_pkts++;
            stat->raw_tx_bytes += len;
        }

        marsio_send_burst(handle->mr_path, thr_idx, &mbuff, 1);
        packet_free(pkt);
    }
}

void marsio_io_drop(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts)
{
    struct packet *pkt;
    marsio_buff_t *mbuff;
    struct packet_io_stat *stat = &handle->stat[thr_idx];

    for (int i = 0; i < nr_pkts; i++)
    {
        pkt = &pkts[i];
        mbuff = (marsio_buff_t *)packet_get_origin_ctx(pkt);
        if (mbuff)
        {
            stat->drop_pkts++;
            stat->drop_bytes += packet_get_raw_len(pkt);
            marsio_buff_free(handle->mr_ins, &mbuff, 1, 0, thr_idx);
        }
        packet_free(pkt);
    }
}

int marsio_io_inject(struct marsio_io *handle, uint16_t thr_idx, struct packet *pkts, int nr_pkts)
{
    int len;
    int nr_inject = 0;
    char *ptr;
    struct packet *pkt;
    marsio_buff_t *mbuff;
    struct packet_io_stat *stat = &handle->stat[thr_idx];

    for (int i = 0; i < nr_pkts; i++)
    {
        pkt = &pkts[i];
        len = packet_get_raw_len(pkt);

        if (marsio_buff_malloc_global(handle->mr_ins, &mbuff, 1, MARSIO_SOCKET_ID_ANY, MARSIO_LCORE_ID_ANY) < 0)
        {
            PACKET_IO_LOG_ERROR("unable to allocate marsio buffer for inject packet");
            continue;
        }

        stat->inject_pkts++;
        stat->inject_bytes += len;

        stat->raw_tx_pkts++;
        stat->raw_tx_bytes += len;

        stat->dev_tx_pkts++;
        stat->dev_tx_bytes += len;

        nr_inject++;

        ptr = marsio_buff_append(mbuff, len);
        memcpy(ptr, packet_get_raw_data(pkt), len);
        metadata_from_packet_to_mbuff(pkt, mbuff);

        marsio_send_burst_with_options(handle->mr_path, thr_idx, &mbuff, 1, MARSIO_SEND_OPT_REHASH);
        packet_free(pkt);
    }

    return nr_inject;
}

void marsio_io_yield(struct marsio_io *handle, uint16_t thr_idx, uint64_t timeout_ms)
{
    struct mr_vdev *vdevs[1] = {
        handle->mr_dev,
    };

    marsio_poll_wait(handle->mr_ins, vdevs, 1, thr_idx, timeout_ms);
}

struct packet_io_stat *marsio_io_stat(struct marsio_io *handle, uint16_t thr_idx)
{
    return &handle->stat[thr_idx];
}