/*
    \brief 数据分发虚拟设备实现
    数据分发虚设备用于MARSIO服务进程向各从应用分发数据，并将各从应用发送的数据转发到物理网卡。

    \author Qiuwen Lu<luqiuwen@iie.ac.cn>
    \date 2017-05-09
*/

#include <assert.h>
#include <rte_malloc.h>

#include <common.h>
#include <sc_common.h>
#include <sc_vdev.h>

int vdev_data_dispatch(struct _vdev * _vdev, queue_id_t qid, struct rte_mbuf * pkts[], unsigned int nr_pkts,
                              int flags)
{
    hash_t hash_value[MR_BURST_MAX];
    assert(nr_pkts <= RTE_DIM(hash_value));

    for (unsigned int i = 0; i < nr_pkts; i++)
    {
        hash_value[i] = pkts[i]->hash.usr;
    }

    for (unsigned int i = 0; i < nr_pkts; i++)
    {
        __rte_mbuf_sanity_check(pkts[i], 1);
    }

    return vnode_mirror_enqueue_bulk(_vdev->vnode_rx_prod, qid, pkts, hash_value, nr_pkts);
}

int vdev_data_collect(struct _vdev * _vdev, queue_id_t qid, struct rte_mbuf * pkts[], unsigned int nr_pkts,
                             int flags)
{
    int __nr_mbufs_out = 0;
    int __nr_mbufs_left = nr_pkts;
    int ret;

    // 从FTX通道取包
    ret = vnode_mirror_dequeue_burst(_vdev->vnode_ftx_cons, qid, &pkts[__nr_mbufs_out], __nr_mbufs_left);

    __nr_mbufs_out += ret;
    __nr_mbufs_left -= ret;
    assert(__nr_mbufs_left >= 0);

    // 剩余的容量，从TX通道取
    ret = vnode_mirror_dequeue_burst(_vdev->vnode_tx_cons, qid, &pkts[__nr_mbufs_out], __nr_mbufs_left);

    __nr_mbufs_out += ret;
    __nr_mbufs_left -= ret;
    assert(__nr_mbufs_left >= 0);

    for (int i = 0; i < __nr_mbufs_out; i++)
    {
        __rte_mbuf_sanity_check(pkts[i], 1);
    }

    return __nr_mbufs_out;
}

int vdev_data_rt_pkts_retrieve(struct _vdev * _vdev, queue_id_t qid, struct rte_mbuf * pkts[], unsigned int nr_pkts)
{
    return vnode_mirror_rt_object_retrieve(_vdev->vnode_rx_prod, qid, pkts, nr_pkts);
}

int vdev_data_idle_poll(struct _vdev * _vdev, queue_id_t qid)
{
    vnode_mirror_flush(_vdev->vnode_rx_prod, qid);
    return 0;
}

int vdev_data_stats_get(struct _vdev * _vdev, struct vdev_stat_info * stat_info)
{
    struct vnode_prod_stat * st_prod_rx = vnode_mirror_prod_stat_get(_vdev->vnode_rx_prod);
    struct vnode_cons_stat * st_cons_tx = vnode_mirror_cons_stat_get(_vdev->vnode_tx_cons);
    struct vnode_cons_stat * st_cons_ftx = vnode_mirror_cons_stat_get(_vdev->vnode_ftx_cons);

    // TODO: What about nr_rxstream, nr_txstream is different?
    for (int i = 0; i < _vdev->nr_rxstream; i++)
    {
        stat_info->rx_on_line[i] = VNODE_STAT_READ(&st_prod_rx[i].on_line);
        stat_info->rx_deliver[i] = VNODE_STAT_READ(&st_prod_rx[i].deliver);
        stat_info->rx_missed[i] = VNODE_STAT_READ(&st_prod_rx[i].missed);
        stat_info->rx_total_len[i] = VNODE_STAT_READ(&st_prod_rx[i].total_len);
        stat_info->batch_size_count[i] = VNODE_STAT_READ(&st_prod_rx[i].batch_size_count);
        stat_info->batch_size_total[i] = VNODE_STAT_READ(&st_prod_rx[i].batch_size_total);
    }

    for (int i = 0; i < _vdev->nr_txstream; i++)
    {
        stat_info->tx_on_line[i] = VNODE_STAT_READ(&st_cons_tx[i].on_line);
        stat_info->tx_deliver[i] = VNODE_STAT_READ(&st_cons_tx[i].deliver);
        stat_info->tx_missed[i] = VNODE_STAT_READ(&st_cons_tx[i].missed);
        stat_info->tx_total_len[i] = VNODE_STAT_READ(&st_cons_tx[i].total_len);

        stat_info->ftx_on_line[i] = VNODE_STAT_READ(&st_cons_ftx[i].on_line);
        stat_info->ftx_deliver[i] = VNODE_STAT_READ(&st_cons_ftx[i].deliver);
        stat_info->ftx_missed[i] = VNODE_STAT_READ(&st_cons_ftx[i].missed);
        stat_info->ftx_total_len[i] = VNODE_STAT_READ(&st_cons_ftx[i].total_len);
    }

    return 0;
}

int vdev_data_instance_destory(struct vdev_instance * vdi)
{
    if (vdi->vnode_rx_cons != NULL)
        vnode_mirror_delete_cons(vdi->vnode_rx_cons);
    if (vdi->vnode_tx_prod != NULL)
        vnode_mirror_delete_prod(vdi->vnode_tx_prod);
    if (vdi->vnode_ftx_prod != NULL)
        vnode_mirror_delete_prod(vdi->vnode_ftx_prod);

    FREE(vdi);
    return 0;
}

/* 创建虚设备实例，应用使用 */
struct vdev_instance * vdev_data_instance_create(struct _vdev * _vdev, const char * appsym,
                                                        unsigned int nr_rxstream, unsigned int nr_txstream)
{
    struct vdev_instance * vdi = ZMALLOC(sizeof(struct vdev_instance));
    MR_VERIFY_MALLOC(vdi);

    vdi->vdev = &_vdev->vdev;

    if (nr_rxstream > 0)
    {
        vdi->vnode_rx_cons = vnode_mirror_create_cons(_vdev->vnode_rx, appsym, nr_rxstream);
        vdi->vnode_rx_cons_notify = vnode_mirror_notify_ctx_cons(vdi->vnode_rx_cons);
    }

    if (nr_txstream > 0)
    {
        vdi->vnode_tx_prod = vnode_mirror_create_prod(_vdev->vnode_tx, appsym, nr_txstream);
    }

    if (nr_txstream > 0)
    {
        vdi->vnode_ftx_prod = vnode_mirror_create_prod(_vdev->vnode_ftx, appsym, nr_txstream);
    }

    if (nr_rxstream > 0 && vdi->vnode_rx_cons == NULL)
    {
        MR_ERROR("Creating VDI for device %s failed: "
                 "Cannot create rx consumer, nr_rxstream = %d",
                 vdi->vdev->symbol, nr_rxstream);
        goto errout;
    }

    if (nr_txstream > 0 && vdi->vnode_tx_prod == NULL)
    {
        MR_ERROR("Creating VDI for device %s failed: "
                 "Cannot create TX producer, nr_txstream = %d",
                 vdi->vdev->symbol, nr_txstream);
        goto errout;
    }

    if (nr_txstream > 0 && vdi->vnode_ftx_prod == NULL)
    {
        MR_ERROR("Creating VDI for device %s failed: "
                 "Cannot create FTX producer, nr_txstream = %d",
                 vdi->vdev->symbol, nr_txstream);
        goto errout;
    }

    vdi->direct_pool = _vdev->direct_pool;
    vdi->indirect_pool = _vdev->indirect_pool;
    vdi->nr_rxstream = nr_rxstream;
    vdi->nr_txstream = nr_txstream;

    return vdi;

errout:
    vdev_data_instance_destory(vdi);
    return NULL;
}

static int vdev_data_destory(struct _vdev * _vdev)
{
    return 0;
}

int vdev_data_create(struct vdev_main * v_main, const char * symbol, unsigned int nr_rxstream, unsigned int nr_txstream,
                     unsigned int sz_tunnel_rx, unsigned int sz_tunnel_tx, unsigned int sz_max_inflight,
                     unsigned int batch_interval_in_us, struct rte_mempool * direct_pool)
{
    // 检查设备是否已经存在，不允许重复创建
    struct vdev * vdev_info = vdev_lookup(v_main, symbol);
    if (vdev_info != NULL)
    {
        MR_WARNING("vdev %s has been created. failed. \n", vdev_info->symbol);
        return RT_IGNORE;
    }

    // 申请Info结构体的空间
    struct _vdev * _vdev = ZMALLOC(sizeof(struct _vdev));
    MR_VERIFY_MALLOC(_vdev);

    /* 填充vdev信息 */
    vdev_info = &_vdev->vdev;

    /* 填充_vdev信息 */
    snprintf(vdev_info->symbol, sizeof(vdev_info->symbol), "%s", symbol);
    _vdev->nr_rxstream = nr_rxstream;
    _vdev->nr_txstream = nr_txstream;
    _vdev->sz_rx_tunnel = sz_tunnel_rx;
    _vdev->sz_tx_tunnel = sz_tunnel_tx;
    _vdev->sz_max_inflight = sz_max_inflight;
    _vdev->direct_pool = direct_pool;

    char vnode_sym_rx[MR_SYMBOL_MAX * 2];
    snprintf(vnode_sym_rx, sizeof(vnode_sym_rx), "%s-rx", vdev_info->symbol);

    char vnode_sym_tx[MR_SYMBOL_MAX * 2];
    snprintf(vnode_sym_tx, sizeof(vnode_sym_tx), "%s-tx", vdev_info->symbol);

    char vnode_sym_ftx[MR_SYMBOL_MAX * 2];
    snprintf(vnode_sym_ftx, sizeof(vnode_sym_ftx), "%s-ftx", vdev_info->symbol);

    char vnode_sym_ltx[MR_SYMBOL_MAX * 2];
    snprintf(vnode_sym_ltx, sizeof(vnode_sym_ltx), "%s-ltx", vdev_info->symbol);

    /* 创建VNODE */
    _vdev->vnode_rx = vnode_mirror_create(vnode_sym_rx, sz_tunnel_rx, sz_max_inflight, 1, batch_interval_in_us);
    _vdev->vnode_tx = vnode_mirror_create(vnode_sym_tx, sz_tunnel_tx, 0, 0, 0);
    _vdev->vnode_ftx = vnode_mirror_create(vnode_sym_ftx, sz_tunnel_tx, 0, 0, 0);

#define ERR_VERIFY(x, ...)                                                                                             \
    do                                                                                                                 \
    {                                                                                                                  \
        if (x == NULL)                                                                                                 \
        {                                                                                                              \
            MR_ERROR(__VA_ARGS__);                                                                                     \
            goto errout;                                                                                               \
        }                                                                                                              \
    } while (0)

    /* 错误校验 */
    ERR_VERIFY(_vdev->vnode_rx, "Create vdev %s rx vnode failed.", vdev_info->symbol);
    ERR_VERIFY(_vdev->vnode_tx, "Create vdev %s tx vnode failed.", vdev_info->symbol);
    ERR_VERIFY(_vdev->vnode_ftx, "Create vdev %s fast tx vnode failed. ", vdev_info->symbol);

    _vdev->vnode_rx_prod = vnode_mirror_create_prod(_vdev->vnode_rx, "sv", (int)nr_rxstream);
    _vdev->vnode_tx_cons = vnode_mirror_create_cons(_vdev->vnode_tx, "sv", (int)nr_txstream);
    _vdev->vnode_ftx_cons = vnode_mirror_create_cons(_vdev->vnode_ftx, "sv", (int)nr_txstream);

    /* 校验，申请VNODE、VNODE生产者、消费者是否成功 */
    ERR_VERIFY(_vdev->vnode_rx_prod, "Create vdev %s rx vnode producer failed. ", vdev_info->symbol);
    ERR_VERIFY(_vdev->vnode_tx_cons, "Create vdev %s tx vnode consumer failed. ", vdev_info->symbol);
    ERR_VERIFY(_vdev->vnode_ftx_cons, "Create vdev %s fast tx vnode consumer failed. ", vdev_info->symbol);

    /* 加入虚设备列表 */
    MR_VERIFY(v_main->vdev_max_idx <= RTE_DIM(v_main->_vdev_array));
    vdev_info->port_id = v_main->vdev_max_idx;
    v_main->_vdev_array[v_main->vdev_max_idx++] = _vdev;

    return RT_SUCCESS;

errout:
    vdev_data_destory(_vdev);
    rte_free(_vdev);
    return RT_ERR;

#undef ERR_VERIFY
}