path: root/service/src/pdump.c

#include "rte_bpf.h"
#include <common.h>
#include <fcntl.h>
#include <mrb_define.h>
#include <rte_malloc.h>
#include <rte_pcapng.h>
#include <rte_pdump.h>
#include <rte_ring.h>
#include <sc_pdump.h>
#include <sys/utsname.h>
#include <unistd.h>

/* Mrpdump copy function */
typedef void (*shmdev_pdump_func)(struct rte_mempool * mp, struct rte_ring * ring, struct rte_mbuf ** mbufs,
                                  unsigned int nr_pkts, struct rte_pdump_stats * stats, struct rte_bpf * bpf,
                                  uint64_t snaplen);

struct port_resource
{
    uint16_t queue;
    uint64_t snaplen;
    rte_atomic64_t rx_enable;
    rte_atomic64_t tx_enable;
    struct rte_ring * rx_ring;
    struct rte_ring * tx_ring;
    shmdev_pdump_func rx_pdump_func;
    shmdev_pdump_func tx_pdump_func;
    struct rte_bpf * rx_bpf;
    struct rte_bpf * tx_bpf;
    struct rte_pdump_stats rx_stats;
    struct rte_pdump_stats tx_stats;
};
/* Mrpdump */
struct mr_pdump
{
    struct port_resource ports[MR_DEVICE_MAX];
    struct rte_mempool * mr_pdump_shmdev_mempool;
    struct rte_mempool * mr_pdump_phydev_mempool;
};

/* Global mrpdump main */
static struct mr_pdump * g_mr_pdump_main = NULL;

/* Mrpdump for rx */
void mr_pdump_rx(port_id_t shmdev_port_id, queue_id_t qid, struct rte_mbuf ** mbufs, unsigned int nr_pkts)
{
    struct mr_pdump * _mr_pdump_main = g_mr_pdump_main;
    port_id_t port_id = shmdev_port_id + RTE_MAX_ETHPORTS;

    /* Check enable  */
    if (likely(rte_atomic64_read(&_mr_pdump_main->ports[port_id].rx_enable) == 0))
        return;
    if (likely((_mr_pdump_main->ports[port_id].rx_pdump_func == NULL)))
        return;
    if (unlikely((_mr_pdump_main->ports[port_id].queue != RTE_PDUMP_ALL_QUEUES)) &&
        (_mr_pdump_main->ports[port_id].queue != qid))
        return;

    /* Get pdump info */
    struct rte_mempool * mp = _mr_pdump_main->mr_pdump_shmdev_mempool;
    struct rte_ring * ring = _mr_pdump_main->ports[port_id].rx_ring;
    struct rte_pdump_stats * stats = &_mr_pdump_main->ports[port_id].rx_stats;
    shmdev_pdump_func pdump_func = _mr_pdump_main->ports[port_id].rx_pdump_func;
    struct rte_bpf * bpf = _mr_pdump_main->ports[port_id].rx_bpf;
    uint64_t snaplen = _mr_pdump_main->ports[port_id].snaplen;

    /* Dump the pkt */
    pdump_func(mp, ring, mbufs, nr_pkts, stats, bpf, snaplen);
}

/* Mrpdump for tx */
void mr_pdump_tx(port_id_t shmdev_port_id, queue_id_t qid, struct rte_mbuf ** mbufs, unsigned int nr_pkts)
{
    struct mr_pdump * _mr_pdump_main = g_mr_pdump_main;
    port_id_t port_id = shmdev_port_id + RTE_MAX_ETHPORTS;

    /* Check enable  */
    if (likely(rte_atomic64_read(&_mr_pdump_main->ports[port_id].tx_enable) == 0))
        return;
    if (likely((_mr_pdump_main->ports[port_id].tx_pdump_func == NULL)))
        return;
    if (unlikely((_mr_pdump_main->ports[port_id].queue != RTE_PDUMP_ALL_QUEUES)) &&
        (_mr_pdump_main->ports[port_id].queue != qid))
        return;

    /* Get pdump info */
    struct rte_mempool * mp = _mr_pdump_main->mr_pdump_shmdev_mempool;
    struct rte_ring * ring = _mr_pdump_main->ports[port_id].tx_ring;
    struct rte_pdump_stats * stats = &_mr_pdump_main->ports[port_id].tx_stats;
    shmdev_pdump_func pdump_func = _mr_pdump_main->ports[port_id].tx_pdump_func;
    struct rte_bpf * bpf = _mr_pdump_main->ports[port_id].tx_bpf;
    uint64_t snaplen = _mr_pdump_main->ports[port_id].snaplen;

    /* Dump the pkt */
    pdump_func(mp, ring, mbufs, nr_pkts, stats, bpf, snaplen);
}

void shmdev_pdump_copy(struct rte_mempool * mp, struct rte_ring * ring, struct rte_mbuf ** mbufs, unsigned int nr_pkts,
                       struct rte_pdump_stats * stats, struct rte_bpf * bpf, uint64_t snaplen)
{
    uint16_t d_pkts = 0;
    uint64_t rcs[nr_pkts];
    struct rte_mbuf * dup_bufs[nr_pkts];

    if (bpf != NULL)
    {
        rte_bpf_exec_burst(bpf, (void **)mbufs, rcs, nr_pkts);
    }

    for (unsigned int i = 0; i < nr_pkts; i++)
    {

        if ((bpf != NULL) && rcs[i] == 0)
        {
            __atomic_fetch_add(&stats->filtered, 1, __ATOMIC_RELAXED);
            continue;
        }

        struct rte_mbuf * new_mbuf = rte_pktmbuf_copy(mbufs[i], mp, 0, snaplen);
        if (unlikely(new_mbuf == NULL))
            __atomic_fetch_add(&stats->nombuf, 1, __ATOMIC_RELAXED);
        else
        {
            dup_bufs[d_pkts++] = new_mbuf;
            new_mbuf->tx_offload = rte_pktmbuf_pkt_len(mbufs[i]);
            struct mrb_metadata * mrb_metadata = mrbuf_cz_data(mbufs[i], MR_NODE_CTRLZONE_ID);
            struct mrb_metadata * mrb_metadata_copy = mrbuf_cz_data(new_mbuf, MR_NODE_CTRLZONE_ID);
            memcpy(mrb_metadata_copy, mrb_metadata, sizeof(struct mrb_metadata));
        }
    }

    __atomic_fetch_add(&stats->accepted, d_pkts, __ATOMIC_RELAXED);

    int ring_enq = rte_ring_enqueue_burst(ring, (void *)dup_bufs, d_pkts, NULL);
    if (unlikely(ring_enq < d_pkts))
    {
        unsigned int drops = d_pkts - ring_enq;

        __atomic_fetch_add(&stats->ringfull, drops, __ATOMIC_RELAXED);
        rte_pktmbuf_free_bulk(&dup_bufs[ring_enq], drops);
    }
}

/* Mrpdump enable */
int mr_pdump_enable(const struct mr_pdump_request * _cli_req, struct mr_pdump_response * _out_response,
                    struct mr_dev_desc * dev_desc)
{
    struct mr_pdump * _mr_pdump_main = g_mr_pdump_main;
    int32_t err_value = MR_PDUMP_SUCCESS;
    port_id_t port_id = dev_desc->port_id;
    struct rte_mempool * _mempool = NULL;

    /* Dump log */
    MR_INFO("Mrpdump  enable device: %s, prot: %u,  queue: %u, rx/tx: %u.", dev_desc->symbol, port_id, _cli_req->queue,
            _cli_req->flags);

    /* Deal rx */
    if (_cli_req->flags & RTE_PDUMP_FLAG_RX)
    {
        /* Fill the rx response */
        _out_response->rx_ring = _mr_pdump_main->ports[port_id].rx_ring;

        /* Clear rx stats */
        memset(&_mr_pdump_main->ports[port_id].rx_stats, 0, sizeof(struct rte_pdump_stats));

        /* Set shmdev pdump func */
        if (dev_desc->drv_type == MR_DEV_DRV_TYPE_SHMDEV)
        {
            _mempool = _mr_pdump_main->mr_pdump_shmdev_mempool;
            _mr_pdump_main->ports[port_id].rx_pdump_func = (shmdev_pdump_func)shmdev_pdump_copy;
            if (_cli_req->prm != NULL)
                _mr_pdump_main->ports[port_id].rx_bpf = rte_bpf_load(_cli_req->prm);
        }
        else
        {
            _mempool = _mr_pdump_main->mr_pdump_phydev_mempool;
        }

        /* Enable rx pdump */
        rte_atomic64_test_and_set(&_mr_pdump_main->ports[port_id].rx_enable);
    }

    /* Deal tx ring */
    if (_cli_req->flags & RTE_PDUMP_FLAG_TX)
    {
        /* Fill the tx response */
        _out_response->tx_ring = _mr_pdump_main->ports[port_id].tx_ring;

        /* Clear tx stats */
        memset(&_mr_pdump_main->ports[port_id].tx_stats, 0, sizeof(struct rte_pdump_stats));

        /* Set shmdev pdump func */
        if (dev_desc->drv_type == MR_DEV_DRV_TYPE_SHMDEV)
        {
            _mempool = _mr_pdump_main->mr_pdump_shmdev_mempool;
            _mr_pdump_main->ports[port_id].tx_pdump_func = (shmdev_pdump_func)shmdev_pdump_copy;
            if (_cli_req->prm != NULL)
                _mr_pdump_main->ports[port_id].tx_bpf = rte_bpf_load(_cli_req->prm);
        }
        else
        {
            _mempool = _mr_pdump_main->mr_pdump_phydev_mempool;
        }

        /* Enable tx pdump */
        rte_atomic64_test_and_set(&_mr_pdump_main->ports[port_id].tx_enable);
    }

    /* Fill the response for public */
    _out_response->res_op = _cli_req->op;
    _out_response->err_value = err_value;
    _out_response->port_id = port_id;
    _out_response->drv_type = dev_desc->drv_type;
    _out_response->mp = _mempool;
    rte_ether_addr_copy(&dev_desc->eth_addr, &_out_response->eth_addr);

    /* Save the pdump info for public */
    _mr_pdump_main->ports[port_id].queue = _cli_req->queue;
    _mr_pdump_main->ports[port_id].snaplen = _cli_req->snaplen;

    return RT_SUCCESS;
}

/* Mrpdump stats print */
void mr_pdump_stats_dump(struct mr_dev_desc * _dev_desc, struct rte_pdump_stats * _stats, uint32_t flags)
{
    if (_dev_desc->drv_type != MR_DEV_DRV_TYPE_SHMDEV)
        return;

    if (flags & RTE_PDUMP_FLAG_RX)
    {
        MR_INFO("Mrpdump  device: %s, prot: %u, rx accepted: %lu ,rx filtered: %lu ,rx nombuf: %lu ,rx ringfull: %lu ",
                _dev_desc->symbol, _dev_desc->port_id, _stats->accepted, _stats->filtered, _stats->nombuf,
                _stats->ringfull);
    }
    if (flags & RTE_PDUMP_FLAG_TX)
    {
        MR_INFO("Mrpdump  device: %s, prot: %u, tx accepted: %lu ,tx filtered: %lu ,tx nombuf :%lu ,tx ringfull: %lu ",
                _dev_desc->symbol, _dev_desc->port_id, _stats->accepted, _stats->filtered, _stats->nombuf,
                _stats->ringfull);
    }
}

/* Mrpdump disable */
int mr_pdump_disable(const struct mr_pdump_request * _cli_req, struct mr_pdump_response * _out_response,
                     struct mr_dev_desc * _dev_desc)
{
    struct mr_pdump * _mr_pdump_main = g_mr_pdump_main;
    int32_t err_value = MR_PDUMP_SUCCESS;
    port_id_t port_id = _dev_desc->port_id;

    /* Print log */
    MR_INFO("Mrpdump  disable device: %s, prot: %u,  queue: %u, rx/tx: %u.", _dev_desc->symbol, port_id,
            _cli_req->queue, _cli_req->flags);

    /* Deal rx */
    if (_cli_req->flags & RTE_PDUMP_FLAG_RX)
    {
        /* Set rx disable */
        rte_atomic64_clear(&_mr_pdump_main->ports[port_id].rx_enable);

        /* Fill the response for rx */
        _out_response->rx_ring = NULL;

        /* Clear pdump func for rx */
        _mr_pdump_main->ports[port_id].rx_pdump_func = NULL;

        if (_mr_pdump_main->ports[port_id].rx_bpf != NULL)
        {
            rte_bpf_destroy(_mr_pdump_main->ports[port_id].rx_bpf);
            _mr_pdump_main->ports[port_id].rx_bpf = NULL;
        }

        /* Dump and clear rx stats */
        mr_pdump_stats_dump(_dev_desc, &_mr_pdump_main->ports[port_id].rx_stats, RTE_PDUMP_FLAG_RX);
        memset(&_mr_pdump_main->ports[port_id].rx_stats, 0, sizeof(struct rte_pdump_stats));
    }

    /* Deal tx ring */
    if (_cli_req->flags & RTE_PDUMP_FLAG_TX)
    {
        /* Set tx disable */
        rte_atomic64_clear(&_mr_pdump_main->ports[port_id].tx_enable);

        /* Fill the response for tx */
        _out_response->tx_ring = NULL;

        /* Clear pdump func for tx */
        _mr_pdump_main->ports[port_id].tx_pdump_func = NULL;

        if (_mr_pdump_main->ports[port_id].tx_bpf != NULL)
        {
            rte_bpf_destroy(_mr_pdump_main->ports[port_id].tx_bpf);
            _mr_pdump_main->ports[port_id].tx_bpf = NULL;
        }

        /* Dump and clear tx stats */
        mr_pdump_stats_dump(_dev_desc, &_mr_pdump_main->ports[port_id].tx_stats, RTE_PDUMP_FLAG_TX);
        memset(&_mr_pdump_main->ports[port_id].tx_stats, 0, sizeof(struct rte_pdump_stats));
    }

    /* Fill the response for public */
    _out_response->res_op = _cli_req->op;
    _out_response->err_value = err_value;
    _out_response->port_id = 0;
    _out_response->drv_type = 0;
    _out_response->mp = NULL;

    /* Save pdump info for public */
    _mr_pdump_main->ports[port_id].queue = 0;

    return RT_SUCCESS;
}

/* Get os info */
static char * get_os_info(void)
{
    struct utsname uts;
    char * os_name = NULL;

    if (uname(&uts) < 0)
        return NULL;

    if (asprintf(&os_name, "%s %s", uts.sysname, uts.release) == -1)
        return NULL;

    return os_name;
}

/* Pcapng header write */
void mr_pdump_pcapng_header_write(int pcapng_fd, struct mr_pdump_response * _out_response)
{
    /* Write the pcapng header for the pcapng fd */
    _out_response->res_op = WRITE_PCAPNG_HEADER;
    rte_pcapng_t * _pcapng_item = rte_pcapng_fdopen(pcapng_fd, get_os_info(), NULL, "mrpdump v1.0", NULL);
    if (_pcapng_item == NULL)
    {
        _out_response->err_value = MR_PDUMP_ERR_FOR_PCAPNG_CREATE;
        MR_ERROR("mrpdump write the pcapng header err.");
    }
    else
    {
        _out_response->err_value = MR_PDUMP_SUCCESS;
        MR_INFO("mrpdump write the pcapng header success.");
    }
}

/* Pcapng server */
static int mr_pdump_server(const struct rte_mp_msg * mp_msg, const void * peer)
{
    struct sc_main * sc = sc_main_get();
    struct mr_dev_desc * dev_desc = NULL;
    struct mr_pdump_response response = {};
    struct rte_mp_msg mp_resp = {};
    struct mr_pdump_response * _resp = (struct mr_pdump_response *)&mp_resp.param;
    rte_strscpy(mp_resp.name, MR_PDUMP_MP, RTE_MP_MAX_NAME_LEN);
    mp_resp.len_param = sizeof(*_resp);

    /* recv client requests */
    const struct mr_pdump_request * _cli_req = (const struct mr_pdump_request *)mp_msg->param;
    if (mp_msg->len_param != sizeof(*_cli_req))
    {
        MR_ERROR("mrpdump failed to recv from client");
        response.err_value = MR_PDUMP_ERR_FOR_MESSAGE_INVALID;
        goto done;
    }

    if (_cli_req->op != WRITE_PCAPNG_HEADER)
    {
        /* Check devices name */
        dev_desc = mr_dev_desc_lookup(sc->devmgr_main, _cli_req->dev_symbol);
        if (dev_desc == NULL)
        {
            MR_ERROR("mrpdump invalid devices(%s).", _cli_req->dev_symbol);
            response.err_value = MR_PDUMP_ERR_FOR_PORT_INVALID;
            goto done;
        }
    }

    /* Check res op */
    if (_cli_req->op == ENABLE)
    {
        mr_pdump_enable(_cli_req, &response, dev_desc);
    }
    else if (_cli_req->op == DISABLE)
    {
        mr_pdump_disable(_cli_req, &response, dev_desc);
    }
    else if (_cli_req->op == WRITE_PCAPNG_HEADER)
    {
        mr_pdump_pcapng_header_write(mp_msg->fds[0], &response);
    }

done:
    /* Send response */
    memcpy(_resp, &response, sizeof(struct mr_pdump_response));
    int ret = rte_mp_reply(&mp_resp, peer);

    /* Check response */
    if (ret < 0)
    {
        MR_ERROR("mrpdump failed to send to client:%s.", strerror(rte_errno));
        return -1;
    }
    return 0;
}

/* Mrpdump init */
int mr_pdump_init(struct sc_main * sc)
{
    struct mr_pdump * mr_pdump_main = ZMALLOC(sizeof(struct mr_pdump));
    MR_VERIFY_MALLOC(mr_pdump_main);
    g_mr_pdump_main = mr_pdump_main;

    /* Message register */
    int ret = rte_mp_action_register(MR_PDUMP_MP, mr_pdump_server);
    if (ret < 0)
    {
        MR_ERROR("mrpdump message register failed.");
        return RT_ERR;
    }

    /* Create mempool */
    char _pool_symbol[MR_SYMBOL_MAX];
    snprintf(_pool_symbol, sizeof(_pool_symbol) - 1, "MZ_%s", MR_PDUMP_SHMDEV_MEMPOOL_NAME);
    struct rte_mempool * mr_pdump_shmdev_mempool = rte_mempool_lookup(_pool_symbol);
    if (mr_pdump_shmdev_mempool == NULL)
    {
        MR_ERROR("mrpdump shmdev mempool no create .");
        return RT_ERR;
    }

    snprintf(_pool_symbol, sizeof(_pool_symbol) - 1, "MZ_%s", MR_PDUMP_PHYDEV_MEMPOOL_NAME);
    struct rte_mempool * mr_pdump_phydev_mempool = rte_mempool_lookup(_pool_symbol);
    if (mr_pdump_phydev_mempool == NULL)
    {
        MR_ERROR("mrpdump phydev mempool no create .");
        return RT_ERR;
    }

    /* Save mempool */
    mr_pdump_main->mr_pdump_shmdev_mempool = mr_pdump_shmdev_mempool;
    mr_pdump_main->mr_pdump_phydev_mempool = mr_pdump_phydev_mempool;

    /* RX/TX Ring Create */
    char ring_name[MR_PDUMP_RING_NAME_SIZE] = {};
    for (int index = 0; index < MR_DEVICE_MAX; index++)
    {
        /* Create a new rx ring */
        snprintf(ring_name, MR_PDUMP_RING_NAME_SIZE, MR_PDUMP_RX_RING, index);
        struct rte_ring * rx_ring = rte_ring_create(ring_name, MR_PDUMP_RING_SIZE, rte_socket_id(), 0);
        if (rx_ring == NULL)
        {
            MR_ERROR("mrpdump rx ring (%s) create err.", ring_name);
            return RT_ERR;
        }

        /* Create a new tx ring */
        snprintf(ring_name, MR_PDUMP_RING_NAME_SIZE, MR_PDUMP_TX_RING, index);
        struct rte_ring * tx_ring = rte_ring_create(ring_name, MR_PDUMP_RING_SIZE, rte_socket_id(), 0);
        if (tx_ring == NULL)
        {
            MR_ERROR("mrpdump tx ring (%s) create err.", ring_name);
            return RT_ERR;
        }

        mr_pdump_main->ports[index].rx_ring = rx_ring;
        mr_pdump_main->ports[index].tx_ring = tx_ring;
    }

    /* Init enable switch */
    for (int index = 0; index < MR_DEVICE_MAX; index++)
    {
        rte_atomic64_init(&mr_pdump_main->ports[index].rx_enable);
        rte_atomic64_init(&mr_pdump_main->ports[index].tx_enable);
    }

    return RT_SUCCESS;
}