path: root/service/src/devmgr.c

/// 物理设备管理器

#include <assert.h>
#include <net/if.h>
#include <stdio.h>
#include <sys/ioctl.h>
#include <sys/queue.h>
#include <unistd.h>

#include <rte_bus_pci.h>
#include <rte_config.h>
#include <rte_debug.h>
#include <rte_ethdev.h>
#include <rte_ether.h>
#include <rte_flow.h>
#include <rte_graph.h>
#include <rte_kni.h>
#include <rte_malloc.h>
#include <rte_mbuf.h>
#include <rte_node_eth_api.h>
#include <rte_pci.h>
#include <rte_string_fns.h>

#include <MESA_prof_load.h>
#include <cJSON.h>
#include <common.h>
#include <sc_common.h>
#include <sc_devmgr.h>
#include <sc_mrb.h>
#include <sc_vdev.h>

static struct rte_eth_conf eth_conf_default = {
    .rxmode =
        {
            .mq_mode = ETH_MQ_RX_NONE,
            .split_hdr_size = 0,
        },
    .txmode =
        {
            .mq_mode = ETH_MQ_TX_NONE,
        },
};

static uint8_t default_sym_rss_key[40] = {0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a,
                                          0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a,
                                          0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a,
                                          0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a, 0x6d, 0x5a};

static const char * mr_dev_driver_as_str[] = {
    [MR_DEV_DRV_TYPE_DPDK_PCI] = "dpdk_pci",
    [MR_DEV_DRV_TYPE_DPDK_VIRTIO_USER] = "dpdk_virtio_user",
    [MR_DEV_DRV_TYPE_DPDK_AF_PACKET] = "dpdk_af_packet",
    [MR_DEV_DRV_TYPE_SHMDEV] = "shmdev",
};

static const char * mr_dev_role_as_str[] = {
    [MR_DEV_ROLE_NONE] = "none",
    [MR_DEV_ROLE_VWIRE_INTERFACE] = "virtual wire",
    [MR_DEV_ROLE_TAP_INTERFACE] = "tap",
    [MR_DEV_ROLE_ENDPOINT_INTERFACE] = "endpoint",
    [MR_DEV_ROLE_NF_INTERFACE] = "network function",
    [MR_DEV_ROLE_SYSTEM_INTERFACE] = "system",
};

struct mr_dev_ops
{
};

struct dpdk_dev_candidate
{
    TAILQ_ENTRY(dpdk_dev_candidate) next;
    /* kernel name */
    char kernel_name[MR_SYMBOL_MAX];
    /* vdev name, for dpdk vdevs */
    char dpdk_vdev_name[MR_SYMBOL_MAX];
    /* PCI address */
    struct rte_pci_addr pci_addr;
    /* STR PCI address */
    char str_pci_addr[PCI_PRI_STR_SIZE];
    /* vendor */
    char vendor[MR_SYMBOL_MAX];
    /* driver */
    char driver[MR_SYMBOL_MAX];
};

struct devmgr_main
{
    /* dpdk devices candidate */
    TAILQ_HEAD(dpdk_dev_candidate_db_head, dpdk_dev_candidate) dpdk_dev_candidate_list;
    /* device desc for all kind of driver */
    struct mr_dev_desc * dev_descs[MR_DEVICE_MAX];
    /* shmdev port_id counter */
    unsigned int shmdev_port_id_counter;

    /* hardware info for dpdk devices,
     * this file is generated by the helper script before the main process run */
    struct cJSON * j_hwfile;
    /* sc_main handler */
    struct sc_main * sc;
};

static const char * str_rssmode(unsigned int rssmode)
{
    switch (rssmode)
    {
    case MR_DEV_RSSMODE_DEFAULT:
        return "Default";
    case MR_DEV_RSSMODE_2TUPLE_SYM:
        return "SAddr/DAddr(Sym)";
    case MR_DEV_RSSMODE_4TUPLE_SYM:
        return "SAddr/DAddr/SPort/DPort(Sym)";
    case MR_DEV_RSSMODE_4TUPLE_ASYM:
        return "SAddr/DAddr/SPort/DPort(Asym)";
    default:
        return "Unknown";
    }
}

static const char * str_enable_or_disable(unsigned int value)
{
    if (value)
        return "ENABLE";
    else
        return "DISABLE";
}

static int dpdk_dev_status_print(struct dpdk_dev * dev)
{
    char str_phy_addr[MR_SYMBOL_MAX];
    rte_ether_format_addr(str_phy_addr, sizeof(str_phy_addr), &dev->ether_addr);

    MR_INFO(" ");
    MR_INFO("DPDK based device %s: PortID = %d", dev->symbol, dev->port_id);
    MR_INFO("   HWADDR                         : %s", str_phy_addr);
    MR_INFO("   Maximum Transmission Unit      : %u", dev->mtu);
    MR_INFO("   Promisc                        : %s", str_enable_or_disable(dev->promisc));
    MR_INFO("   VLAN-Filter                    : %s", str_enable_or_disable(dev->en_vlan_filter));
    MR_INFO("   VLAN-Strip                     : %s", str_enable_or_disable(dev->en_vlan_strip));
    MR_INFO("   Drop-En                        : %s", str_enable_or_disable(dev->en_drop));
    MR_INFO("   RSSMode                        : %s", str_rssmode(dev->rssmode));

    return 0;
}

struct mr_dev_desc * mr_dev_desc_create(struct devmgr_main * devmgr_main, const char * devsym)
{
    struct mr_dev_desc * dev_desc = ZMALLOC(sizeof(struct mr_dev_desc));
    MR_VERIFY_MALLOC(dev_desc);

    snprintf(dev_desc->symbol, sizeof(dev_desc->symbol) - 1, "%s", devsym);
    dev_desc->rx_node_id = RTE_NODE_ID_INVALID;
    dev_desc->tx_node_id = RTE_NODE_ID_INVALID;
    dev_desc->port_id = (uint32_t)(-1);
    return dev_desc;
}

int mr_dev_desc_config_load(struct devmgr_main * devmgr_main, struct mr_dev_desc * dev_desc)
{
    const char * cfgfile = devmgr_main->sc->local_cfgfile;
    char str_section[MR_SYMBOL_MAX] = {};

    snprintf(str_section, sizeof(str_section) - 1, "device:%s", dev_desc->symbol);
    MESA_load_profile_uint_def(cfgfile, str_section, "driver", &dev_desc->drv_type, MR_DEV_DRV_TYPE_DPDK_PCI);
    MESA_load_profile_uint_def(cfgfile, str_section, "role", &dev_desc->role_type, MR_DEV_ROLE_NONE);

    /* for endpoint and route, try to load ip addr settings */
    char str_in_addr[INET_ADDRSTRLEN];
    if (MESA_load_profile_string_nodef(cfgfile, str_section, "in_addr", str_in_addr, sizeof(str_in_addr)) < 0)
    {
        return RT_SUCCESS;
    }

    int ret = inet_pton(AF_INET, str_in_addr, &dev_desc->in_addr);
    if (ret < 0)
    {
        MR_CFGERR_INVALID_FORMAT(cfgfile, str_section, "in_addr");
        return RT_ERR;
    }

    char str_in_mask[INET_ADDRSTRLEN];
    if (MESA_load_profile_string_nodef(cfgfile, str_section, "in_mask", str_in_mask, sizeof(str_in_mask)) < 0)
    {
        return RT_ERR;
    }

    ret = inet_pton(AF_INET, str_in_mask, &dev_desc->in_mask);
    if (ret < 0)
    {
        MR_CFGERR_INVALID_FORMAT(cfgfile, str_section, "in_mask");
        return RT_ERR;
    }

    char str_gateway[INET_ADDRSTRLEN];
    if (MESA_load_profile_string_nodef(cfgfile, str_section, "gateway", str_gateway, sizeof(str_gateway)) >= 0)
    {
        ret = inet_pton(AF_INET, str_gateway, &dev_desc->in_gateway);
        if (ret < 0)
        {
            MR_CFGERR_INVALID_FORMAT(cfgfile, str_section, "gateway");
            return RT_ERR;
        }
    }

    return RT_SUCCESS;
}

struct mr_dev_desc * mr_dev_desc_lookup(struct devmgr_main * devmgr_main, const char * devsym)
{
    for (unsigned int i = 0; i < RTE_DIM(devmgr_main->dev_descs); i++)
    {
        struct mr_dev_desc * dev_desc = devmgr_main->dev_descs[i];
        if (dev_desc == NULL || strcasecmp(dev_desc->symbol, devsym) != 0)
        {
            continue;
        }

        return dev_desc;
    }

    return NULL;
}

struct mr_dev_desc * mr_dev_desc_iterate(struct devmgr_main * devmgr_main, unsigned int * iterator)
{
    for (; *iterator < RTE_DIM(devmgr_main->dev_descs); (*iterator)++)
    {
        struct mr_dev_desc * dev_desc = devmgr_main->dev_descs[*iterator];
        if (dev_desc == NULL)
            continue;

        (*iterator)++;
        return dev_desc;
    }

    return NULL;
}

struct mr_dev_desc * mr_dev_desc_lookup_by_port_id(struct devmgr_main * devmgr_main, port_id_t port_id)
{
    return devmgr_main->dev_descs[port_id];
}

static char ** gcfg_device_syms_get_by_drv(struct sc_main * sc, enum mr_dev_driver drv_type, unsigned int * nr_drvs)
{
    char str_dev_list[MR_STRING_MAX] = {};
    MESA_load_profile_string_nodef(sc->local_cfgfile, "device", "device", str_dev_list, sizeof(str_dev_list));

    /* 从列表中抽取每一个网卡名 */
    char * str_dev_tokens[MR_TOKENS_MAX] = {};
    int nr_str_tokens = rte_strsplit(str_dev_list, sizeof(str_dev_list), str_dev_tokens, MR_TOKENS_MAX, ',');
    if (nr_str_tokens < 0)
    {
        return NULL;
    }

    char ** out_dev_symbols = malloc(MR_TOKENS_MAX * sizeof(char *));
    unsigned int nr_out_dev_symbols = 0;

    /* 遍历所有dev设备 */
    for (int i = 0; i < nr_str_tokens; i++)
    {
        /* device name */
        char * str_dev_symbol = str_dev_tokens[i];
        assert(str_dev_symbol != NULL);

        /* read the driver setting */
        char str_section[MR_SYMBOL_MAX];
        snprintf(str_section, sizeof(str_section) - 1, "device:%s", str_dev_symbol);

        unsigned int cfg_driver = 0;
        MESA_load_profile_uint_def(sc->local_cfgfile, str_section, "driver", &cfg_driver, MR_DEV_DRV_TYPE_DPDK_PCI);

        if (cfg_driver != drv_type)
            continue;

        out_dev_symbols[nr_out_dev_symbols] = strdup(str_dev_symbol);
        nr_out_dev_symbols++;
    }

    *nr_drvs = nr_out_dev_symbols;
    return out_dev_symbols;
}

void shmdev_config_load(struct devmgr_main * devmgr_main, const char * devsym, unsigned int * sz_tunnel,
                        unsigned int * sz_buffer, char mempool_direct_pool[MR_SYMBOL_MAX],
                        char mempool_indirect_pool[MR_SYMBOL_MAX])
{
    unsigned int default_sz_tunnel;
    unsigned int default_sz_buffer;
    const char * cfgfile = devmgr_main->sc->local_cfgfile;

    MESA_load_profile_uint_def(cfgfile, "device", "sz_tunnel", &default_sz_tunnel, 4096);
    MESA_load_profile_uint_def(cfgfile, "device", "sz_buffer", &default_sz_buffer, 32);

    char str_dev_section[MR_SYMBOL_MAX];
    snprintf(str_dev_section, sizeof(str_dev_section), "device:%s", devsym);

    MESA_load_profile_uint_def(cfgfile, str_dev_section, "sz_tunnel", sz_tunnel, default_sz_tunnel);
    MESA_load_profile_uint_def(cfgfile, str_dev_section, "sz_buffer", sz_buffer, default_sz_buffer);

    MESA_load_profile_string_def(cfgfile, str_dev_section, "direct-pool", mempool_direct_pool, MR_SYMBOL_MAX, devsym);
    MESA_load_profile_string_def(cfgfile, str_dev_section, "indirect-pool", mempool_indirect_pool, MR_SYMBOL_MAX,
                                 mempool_direct_pool);
}

int shmdev_setup_one_device(struct devmgr_main * devmgr_main, const char * devsym)
{
    /* prepare the dev_desc */
    struct mr_dev_desc * dev_desc = mr_dev_desc_create(devmgr_main, devsym);
    int ret = mr_dev_desc_config_load(devmgr_main, dev_desc);

    if (unlikely(ret < 0))
    {
        MR_ERROR("Failed at loading config for the shmdev device %s", devsym);
        return RT_ERR;
    }

    /* TODO: check the name conflict */
    unsigned int sz_tunnel = UINT32_MAX;
    unsigned int sz_buffer = UINT32_MAX;
    char sym_direct_mempool[MR_SYMBOL_MAX] = {0};
    char sym_indirect_mempool[MR_SYMBOL_MAX] = {0};

    shmdev_config_load(devmgr_main, devsym, &sz_tunnel, &sz_buffer, sym_direct_mempool, sym_indirect_mempool);
    assert(sz_tunnel != UINT32_MAX & sz_buffer != UINT32_MAX);

    struct sc_main * sc = devmgr_main->sc;

    /* TODO: allow set different i/o thread for each shmdev */
    unsigned int nr_rxstream = sc->nr_io_thread;
    unsigned int nr_txstream = sc->nr_io_thread;

    /* Get indirect pool */
    struct rte_mempool * direct_pool = mrb_direct_mempool_locate(sc->mrb_pool_main, sym_direct_mempool, 0, 0);
    if (direct_pool == NULL)
    {
        MR_ERROR("Direct mempool %s for virtual device %s is not existed. ", sym_direct_mempool, devsym);
        return RT_ERR;
    }

    struct rte_mempool * indirect_pool = mrb_indirect_mempool_locate(sc->mrb_pool_main, sym_indirect_mempool, 0, 0);
    if (indirect_pool == NULL)
    {
        MR_ERROR("Indirect mempool %s for virtual device %s is not existed. ", sym_indirect_mempool, devsym);
        return RT_ERR;
    }

    ret = vdev_data_create(sc->vdev_main, devsym, nr_rxstream, nr_txstream, sz_tunnel, sz_buffer, direct_pool,
                           indirect_pool);
    if (unlikely(ret < 0))
    {
        MR_ERROR("Failed at creating shmdev resources for device %s", devsym);
        return RT_ERR;
    }

    struct vdev * shmdev_handle = vdev_lookup(sc->vdev_main, devsym);
    MR_VERIFY_2(shmdev_handle != NULL, "vdev_lookup() returns NULL");

    dev_desc->port_id = devmgr_main->shmdev_port_id_counter++;
    dev_desc->drv_type = MR_DEV_DRV_TYPE_SHMDEV;
    dev_desc->shm_dev_desc = shmdev_handle;

    /* add to index array */
    devmgr_main->dev_descs[dev_desc->port_id] = dev_desc;
    return RT_SUCCESS;
}

int shmdev_init(struct devmgr_main * devmgr_main)
{
    /* query the shmdev list */
    unsigned int nr_devsyms = 0;
    char ** devsyms = gcfg_device_syms_get_by_drv(devmgr_main->sc, MR_DEV_DRV_TYPE_SHMDEV, &nr_devsyms);

    if (nr_devsyms == 0)
    {
        MR_INFO("No shmdev configuration found, ignore it.");
        return RT_SUCCESS;
    }

    for (unsigned int i = 0; i < nr_devsyms; i++)
    {
        int ret = shmdev_setup_one_device(devmgr_main, devsyms[i]);
        if (unlikely(ret < 0))
        {
            MR_ERROR("Failed at init shmdev %s.", devsyms[i]);
            return RT_ERR;
        }
    }

    return RT_SUCCESS;
}

/* Set Virtio Tap Up */
int vhost_dev_setup(const char * virtio_sym)
{
    int fd;
    struct ifreq ifr;

    fd = socket(AF_INET, SOCK_DGRAM, 0);
    strncpy(ifr.ifr_name, virtio_sym, IFNAMSIZ - 1);

    if (ioctl(fd, SIOCGIFFLAGS, &ifr) != 0)
    {
        return RT_ERR;
    }

    ifr.ifr_flags |= IFF_UP | IFF_RUNNING;
    if (ioctl(fd, SIOCSIFFLAGS, &ifr) != 0)
    {
        return RT_ERR;
    }
    close(fd);
    return RT_SUCCESS;
}

/* 用户参数解析：网卡参数设置 */
static int gen_dpdk_dev_ethconf(struct dpdk_dev * dev, unsigned nr_rxq_use, struct rte_eth_conf * out_eth_conf)
{
    struct mr_dev_desc * dev_desc = dev->ref_dev_desc;
    struct rte_eth_conf eth_conf = eth_conf_default;

    /* PCI devices */
    if ((dev_desc->drv_type == MR_DEV_DRV_TYPE_DPDK_PCI) && (nr_rxq_use > 1))
    {
        /* only PCI devices can run at RSS mode. */
        eth_conf.rxmode.mq_mode = ETH_MQ_RX_RSS;

        /* setup how NICs distributes packets */
        if (dev->rssmode == MR_DEV_RSSMODE_2TUPLE_SYM)
        {
            eth_conf.rx_adv_conf.rss_conf.rss_hf = ETH_RSS_IP;
            eth_conf.rx_adv_conf.rss_conf.rss_key = default_sym_rss_key;
            eth_conf.rx_adv_conf.rss_conf.rss_key_len = sizeof(default_sym_rss_key);
        }
        else if (dev->rssmode == MR_DEV_RSSMODE_4TUPLE_SYM)
        {
            eth_conf.rx_adv_conf.rss_conf.rss_hf = ETH_RSS_NONFRAG_IPV4_TCP | ETH_RSS_NONFRAG_IPV6_TCP |
                                                   ETH_RSS_NONFRAG_IPV4_UDP | ETH_RSS_NONFRAG_IPV6_UDP;
            eth_conf.rx_adv_conf.rss_conf.rss_key = default_sym_rss_key;
            eth_conf.rx_adv_conf.rss_conf.rss_key_len = sizeof(default_sym_rss_key);
        }
        else if (dev->rssmode == MR_DEV_RSSMODE_4TUPLE_ASYM)
        {
            eth_conf.rx_adv_conf.rss_conf.rss_hf = ETH_RSS_NONFRAG_IPV4_TCP | ETH_RSS_NONFRAG_IPV6_TCP |
                                                   ETH_RSS_NONFRAG_IPV4_UDP | ETH_RSS_NONFRAG_IPV6_UDP;
            eth_conf.rx_adv_conf.rss_conf.rss_key = NULL;
        }
    }
    else if (dev_desc->drv_type == MR_DEV_DRV_TYPE_DPDK_VIRTIO_USER ||
             dev_desc->drv_type == MR_DEV_DRV_TYPE_DPDK_AF_PACKET)
    {
        /* the virtio and af_packet is not support rss */
        eth_conf.rxmode.mq_mode = ETH_MQ_RX_NONE;
    }

    if (dev->en_vlan_strip)
    {
        eth_conf.txmode.offloads |= DEV_TX_OFFLOAD_VLAN_INSERT;
    }

    *out_eth_conf = eth_conf;
    return 0;
}

static int dpdk_dev_queue_setup_rss_or_ctx(struct dpdk_dev * dev, socket_id_t socket_id, struct rte_mempool * pool,
                                           unsigned int nr_rxq_use, unsigned int nr_txq_use,
                                           unsigned int rxq_index_begin, unsigned int txq_index_begin)
{
    /* TODO: use different mem pool according to the socket id of i/o core */

    /* setup the rxconf and txconf for queue */
    struct rte_eth_dev_info dev_info;
    rte_eth_dev_info_get(dev->port_id, &dev_info);

    /* default rxconf and txconf */
    struct rte_eth_rxconf rxconf = dev_info.default_rxconf;
    struct rte_eth_txconf txconf = dev_info.default_txconf;

    /* suggestion from dpdk's pmd */
    struct rte_eth_dev_portconf * default_rxportconf = &dev_info.default_rxportconf;
    struct rte_eth_dev_portconf * default_txportconf = &dev_info.default_txportconf;

    if (dev->nr_rx_descs > 0)
    {
        dev->nr_rx_descs = RTE_MIN(dev->nr_rx_descs, dev_info.rx_desc_lim.nb_max);
        dev->nr_rx_descs = RTE_MAX(dev->nr_rx_descs, dev_info.rx_desc_lim.nb_min);
    }
    else
    {
        dev->nr_rx_descs = default_rxportconf->ring_size;
    }

    if (dev->nr_tx_descs > 0)
    {
        dev->nr_tx_descs = RTE_MIN(dev->nr_tx_descs, dev_info.tx_desc_lim.nb_max);
        dev->nr_tx_descs = RTE_MAX(dev->nr_tx_descs, dev_info.tx_desc_lim.nb_min);
    }
    else
    {
        dev->nr_tx_descs = default_txportconf->ring_size;
    }

    int ret = 0;

    for (unsigned int rxq = rxq_index_begin; rxq < rxq_index_begin + nr_rxq_use; rxq++)
    {
        ret = rte_eth_rx_queue_setup(dev->port_id, rxq, dev->nr_rx_descs, socket_id, &rxconf, pool);
        if (ret < 0)
        {
            MR_ERROR("dpdk device %s RXQ %d setup failed, errno = %d", dev->symbol, rxq, ret);
            goto err;
        }
    }

    for (unsigned int txq = txq_index_begin; txq < txq_index_begin + nr_txq_use; txq++)
    {
        ret = rte_eth_tx_queue_setup(dev->port_id, txq, dev->nr_tx_descs, socket_id, &txconf);
        if (ret < 0)
        {
            MR_ERROR("dpdk device %s TXQ %d setup failed, errno = %d", dev->symbol, txq, ret);
            goto err;
        }
    }

    return 0;

err:
    return ret;
}

static int dpdk_dev_queue_setup_hairpin(struct dpdk_dev * dev, unsigned int nr_hairpin_q, unsigned int rxq_index_begin,
                                        unsigned int txq_index_begin)
{
    /* create hairpin queues on both ports*/
    unsigned int q_index_hairpin = 0;
    unsigned int q_index_hairpin_peer = 0;

    struct rte_eth_hairpin_conf hairpin_conf = {
        .peer_count = 1,
        .manual_bind = 0,
        .tx_explicit = 0,
    };

    int ret = 0;

    for (q_index_hairpin = rxq_index_begin, q_index_hairpin_peer = txq_index_begin;
         q_index_hairpin < rxq_index_begin + nr_hairpin_q; q_index_hairpin++, q_index_hairpin_peer++)
    {
        hairpin_conf.peers[0].port = dev->port_id;
        hairpin_conf.peers[0].queue = q_index_hairpin_peer;

        MR_DEBUG("Prepare to setup rx hairpin for device %s, hairpin_queue_id = %d, peer_hairpin_queue_id = %d",
                 dev->symbol, q_index_hairpin, q_index_hairpin_peer);

        ret = rte_eth_rx_hairpin_queue_setup(dev->port_id, q_index_hairpin, dev->nr_rx_descs, &hairpin_conf);
        if (unlikely(ret != 0))
        {
            MR_ERROR("Failed at setup rx hairpin queue at port = %d, queue = %d, peer_queue = %d, ret = %d: %s",
                     dev->port_id, q_index_hairpin, q_index_hairpin_peer, ret, rte_strerror(errno));
            return ret;
        }
    }

    for (q_index_hairpin = txq_index_begin, q_index_hairpin_peer = rxq_index_begin;
         q_index_hairpin < txq_index_begin + nr_hairpin_q; q_index_hairpin++, q_index_hairpin_peer++)
    {
        hairpin_conf.peers[0].port = dev->port_id;
        hairpin_conf.peers[0].queue = q_index_hairpin_peer;

        MR_DEBUG("Prepare to setup tx hairpin for device %s, hairpin_queue_id = %d, peer_hairpin_queue_id = %d",
                 dev->symbol, q_index_hairpin, q_index_hairpin_peer);

        ret = rte_eth_tx_hairpin_queue_setup(dev->port_id, q_index_hairpin, dev->nr_tx_descs, &hairpin_conf);
        if (unlikely(ret != 0))
        {
            MR_ERROR("Failed at setup tx hairpin queue at port = %d, queue = %d, peer_queue = %d, ret = %d: %s",
                     dev->port_id, q_index_hairpin, q_index_hairpin_peer, ret, rte_strerror(errno));
            return ret;
        }

        dev->hairpin_q = q_index_hairpin;
    }

    MR_INFO("device %s hairpin setup successfully.", dev->symbol);
    return 0;
}

/* A matrix can be used to do symmetric hash */
static const uint8_t symmetric_rss_key[] = {
    0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A,
    0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A,
    0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A, 0x6D, 0x5A,
};

static int dpdk_dev_setup_default_flows(struct devmgr_main * devmgr_main, struct dpdk_dev * dev)
{
    /* -----------------  GROUP 0 -------------------- */
    struct rte_flow * flow = NULL;
    struct rte_flow_attr attr = {
        .group = 0,   /* set the rule on the main group. */
        .ingress = 1, /* Rx flow. */
        .priority = 3,
    };

    struct rte_flow_error flow_error = {};

    /* Define the pattern to match the packet */
    struct rte_flow_item pattern[] = {
        [0] = {.type = RTE_FLOW_ITEM_TYPE_ETH},
        [1] = {.type = RTE_FLOW_ITEM_TYPE_END},
    };

    /* Jump parameters */
    struct rte_flow_action_jump jump_to_group_1 = {.group = 1};

    /* Jump actions */

    struct rte_flow_action actions[] = {
        [0] =
            {
                .type = RTE_FLOW_ACTION_TYPE_JUMP,
                .conf = &jump_to_group_1,
            },
        [1] =
            {
                .type = RTE_FLOW_ACTION_TYPE_END,
                .conf = NULL,
            },
    };

    /* FLOW 0 */
    flow = rte_flow_create(dev->port_id, &attr, pattern, actions, &flow_error);
    if (unlikely(flow == NULL))
    {
        MR_ERROR("Failed at install default rule for device %s: %s", dev->symbol, flow_error.message);
        goto err;
    }

    dev->default_flow_handles[dev->nr_default_flow_handles++] = flow;

    /* ------------------ GROUP 1 ----------------------- */
    attr.group = 1;
    attr.ingress = 1;
    attr.priority = 3;

    memset(&flow_error, 0, sizeof(flow_error));
    // memset(&pattern, 0, sizeof(pattern));
    memset(&actions, 0, sizeof(actions));

    uint16_t target_queue_id[RTE_MAX_QUEUES_PER_PORT] = {};
    for (unsigned int i = 0; i < dev->nr_rxq; i++)
    {
        target_queue_id[i] = i;
    }

    uint64_t rss_type = 0;
    const uint8_t * rss_key = NULL;
    size_t rss_key_len = 0;

    /* setup how nics distributes packets */
    if (dev->rssmode == MR_DEV_RSSMODE_2TUPLE_SYM)
    {
        rss_type = ETH_RSS_IP;
        rss_key = default_sym_rss_key;
        rss_key_len = sizeof(default_sym_rss_key);
    }
    else if (dev->rssmode == MR_DEV_RSSMODE_4TUPLE_SYM)
    {
        rss_type =
            ETH_RSS_NONFRAG_IPV4_TCP | ETH_RSS_NONFRAG_IPV6_TCP | ETH_RSS_NONFRAG_IPV4_UDP | ETH_RSS_NONFRAG_IPV6_UDP;
        rss_key = default_sym_rss_key;
        rss_key_len = sizeof(default_sym_rss_key);
    }
    else if (dev->rssmode == MR_DEV_RSSMODE_4TUPLE_ASYM)
    {
        rss_type =
            ETH_RSS_NONFRAG_IPV4_TCP | ETH_RSS_NONFRAG_IPV6_TCP | ETH_RSS_NONFRAG_IPV4_UDP | ETH_RSS_NONFRAG_IPV6_UDP;
        rss_key = NULL;
    }

    /* TODO: maybe we have much more better way to do the symmetric hash */
    struct rte_flow_action_rss rss = {
        .level = 0,
        .queue = target_queue_id,
        .queue_num = dev->nr_rxq,
        .types = rss_type,
        .key = rss_key,
        .key_len = rss_key_len,
    };

    actions[0].type = RTE_FLOW_ACTION_TYPE_RSS;
    actions[0].conf = &rss;
    actions[1].type = RTE_FLOW_ACTION_TYPE_END;
    actions[1].conf = NULL;

    flow = rte_flow_create(dev->port_id, &attr, pattern, actions, &flow_error);
    if (unlikely(flow == NULL))
    {
        MR_ERROR("Failed at install RSS flow at device %s: %s", dev->symbol, flow_error.message);
        return RT_ERR;
    }

    /* Save the default flows, these rules should be destroy when the device is stop */
    dev->default_flow_handles[dev->nr_default_flow_handles++] = flow;
    return RT_SUCCESS;

err:
    return RT_ERR;
}

static int dpdk_dev_setup_common(struct devmgr_main * devmgr_main, struct dpdk_dev * dev)
{
    int ret = 0;
    struct sc_main * sc = devmgr_main->sc;
    socket_id_t socket_id = rte_eth_dev_socket_id(dev->port_id);

    struct rte_mempool * direct_pool = NULL;
    struct rte_mempool * indirect_pool = NULL;
    direct_pool = mrb_direct_mempool_locate(sc->mrb_pool_main, dev->str_direct_pool, socket_id, 0);
    indirect_pool = mrb_indirect_mempool_locate(sc->mrb_pool_main, dev->str_indirect_pool, socket_id, 0);

    if (direct_pool == NULL || indirect_pool == NULL)
    {
        MR_ERROR("dpdk device %s setup failed: Cannot find suitable mempool"
                 " on socket %d, core %d",
                 dev->symbol, socket_id, 0);
        return RT_ERR;
    }

    unsigned nr_rxq_use = 0;
    unsigned nr_txq_use = 0;

    // TODO: 目前实现按每个服务线程处理一个队列考虑
    dev->nr_rxq = CPU_COUNT(&sc->cpu_set_io);
    dev->nr_txq = CPU_COUNT(&sc->cpu_set_io);

    /* SmartOffload CPUSETS */
    dev->nr_ctx_rxq = CPU_COUNT(&sc->cpu_set_offload);
    dev->nr_ctx_txq = CPU_COUNT(&sc->cpu_set_offload);

    /*
        Check this NIC can support hairpin or not.
        The SmartOffload can not use when hairpin mode is unavailable.
    */
    if (dev->en_smartoffload)
    {
        struct rte_eth_hairpin_cap hairpin_cap = {};
        ret = rte_eth_dev_hairpin_capability_get(dev->port_id, &hairpin_cap);
        if (unlikely(ret != 0))
        {
            MR_ERROR("dpdk device %s hairpin mode is unavailable, smartoffload can not be enabled: ret = %d, %s",
                     dev->symbol, ret, rte_strerror(ret));
            return RT_ERR;
        }
    }

    /*
        Need to setup haripin when enable smartoffload,
        For now, for each port, there's only one rx queue and tx queue for hairpin.
    */
    dev->nr_hairpin_q = dev->en_smartoffload ? 1 : 0;
    nr_rxq_use = dev->nr_rxq + dev->nr_ctx_rxq + dev->nr_hairpin_q;
    nr_txq_use = dev->nr_txq + dev->nr_ctx_txq + dev->nr_hairpin_q;

    // 配置端口信息
    struct rte_eth_conf local_eth_conf;
    gen_dpdk_dev_ethconf(dev, nr_rxq_use, &local_eth_conf);

    ret = rte_eth_dev_configure(dev->port_id, nr_rxq_use, nr_txq_use, &local_eth_conf);

    if (ret != 0)
    {
        MR_ERROR("dpdk device %s configure error: %s, errno = %d", dev->symbol, strerror(ret), ret);
        return ret;
    }

    unsigned int nr_rxq_index = 0;
    unsigned int nr_txq_index = 0;

    /* Configure the RX, TX queues */
    if (dev->nr_rxq > 0 || dev->nr_txq > 0)
    {
        ret = dpdk_dev_queue_setup_rss_or_ctx(dev, socket_id, direct_pool, dev->nr_rxq, dev->nr_txq, nr_rxq_index,
                                              nr_txq_index);

        if (ret)
            return RT_ERR;
    }

    nr_rxq_index += dev->nr_rxq;
    nr_txq_index += dev->nr_txq;

    /* Configure the RX, TX queues for control and exception path */
    if (dev->nr_ctx_rxq > 0 || dev->nr_ctx_txq > 0)
    {
        ret = dpdk_dev_queue_setup_rss_or_ctx(dev, socket_id, direct_pool, dev->nr_ctx_rxq, dev->nr_ctx_txq,
                                              nr_rxq_index, nr_txq_index);

        if (ret)
            return RT_ERR;
    }

    nr_rxq_index += dev->nr_ctx_rxq;
    nr_txq_index += dev->nr_ctx_txq;

    /* Configure the RX, TX queues for same port hairpin */
    if (dev->nr_hairpin_q > 0)
    {
        ret = dpdk_dev_queue_setup_hairpin(dev, dev->nr_hairpin_q, nr_rxq_index, nr_txq_index);
        if (ret)
            return RT_ERR;
    }

    /* MTU */
    if (dev->mtu != 0 && (ret = rte_eth_dev_set_mtu(dev->port_id, dev->mtu)) < 0)
    {
        MR_WARNING("dpdk device %s MTU setup failed : %s", dev->symbol, strerror(-ret));
    }

    /* Multicast */
    if (dev->allmulticast)
    {
        rte_eth_allmulticast_enable(dev->port_id);
    }
    else
    {
        rte_eth_allmulticast_disable(dev->port_id);
    }

    /* 混杂模式设置 */
    if (dev->promisc)
    {
        rte_eth_promiscuous_enable(dev->port_id);
    }
    else
    {
        rte_eth_promiscuous_disable(dev->port_id);
    }

    ret = rte_eth_dev_start(dev->port_id);
    if (ret < 0)
    {
        MR_ERROR("dpdk device %s start failed, Errno = %d(%s)", dev->symbol, ret, strerror(-ret));
        return ret;
    }

    /* setup default flows, the default flows need to install after the device start */
    /* for now, only devices which smartoffload is enabled will be installed default flows,
       because the vf interfaces does not support jump action */

    if (dev->en_smartoffload)
    {
        ret = dpdk_dev_setup_default_flows(devmgr_main, dev);
        if (ret)
            return RT_ERR;
    }

    dev->nr_rxq = nr_rxq_use;
    dev->nr_txq = nr_txq_use;

    rte_eth_dev_get_mtu(dev->port_id, &dev->mtu);
    rte_eth_macaddr_get(dev->port_id, &dev->ether_addr);
    dev->promisc = rte_eth_promiscuous_get(dev->port_id);

    int vlan_offload_mask = 0;
    if (dev->en_vlan_strip)
    {
        vlan_offload_mask |= ETH_VLAN_STRIP_OFFLOAD;
    }

    if (dev->en_vlan_filter)
    {
        vlan_offload_mask |= ETH_VLAN_FILTER_OFFLOAD;
    }

    if (vlan_offload_mask != 0)
    {
        if ((ret = rte_eth_dev_set_vlan_offload(dev->port_id, vlan_offload_mask)) < 0)
        {
            MR_WARNING("dpdk device %s VLAN offload mask setup failed: mask=%x, %s", dev->symbol, vlan_offload_mask,
                       strerror(-ret));
        }
    }

    /* VLAN Filter设置 */
    if (dev->en_vlan_filter)
    {
        for (unsigned int i = 0; i < dev->nr_vlan_id_allow; i++)
        {
            uint16_t vlan_id = (uint16_t)(dev->vlan_id_allow[i]);
            if ((ret = rte_eth_dev_vlan_filter(dev->port_id, vlan_id, 1)) < 0)
            {
                MR_WARNING("dpdk device %s VLAN filter allow vlan-id %d setup failed: %s", dev->symbol, vlan_id,
                           strerror(-ret));
            }
        }

        for (unsigned int i = 0; i < dev->nr_vlan_id_deny; i++)
        {
            uint16_t vlan_id = (uint16_t)(dev->vlan_id_deny[i]);
            if ((ret = rte_eth_dev_vlan_filter(dev->port_id, vlan_id, 0)) < 0)
            {
                MR_WARNING("dpdk device %s VLAN filter deny vlan-id %d setup failed: %s", dev->symbol, vlan_id,
                           strerror(-ret));
            }
        }
    }

    dpdk_dev_status_print(dev);
    dev->inited = 1;
    dev->enable = 1;
    return 0;
}

/* 从全局配置文件读硬件配置信息 */
void dpdk_dev_config_load(struct dpdk_dev * dev_dpdk, const char * cfg)
{
    char str_section[MR_SYMBOL_MAX];
    snprintf(str_section, sizeof(str_section), "device:%s", dev_dpdk->symbol);

    MESA_load_profile_uint_def(cfg, str_section, "vlan-strip", &dev_dpdk->en_vlan_strip, 0);
    MESA_load_profile_uint_def(cfg, str_section, "vlan-filter", &dev_dpdk->en_vlan_filter, 0);

    // 允许通过的VLAN ID列表，仅当vlan-filter开启后有效

    int ret = 0;
    ret = MESA_load_profile_uint_range(cfg, str_section, "vlan-id-allow", RTE_DIM(dev_dpdk->vlan_id_allow),
                                       dev_dpdk->vlan_id_allow);

    if (ret >= 0)
    {
        dev_dpdk->nr_vlan_id_allow = ret;
    }

    // 禁止通过的VLAN ID列表，仅当vlan-filter开启后有效
    ret = MESA_load_profile_uint_range(cfg, str_section, "vlan-id-deny", RTE_DIM(dev_dpdk->vlan_id_deny),
                                       dev_dpdk->vlan_id_deny);

    if (ret >= 0)
    {
        dev_dpdk->nr_vlan_id_deny = ret;
    }

    // 丢包选项
    MESA_load_profile_uint_def(cfg, str_section, "drop_en", &dev_dpdk->en_drop, 0);
    // 分流模式
    MESA_load_profile_uint_def(cfg, str_section, "rssmode", &dev_dpdk->rssmode, 0);
    // RX描述符数量
    MESA_load_profile_uint_def(cfg, str_section, "nr_rxdesc", &dev_dpdk->nr_rx_descs, 0);
    // TX描述符数量
    MESA_load_profile_uint_def(cfg, str_section, "nr_txdesc", &dev_dpdk->nr_tx_descs, 0);
    // 读MTU，网卡自适应。
    MESA_load_profile_short_def(cfg, str_section, "mtu", (short *)&dev_dpdk->mtu, 0);
    // 读混杂模式
    MESA_load_profile_uint_def(cfg, str_section, "promisc", &dev_dpdk->promisc, 0);
    // MULTICAST
    MESA_load_profile_uint_def(cfg, str_section, "allmulticast", &dev_dpdk->allmulticast, 0);
    // SMARTOFFLOAD
    MESA_load_profile_uint_def(cfg, str_section, "en_smartoffload", &dev_dpdk->en_smartoffload, 0);

    // which mempool used
    memset(dev_dpdk->str_direct_pool, 0, sizeof(dev_dpdk->str_direct_pool));
    memset(dev_dpdk->str_indirect_pool, 0, sizeof(dev_dpdk->str_indirect_pool));

    MESA_load_profile_string_def(cfg, str_section, "direct-pool", dev_dpdk->str_direct_pool,
                                 sizeof(dev_dpdk->str_direct_pool), "");
    MESA_load_profile_string_def(cfg, str_section, "indirect-pool", dev_dpdk->str_indirect_pool,
                                 sizeof(dev_dpdk->str_indirect_pool), "");

    /* TODO: allow the user to set different io cores for each devices */
}

static struct dpdk_dev_candidate * dpdk_dev_candidate_lookup_by_pci_addr(struct devmgr_main * devmgr_main,
                                                                         struct rte_pci_addr pci_addr)
{
    struct dpdk_dev_candidate * dev_iter = NULL;
    TAILQ_FOREACH(dev_iter, &devmgr_main->dpdk_dev_candidate_list, next)
    {
        if (rte_pci_addr_cmp(&dev_iter->pci_addr, &pci_addr) == 0)
            return dev_iter;
    }

    return NULL;
}

static struct dpdk_dev_candidate * dpdk_dev_candidate_lookup_by_vdev_name(struct devmgr_main * devmgr_main,
                                                                          const char * vdev_name)
{
    struct dpdk_dev_candidate * dev_iter = NULL;
    TAILQ_FOREACH(dev_iter, &devmgr_main->dpdk_dev_candidate_list, next)
    {
        if (strcmp(dev_iter->dpdk_vdev_name, vdev_name) == 0)
            return dev_iter;
    }

    return NULL;
}

static int dpdk_dev_candidate_vdev_register(struct devmgr_main * devmgr_main, const char * kernel_name,
                                            const char * vdev_name)
{
    struct dpdk_dev_candidate * dev = malloc(sizeof(struct dpdk_dev_candidate));
    MR_VERIFY_MALLOC(dev);

    snprintf(dev->kernel_name, sizeof(dev->kernel_name) - 1, "%s", kernel_name);
    snprintf(dev->dpdk_vdev_name, sizeof(dev->dpdk_vdev_name) - 1, "%s", vdev_name);
    TAILQ_INSERT_TAIL(&devmgr_main->dpdk_dev_candidate_list, dev, next);

    return RT_SUCCESS;
}

static int dpdk_dev_candidate_pci_register(struct devmgr_main * devmgr_main, const char * devsym)
{
    /* 在JSON文件中查找设备，根据网卡名称查找 */
    cJSON * j_hwfile = devmgr_main->j_hwfile;
    cJSON * j_dev = NULL;

    if (j_hwfile == NULL)
    {
        return RT_SUCCESS;
    }

    cJSON_ArrayForEach(j_dev, j_hwfile)
    {
        /* 先获取设备名称 */
        cJSON * j_interface = cJSON_GetObjectItem(j_dev, "Interface");
        MR_VERIFY(j_interface != NULL);

        if (strcasecmp(devsym, j_interface->valuestring) != 0)
            continue;

        /* 获取PCI号 */
        cJSON * j_slot = cJSON_GetObjectItem(j_dev, "Slot");
        MR_VERIFY(j_slot != NULL);

        /* Driver */
        cJSON * j_driver = cJSON_GetObjectItem(j_dev, "Driver_str");
        MR_VERIFY(j_driver != NULL);

        struct dpdk_dev_candidate * dev = malloc(sizeof(struct dpdk_dev_candidate));
        MR_VERIFY_MALLOC(dev);

        snprintf(dev->kernel_name, sizeof(dev->kernel_name) - 1, "%s", j_interface->valuestring);
        snprintf(dev->driver, sizeof(dev->driver) - 1, "%s", j_driver->valuestring);
        snprintf(dev->str_pci_addr, sizeof(dev->str_pci_addr) - 1, "%s", j_slot->valuestring);

        /* parse the pci addr */
        int ret = rte_pci_addr_parse(j_slot->valuestring, &dev->pci_addr);
        if (unlikely(ret < 0))
        {
            MR_ERROR("failed at parsing device %s's pci_addr from hwfile, ignore.", dev->kernel_name);
            free(dev);
            continue;
        }

        MR_DEBUG("DPDK device candidate: kernel_name=%s, pci_addr=%s, driver=%s", dev->kernel_name, dev->str_pci_addr,
                 dev->driver);

        TAILQ_INSERT_TAIL(&devmgr_main->dpdk_dev_candidate_list, dev, next);
        return RT_SUCCESS;
    }

    return RT_NOEXIST;
}

/* 早期设备扫描，从HWFILE中获取设备定义的信息 */
static int dpdk_dev_early_scan(struct devmgr_main * devmgr_main)
{
    char str_device[MR_STRING_MAX];

    unsigned int nr_dev_symbols = 0;
    char ** str_dev_symbols = gcfg_device_syms_get_by_drv(devmgr_main->sc, MR_DEV_DRV_TYPE_DPDK_PCI, &nr_dev_symbols);

    /* 遍历所有dev设备 */
    for (int i = 0; i < nr_dev_symbols; i++)
    {
        int ret = dpdk_dev_candidate_pci_register(devmgr_main, str_dev_symbols[i]);
        if (unlikely(ret < 0))
        {
            MR_WARNING("device %s cannot be used as dpdk_pci device: not existed in the hwfile.", str_dev_symbols[i]);
            continue;
        }
    }

    for (int i = 0; i < nr_dev_symbols; i++)
    {
        free(str_dev_symbols[i]);
    }

    free(str_dev_symbols);
    return RT_SUCCESS;
}

static int dpdk_dev_af_packet_attach(struct devmgr_main * devmgr_main)
{
    /* TODO */
    return RT_SUCCESS;
}

/* VIRTIO Devices */
static int dpdk_dev_virtio_user_attach(struct devmgr_main * devmgr_main)
{
    /* Get all virtio_user devices */
    char ** virtio_devsyms = NULL;
    unsigned int nr_virtio_devsyms = 0;

    /* main handle */
    struct sc_main * sc = devmgr_main->sc;
    assert(sc != NULL);

    /* query the virtio device symbols */
    virtio_devsyms = gcfg_device_syms_get_by_drv(sc, MR_DEV_DRV_TYPE_DPDK_VIRTIO_USER, &nr_virtio_devsyms);
    if (unlikely(nr_virtio_devsyms < 0))
    {
        return RT_SUCCESS;
    }

    for (unsigned int i = 0; i < nr_virtio_devsyms; i++)
    {
        uint16_t port_id;

        char vdev_name[MR_STRING_MAX];
        /* dpdk会根据vdev_name获取对应的驱动，所以针对virtio类型的设备，此处命名必须为virtio_user开头 */
        snprintf(vdev_name, sizeof(vdev_name) - 1, "virtio_user%d", i);

        char vdev_args[MR_STRING_MAX];
        snprintf(vdev_args, sizeof(vdev_args) - 1, "queues=%d,queue_size=1024,path=/dev/vhost-net,iface=%s",
                 sc->nr_io_thread, virtio_devsyms[i]);

        int ret = rte_eal_hotplug_add("vdev", vdev_name, vdev_args);
        if (ret < 0)
        {
            MR_ERROR("Attaching virtio-user device %s failed, errno = %d", virtio_devsyms[i], ret);
            return RT_ERR;
        }

        ret = vhost_dev_setup(virtio_devsyms[i]);
        if (ret < 0)
        {
            MR_ERROR("Cannot set the tap to up for device: %s\n", virtio_devsyms[i]);
            return RT_ERR;
        }

        ret = rte_eth_dev_get_port_by_name(vdev_name, &port_id);
        if (ret < 0)
        {
            rte_eal_hotplug_remove("vdev", vdev_name);
            MR_ERROR("Cannot find added vdev %s failed, errno = %d", virtio_devsyms[i], ret);
            return RT_ERR;
        }

        dpdk_dev_candidate_vdev_register(devmgr_main, virtio_devsyms[i], vdev_name);
        MR_INFO("virtio_user device: %s attach successful, port_id=%d", virtio_devsyms[i], port_id);
    }

    for (unsigned int i = 0; i < nr_virtio_devsyms; i++)
    {
        free(virtio_devsyms[i]);
    }

    free(virtio_devsyms);
    return RT_SUCCESS;
}

void * dpdk_dev_link_state_update_thread(void * arg);

int dpdk_dev_setup_from_candidate(struct devmgr_main * devmgr_main, struct dpdk_dev_candidate * dev_can,
                                  port_id_t port_id)
{
    struct dpdk_dev * dev_dpdk = ZMALLOC(sizeof(struct dpdk_dev));
    MR_VERIFY_MALLOC(dev_dpdk);

    /* copy name and pci address from candidate */
    dev_dpdk->port_id = port_id;
    strncpy(dev_dpdk->symbol, dev_can->kernel_name, sizeof(dev_dpdk->symbol) - 1);
    memcpy(&dev_dpdk->pci_addr, &dev_can->pci_addr, sizeof(dev_dpdk->pci_addr));

    /* next, create the dev desc */
    struct mr_dev_desc * dev_desc = mr_dev_desc_create(devmgr_main, dev_dpdk->symbol);
    if (unlikely(dev_desc == NULL))
    {
        MR_ERROR("Failed at create device descriptor for device %s", dev_dpdk->symbol);
        return RT_ERR;
    }

    /* load config */
    int ret = mr_dev_desc_config_load(devmgr_main, dev_desc);
    if (unlikely(ret < 0))
    {
        MR_ERROR("Failed at loading config for the dpdk device %s", dev_dpdk->symbol);
        return RT_ERR;
    }

    /* attach dpdk_dev structure to dev_desc */
    assert(dev_desc->drv_type == MR_DEV_DRV_TYPE_DPDK_PCI || dev_desc->drv_type == MR_DEV_DRV_TYPE_DPDK_VIRTIO_USER ||
           dev_desc->drv_type == MR_DEV_DRV_TYPE_DPDK_AF_PACKET);

    /* create dual-link between dev_desc and dev_dpdk */
    dev_desc->port_id = port_id;
    dev_desc->dpdk_dev_desc = dev_dpdk;
    dev_dpdk->ref_dev_desc = dev_desc;

    /* read device's capability */
    struct rte_eth_dev_info dev_info;
    rte_eth_dev_info_get(port_id, &dev_info);

    /* default mac addr */
    rte_eth_macaddr_get(dev_dpdk->port_id, &dev_dpdk->ether_addr);
    rte_eth_dev_get_mtu(dev_dpdk->port_id, &dev_dpdk->mtu);

    /* load user settings */
    const char * cfgfile = devmgr_main->sc->local_cfgfile;
    dpdk_dev_config_load(dev_dpdk, cfgfile);

    ret = dpdk_dev_setup_common(devmgr_main, dev_dpdk);
    if (unlikely(ret < 0))
    {
        MR_ERROR("Failed at setup the dpdk device %s", dev_dpdk->symbol);
        return RT_ERR;
    }

    /* join the device list */
    assert(devmgr_main->dev_descs[port_id] == NULL);
    devmgr_main->dev_descs[port_id] = dev_desc;

    return RT_SUCCESS;
}

int dpdk_dev_init(struct devmgr_main * devmgr_main)
{
    /* setup all dpdk devices, at first, iterate the pci bus */
    port_id_t port_id;
    int ret = 0;

    /* all ethernet devices connect to pci bus */
    RTE_ETH_FOREACH_DEV(port_id)
    {
        struct rte_eth_dev_info dev_info_handler;
        ret = rte_eth_dev_info_get(port_id, &dev_info_handler);
        MR_VERIFY(ret == 0);

        struct rte_device * p_device = dev_info_handler.device;
        struct rte_bus * p_bus = rte_bus_find_by_device(p_device);
        MR_VERIFY(p_bus != NULL);

        struct dpdk_dev_candidate * dev_can = NULL;
        if (strncmp(p_bus->name, "pci", strlen("pci")) == 0)
        {
            dev_can = dpdk_dev_candidate_lookup_by_pci_addr(devmgr_main, RTE_DEV_TO_PCI_CONST(p_device)->addr);
        }
        else if (strncmp(p_bus->name, "vdev", strlen("vdev")) == 0)
        {
            dev_can = dpdk_dev_candidate_lookup_by_vdev_name(devmgr_main, p_device->name);
        }
        else
        {
            MR_ERROR("Unknown p_bus for the dpdk device, ignore it: port_id=%d, p_bus=%s", port_id, p_bus->name);
            continue;
        }

        if (unlikely(dev_can == NULL))
        {
            MR_ERROR("Failed at lookup dpdk_dev_candidate, ignore it: port_id=%d", port_id);
            continue;
        }

        ret = dpdk_dev_setup_from_candidate(devmgr_main, dev_can, port_id);
        if (ret < 0)
        {
            MR_ERROR("Failed at dpdk device setup from candidate: port_id = %d", port_id);
            continue;
        }

        /* success, remove the candidate from list */
        TAILQ_REMOVE(&devmgr_main->dpdk_dev_candidate_list, dev_can, next);
    }

    return RT_SUCCESS;
}

void dpdk_dev_stat_get(struct dpdk_dev * dev, struct dpdk_dev_stats * dpdk_dev_stat)
{
    rte_eth_stats_get(dev->port_id, &dev->stat.rte);
    *dpdk_dev_stat = dev->stat;
}

void dpdk_dev_stat_last_save(struct dpdk_dev * dev, struct dpdk_dev_stats * dpdk_dev_stat_last)
{
    dev->stat_last = *dpdk_dev_stat_last;
}

void dpdk_dev_stat_last_get(struct dpdk_dev * dev, struct dpdk_dev_stats * dpdk_dev_stat_last)
{
    *dpdk_dev_stat_last = dev->stat_last;
}

void * dpdk_dev_link_state_update_thread(void * arg)
{
    struct devmgr_main * devmgr_main = (struct devmgr_main *)arg;
    pthread_detach(pthread_self());
    mr_thread_setname(pthread_self(), "MRZCPD_DPDK_DEV_LINK_UPDATE");

    while (g_keep_running)
    {
        for (unsigned int i = 0; i < RTE_DIM(devmgr_main->dev_descs); i++)
        {
            struct mr_dev_desc * dev_desc = devmgr_main->dev_descs[i];
            if (dev_desc == NULL || dev_desc->dpdk_dev_desc == NULL)
                continue;

            struct rte_eth_link now_eth_link;
            struct dpdk_dev * dpdk_dev = dev_desc->dpdk_dev_desc;

            rte_eth_link_get_nowait(dpdk_dev->port_id, &now_eth_link);
            struct rte_eth_link old_eth_link = dpdk_dev->link_status;

            /* the rte_eth_link must be uint64 */
            static_assert(sizeof(struct rte_eth_link) == sizeof(uint64_t), "the struct rte_eth_link must be uint64");

            /* print log when the link status changed */
            if (*(uint64_t *)(&now_eth_link) != *(uint64_t *)(&old_eth_link))
            {
                char str_eth_link_status[MR_STRING_MAX];
                rte_eth_link_to_str(str_eth_link_status, sizeof(str_eth_link_status) - 1, &now_eth_link);

                MR_INFO("device %s link status changed: %s", dev_desc->symbol, str_eth_link_status);
            }

            /* remember the new status */
            dpdk_dev->link_status = now_eth_link;
        }

        sleep(1);
    }

    return (void *)0;
}

static int devmgr_hwfile_load(struct devmgr_main * devmgr_main, const char * hwfile_path)
{
    FILE * f = fopen(hwfile_path, "rb");
    if (f == NULL)
    {
        MR_ERROR("Cannot open hardware file %s: %s", hwfile_path, strerror(errno));
        return RT_ERR;
    }

    fseek(f, 0, SEEK_END);
    long fsize = ftell(f);
    fseek(f, 0, SEEK_SET); // same as rewind(f);

    char * string = malloc(fsize + 1);
    MR_VERIFY_MALLOC(string);

    /* 读文件 */
    fread(string, fsize, 1, f);
    fclose(f);

    string[fsize] = 0;

    cJSON * j_hwfile = cJSON_Parse(string);
    if (j_hwfile == NULL)
    {
        MR_ERROR("Hardware file %s parse failed. ", hwfile_path);
        return RT_ERR;
    }

    devmgr_main->j_hwfile = j_hwfile;
    return RT_SUCCESS;
}

/* 在EAL启动前调用，生成ARGC, ARGV等参数 */
int devmgr_early_init(struct sc_main * sc)
{
    assert(sc->devmgr_main == NULL);

    /* EAL环境还没有初始化，用malloc申请内存 */
    sc->devmgr_main = malloc(sizeof(struct devmgr_main));
    MR_VERIFY_MALLOC(sc->devmgr_main);

    struct devmgr_main * devmgr_main = sc->devmgr_main;
    memset(devmgr_main, 0, sizeof(struct devmgr_main));

    TAILQ_INIT(&devmgr_main->dpdk_dev_candidate_list);
    devmgr_hwfile_load(devmgr_main, sc->local_hwfile);

    /* the shmdev's port_id is after the dpdk_dev's ports */
    devmgr_main->shmdev_port_id_counter = RTE_MAX_ETHPORTS;
    devmgr_main->sc = sc;

    /* dpdk pci devices need to be pre-parsed before eal_init */
    return dpdk_dev_early_scan(devmgr_main);
}

void devmgr_eal_args_generate(struct devmgr_main * devmgr_main, char * eal_argv[], unsigned int * eal_argc,
                              unsigned int max_argc)
{
    struct dpdk_dev_candidate * dev_can = NULL;
    unsigned int nr_eal_argv = 0;

    TAILQ_FOREACH(dev_can, &devmgr_main->dpdk_dev_candidate_list, next)
    {
        /* for now, all the candidate devices are PCI devices,
         * all the PCi devices need to be in the whitelist */

        char * str_eal_lead = NULL;
        asprintf(&str_eal_lead, "%s", "-a");
        eal_argv[(*eal_argc)++] = str_eal_lead;

        char * str_dev_option = NULL;
        asprintf(&str_dev_option, "%s", dev_can->str_pci_addr);

        /* for the mlx5/6, disable the duplicated flow feature */
        if (strcasecmp(dev_can->driver, "mlx5_core") == 0)
        {
            asprintf(&str_dev_option, "%s,%s", str_dev_option, "allow_duplicate_pattern=0");
        }

        eal_argv[(*eal_argc)++] = str_dev_option;
    }
}

void devmgr_deinit(struct devmgr_main * devmgr_main)
{
    for (unsigned int i = 0; i < RTE_DIM(devmgr_main->dev_descs); i++)
    {
        struct mr_dev_desc * dev_desc = devmgr_main->dev_descs[i];
        if (dev_desc == NULL)
            continue;

        /* for dpdk based devices */
        if (dev_desc->dpdk_dev_desc != NULL)
        {
            struct dpdk_dev * dev_dpdk = dev_desc->dpdk_dev_desc;
            rte_eth_dev_stop(dev_dpdk->port_id);
            rte_eth_dev_close(dev_dpdk->port_id);
        }

        /* for other devices, do nothing */
        MR_INFO("device %s (port_id=%d) closed.", dev_desc->symbol, dev_desc->port_id);
    }
}

int devmgr_init(struct devmgr_main * devmgr_main)
{
    int ret = 0;

    /* attach the virtio_user */
    ret = dpdk_dev_virtio_user_attach(devmgr_main);
    if (unlikely(ret < 0))
    {
        MR_ERROR("Failed at attach virtio_user devices.");
        return RT_ERR;
    }

    ret = dpdk_dev_af_packet_attach(devmgr_main);
    if (unlikely(ret < 0))
    {
        MR_ERROR("Failed at attach af_packet devices.");
        return RT_ERR;
    }

    ret = dpdk_dev_init(devmgr_main);
    if (unlikely(ret < 0))
    {
        MR_ERROR("Failed at setup dpdk devices.");
        return RT_ERR;
    }

    ret = shmdev_init(devmgr_main);
    if (unlikely(ret < 0))
    {
        MR_ERROR("Failed at setup shared-memory devices.");
        return RT_ERR;
    }

    /* shared-memory devices */

    /* 启动物理设备状态更新线程 */
    pthread_t _pid_link_update;
    ret = pthread_create(&_pid_link_update, NULL, dpdk_dev_link_state_update_thread, (void *)devmgr_main);
    if (ret < 0)
    {
        MR_ERROR("PHYDEV link state update thread create failed: %s", strerror(errno));
        return RT_ERR;
    }

    return RT_SUCCESS;
}