#include <tunnat.h>
#include <tunnel.h>
#include <mrtunnat.h>
#include <session.h>
#include <array>

extern "C"
{
#include <marsio.h>
#include <rte_cycles.h>
#include <rte_ether.h>
}

#ifndef MR_TUNNAT_USE_PHONY_ETHER_HEADER
#define MR_TUNNAT_USE_PHONY_ETHER_HEADER    1
#endif

#ifndef IPV4_VERSION
#define IPV4_VERSION 0x04
#endif

#ifndef IPV6_VERSION
#define IPV6_VERSION 0X06
#endif


/* 特殊设备使用的保活协议 */
#define BFD_DEFAULT_PORT	(3784) /* 默认端口 */
struct bfd_header
{
    uint8_t version_diag;
    uint8_t flags;
    uint8_t detect_time_multiplier;
    uint8_t length;
    uint8_t my_discriminator[4];
    uint8_t your_discriminator[4];
    uint8_t desired_min_tx_interval[4];
    uint8_t required_min_rx_interval[4];
    uint8_t required_min_echo_interval[4];
};

static uint32_t __bfd_rehash_index = 0;

static SessionKey convert_addinfo_to_session_key(TunnatInstance * instance, const struct AddressInfo & addrinfo)
{
    in_addr_t s_port = instance->sess_tb_order_by_tuple4 ? addrinfo.s_port : 0;
    in_addr_t d_port = instance->sess_tb_order_by_tuple4 ? addrinfo.d_port : 0;

    if (addrinfo.type == ADDRESS_INFO_TYPE_IPV4)
    {
        return SessionKey(addrinfo.src_addr_v4, addrinfo.dst_addr_v4, s_port, d_port);
    }
    else if (addrinfo.type == ADDRESS_INFO_TYPE_IPV6)
    {
        return SessionKey(addrinfo.src_addr_v6, addrinfo.dst_addr_v6, s_port, d_port);
    }

    assert(0);
    return {};
}

static int __phy_to_virt_pkt_decap(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    marsio_buff_t * mbuf, TunnelContainerArray & tun_array)
{
    const char * pkt_ptr = marsio_buff_mtod(mbuf);
    unsigned int pkt_len = marsio_buff_datalen(mbuf);
    tunnel_type pkt_tun_type = TUNNEL_TYPE_UNKNOWN;

    assert(tun_array.sz_total_len == 0);
    tun_array.sz_array = 0;
    tun_array.sz_total_len = 0;

    /* 解外层隧道，不知道隧道类型，使用猜测的方式 */
    for (unsigned i = 0; i < tun_array.tun_array.size(); i++)
    {
        TunnelContainer & tun_container = tun_array.tun_array[i];

        tun_container.MbufMetaPreParse(mbuf);
        int ret = tun_container.PacketParse(pkt_ptr, pkt_len, pkt_tun_type);
        if (ret < 0) break;

        pkt_ptr += tun_container.ThisTunLength();
        pkt_len -= tun_container.ThisTunLength();
        pkt_tun_type = tun_container.NextTunType();

        tun_array.sz_array++;
        tun_array.sz_total_len += tun_container.ThisTunLength();

        if (pkt_tun_type == TUNNEL_TYPE_END) break;
    }

    /* 没有成功解析隧道，转发应用处理 */
    if (tun_array.sz_array == 0)
        return RT_ERR;

    return RT_SUCCESS;
}

static void __phy_to_virt_link_info_check(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    marsio_buff_t * mbuf, TunnelContainerArray &tun_array)
{
    LinkInfoTable * link_info_table = th_instance->link_info_table;
    tunnat_link_id_t link_id;
    int ret = link_info_table->LookupLinkID(tun_array, link_id);

    if(ret == -ENOENT)
    {
        link_info_table->AddTunnel(tun_array, link_id);
    }
    else if(ret < 0)
    {
        return;
    }

    auto * ctrlzone = static_cast<mr_tunnat_ctrlzone *>(marsio_buff_ctrlzone(mbuf, g_ctrlzone_id));
    ctrlzone->virtual_link_id = link_id;
}

static int __phy_to_virt_session_update_check(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    TunnelContainerArray & tun_array, SessionEntry * ss_entry)
{
    /* Only check if outer tunnel is changed */
    auto & outer_tunnel_in_ss = ss_entry->tun_array.GetOuterTunContainer();
    auto & outer_tunnel_in_tun = tun_array.GetOuterTunContainer();

    if (outer_tunnel_in_ss != outer_tunnel_in_tun)
    {
        TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_SESSION_NEED_UPDATE, 1);
    }

    return 0;
}

static int __phy_to_virt_session_check(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    marsio_buff_t * mbuf, TunnelContainerArray & tun_array)
{
    struct AddressInfo addrinfo;

    auto inner_tun_container = tun_array.GetInnerTunContainer();
    if (inner_tun_container.ThisTunType() != TUNNEL_TYPE_IPV4 &&
        inner_tun_container.ThisTunType() != TUNNEL_TYPE_IPV6)
    {
        return RT_ERR;
    }

    int ret = inner_tun_container.GetAddressInfo(addrinfo);
    if (ret < 0)
    {
        return RT_ERR;
    }

    SessionKey ss_key = convert_addinfo_to_session_key(instance, addrinfo);
    SessionEntry * ss_entry = th_instance->ss_table->Query(ss_key);

    if (ss_entry != nullptr)
    {
        if (instance->sess_tb_check_update)
        {
            __phy_to_virt_session_update_check(instance, th_instance, tun_array, ss_entry);
        }

        return RT_SUCCESS;
    }

    SessionEntry _temp_entry{tun_array};
    th_instance->ss_table->Add(ss_key, _temp_entry);

    return RT_SUCCESS;
}

static int __phy_to_virt_clear_cz(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    marsio_buff_t * mbuf)
{
    auto * ctrlzone = static_cast<mr_tunnat_ctrlzone *>(marsio_buff_ctrlzone(mbuf, g_ctrlzone_id));

    ctrlzone->__encap_type = 0;
    ctrlzone->__encap_len = 0;
    ctrlzone->action = 0;

    return RT_SUCCESS;
}

#include <ldbc.h>

/* 重哈希BFD报文，BFD按Round-Robin方式分配到下个应用的数据面线程 */
static void __phy_to_virt_check_bfd_and_rehash(TunnatInstance * instance,
    TunnatThreadInstance * th_instance,  marsio_buff_t * mbuf)
{
    const char * pkt_ptr = marsio_buff_mtod(mbuf);
    unsigned int pkt_len = marsio_buff_datalen(mbuf);

    /* 校验长度 */
    if(pkt_len < sizeof(struct ether_hdr) +
        sizeof(struct ipv4_hdr) + sizeof(struct udp_hdr)) return;

    const struct ether_hdr * eth_hdr = (const struct ether_hdr *)pkt_ptr;
    if(eth_hdr->ether_type != ntohs(ETHER_TYPE_IPv4)) return;

    const struct ipv4_hdr * ipv4_hdr = (const struct ipv4_hdr *)(eth_hdr + 1);
    if(ipv4_hdr->next_proto_id  != IPPROTO_UDP) return;

    unsigned int ipv4_hdr_len = (ipv4_hdr->version_ihl & 0xfU) * 4;
    const struct udp_hdr * udp_hdr = (const struct udp_hdr *)((const char *)ipv4_hdr + ipv4_hdr_len);
    if(udp_hdr->dst_port != ntohs(BFD_DEFAULT_PORT)) return;

    /* 设置重哈希值，采用Round-Robin方式 */
    marsio_buff_set_rehash_index(mbuf, __bfd_rehash_index++);
    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_BFD_INPUT, 1);

    (void)instance;
    (void)th_instance;
}

static void __phy_to_virt_pkt_fill_addressinfo(TunnatInstance * instance,
    struct mr_tunnat_ctrlzone * ctrlzone, TunnelContainerArray & tun_array)
{
    TunnelContainer * tunnel;
    struct AddressInfo addrinfo;

    switch(instance->ctrlzone_addr_info_type)
    {
        case MR_TUNNAT_CTRLZONE_ADDR_INFO_TUNNEL:
            tunnel = &tun_array.GetOuterTunContainer(); break;
        case MR_TUNNAT_CTRLZONE_ADDR_INFO_REAL:
            tunnel = tun_array.GetTunContainerWithType(TUNNEL_TYPE_ETHER); break;
        default:return;
    }

    if (unlikely(tunnel == nullptr || tunnel->GetAddressInfo(addrinfo)))
    {
        return;
    }

    /* MAC address */
    memcpy(ctrlzone->g_device_mac, addrinfo.src_mac_addr, sizeof(ctrlzone->g_device_mac));
    memcpy(ctrlzone->l_device_mac, addrinfo.dst_mac_addr, sizeof(ctrlzone->l_device_mac));

    /* Only when the outer type is g-vxlan, then fill the l4-address */
    if (tunnel->ThisTunType() == TUNNEL_TYPE_G_VXLAN)
    {
        ctrlzone->l4_src_port = addrinfo.s_port;
        ctrlzone->g_device_in_addr = addrinfo.src_addr_v4.s_addr;
        ctrlzone->l_device_in_addr = addrinfo.dst_addr_v4.s_addr;
        ctrlzone->g_device_vpn_id = addrinfo.tun_vpn_id;
        ctrlzone->g_device_linkpair = addrinfo.tun_link_id;
        ctrlzone->route_dir = addrinfo.dir;
    }
    else if(tunnel->ThisTunType() == TUNNEL_TYPE_VLAN_FLIPPING)
    {
        ctrlzone->route_dir = addrinfo.dir;
    }
}


/* 封装报文修改，去掉封装报文头部 */
static int __phy_to_virt_pkt_modify(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    marsio_buff_t * mbuf, TunnelContainerArray & tun_array)
{
    auto * ctrlzone = static_cast<mr_tunnat_ctrlzone *>(marsio_buff_ctrlzone(mbuf, g_ctrlzone_id));
    assert(ctrlzone != nullptr);

    /* 控制域内填写封装报文类型 */
    TunnelContainer & outer_tunnel = tun_array.GetOuterTunContainer();
    TunnelContainer & inner_tunnel = tun_array.GetInnerTunContainer();

    /* 只有当隧道信息解析成功时，才向控制域填充MAC地址等信息 */
    if (tun_array.sz_array > 0)
    {
        __phy_to_virt_pkt_fill_addressinfo(instance, ctrlzone, tun_array);
    }

    /* 对于内层隧道不是IPv4或IPv6的，不剥离外层隧道，直接送入应用 */
    if (inner_tunnel.ThisTunType() != TUNNEL_TYPE_IPV4 && inner_tunnel.ThisTunType() != TUNNEL_TYPE_IPV6)
    {
        ctrlzone->__encap_len = 0;
    }
    else
    {
        ctrlzone->__encap_len = (uint16_t)(tun_array.sz_total_len - inner_tunnel.ThisTunLength());
    }

    ctrlzone->__encap_type = outer_tunnel.ThisTunType();
    ctrlzone->action = TUNNAT_CZ_ACTION_FORWARD;

    /* 没有封装格式的，不执行剥离动作，送入应用 */
    if (ctrlzone->__encap_len == 0)
    {
        return RT_SUCCESS;
    }

    /* 去掉封装报文头部 */
    void * __adj_check = marsio_buff_adj(mbuf, ctrlzone->__encap_len);
    assert(__adj_check != nullptr);

    /* 构造伪以太网头部 */
    auto * p_ether_hdr = (struct ether_hdr *)marsio_buff_prepend(mbuf, sizeof(struct ether_hdr));
    assert(p_ether_hdr != nullptr);

    ctrlzone->__p_ether_type = p_ether_hdr->ether_type;
    switch (inner_tunnel.ThisTunType())
    {
    case TUNNEL_TYPE_IPV4:
        p_ether_hdr->ether_type = htons(ETHER_TYPE_IPv4);
        break;
    case TUNNEL_TYPE_IPV6:
        p_ether_hdr->ether_type = htons(ETHER_TYPE_IPv6);
        break;
    default:
        assert(0);
    }

    (void)__adj_check;
    return RT_SUCCESS;
}

static void __phy_to_virt_one_device(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    TunnatInstance::_devs * phy_device_hand, TunnatInstance::_devs * virt_device_hand)
{
    unsigned int tid = th_instance->thread_id;
    unsigned int nr_burst = instance->nr_burst;

    marsio_buff_t * mbufs[MR_BURST_MAX];

    int nr_mbufs = marsio_recv_burst(phy_device_hand->vdev_handler, tid, mbufs, nr_burst);
    if (nr_mbufs <= 0) return;

    /* 报文封装信息提取结果 */
    TunnelContainerArray * tun_container_array = th_instance->tun_container_array;
    unsigned int nr_encap_mbufs = 0;

    /* 报文解析，判断封装格式 */
    for (int i = 0; i < nr_mbufs; i++)
    {
        tun_container_array[i].Reset();

        int ret = __phy_to_virt_pkt_decap(instance, th_instance, mbufs[i],
            tun_container_array[i]);

        if (ret >= 0)
        {
            if(instance->is_use_link_info_table)
            {
                __phy_to_virt_link_info_check(instance, th_instance, mbufs[i],
                    tun_container_array[i]);
            }

            __phy_to_virt_session_check(instance, th_instance, mbufs[i],
                tun_container_array[i]);
            __phy_to_virt_pkt_modify(instance, th_instance, mbufs[i],
                tun_container_array[i]);

            nr_encap_mbufs++;
        }
        else
        {
            __phy_to_virt_clear_cz(instance, th_instance, mbufs[i]);
            __phy_to_virt_check_bfd_and_rehash(instance, th_instance, mbufs[i]);
        }
    }

    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_PKT_INPUT, nr_mbufs);
    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_PKT_ENCAP_INPUT, nr_encap_mbufs);
    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_PKT_RAW_INPUT, nr_mbufs - nr_encap_mbufs);

    /* 转发到应用 */
    marsio_send_burst(virt_device_hand->vdev_sendpath, tid, mbufs, nr_mbufs);
    return;
}

/* 转发报文解封装 */
static int __virt_to_phy_pkt_forward(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    marsio_buff_t *& mbuf)
{
    auto * ctrlzone = static_cast<mr_tunnat_ctrlzone *>(marsio_buff_ctrlzone(mbuf, g_ctrlzone_id));
    assert(ctrlzone != nullptr);

    /* 需要回注的报文全部携带forward标记，这个标记是送入应用时标记的 */
    if (!(ctrlzone->action & TUNNAT_CZ_ACTION_FORWARD))
    {
        return RT_ERR;
    }

    /* 对于有封装长度的，恢复伪以太网头部修改的数据，去掉以太网头部 */
    if (ctrlzone->__encap_len)
    {
        struct ether_hdr * p_eth_hdr = (struct ether_hdr *)marsio_buff_mtod(mbuf);
        p_eth_hdr->ether_type = ctrlzone->__p_ether_type;

        void * __adj_check = marsio_buff_adj(mbuf, sizeof(struct ether_hdr));
        assert(__adj_check != nullptr);
        marsio_buff_prepend(mbuf, ctrlzone->__encap_len);

        (void)__adj_check;
    }

    /* 回注前原始报文修正 */
    char * tun_start = marsio_buff_mtod(mbuf);
    auto tun_type = static_cast<tunnel_type>(ctrlzone->__encap_type);

    TunnelContainer::PacketForwardModify(tun_start, ctrlzone->__encap_len, tun_type);
    TunnelContainer::MbufMetaForwardModify(mbuf, tun_type);

    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_FORWARD_TUNNEL_OUTPUT, 1);
    return RT_SUCCESS;
}

static int __virt_to_phy_pkt_encap(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    marsio_buff_t *& mbuf)
{
    auto * ctrlzone = static_cast<mr_tunnat_ctrlzone *>(marsio_buff_ctrlzone(mbuf, g_ctrlzone_id));
    assert(ctrlzone != nullptr);

    /* 封装标记，内外层都不封装 */
    if (!((ctrlzone->action & TUNNAT_CZ_ACTION_ENCAP_INNER)
        && (ctrlzone->action & TUNNAT_CZ_ACTION_ENCAP_OUTER)))
    {
        return RT_ERR;
    }

	enum SessionKeyType ss_key_type;
	auto * ether_hdr = (struct ether_hdr *)marsio_buff_mtod(mbuf);
	if (ntohs(ether_hdr->ether_type) == ETHER_TYPE_IPv4)
    {
        ss_key_type = SESSION_KEY_TYPE_IPV4;
    }
    else if (ntohs(ether_hdr->ether_type) == ETHER_TYPE_IPv6)
    {
        ss_key_type = SESSION_KEY_TYPE_IPV6;
    }
    else
    {
        assert(0);
    }

    /* 去掉以太网头部，因为是应用自己造的包，所以没有备份的以太网类型数据 */
    /* 使用写时复制的报文修改接口，因为此时报文可能被其他人引用 */

    void * __adj_check;
    mbuf = marsio_buff_adj_cw(instance->mr_instance, mbuf, sizeof(struct ether_hdr), &__adj_check);
    
    assert(__adj_check != nullptr);
    assert(mbuf != nullptr);

    SessionKey ss_key;
    AddressInfo addressInfo;

    const char * pktptr = (const char *)marsio_buff_mtod(mbuf);
    unsigned int pktlen = marsio_buff_datalen(mbuf);

    /* 创建一个临时的隧道解析容器，获取地址信息
     * 然后将地址信息转换为SessionKey查询 */
    if (ss_key_type == SESSION_KEY_TYPE_IPV4)
    {

        TunInnerIPv4 tun_v4;
        int ret = tun_v4.PacketParse(pktptr, pktlen);
        if(unlikely(ret < 0)) goto __errout;

        tun_v4.GetAddressInfo(addressInfo);
        ss_key = convert_addinfo_to_session_key(instance, addressInfo);
    }
    else
    {
        TunInnerIPv6 tun_v6;
        int ret = tun_v6.PacketParse(pktptr, pktlen);
        if (unlikely(ret < 0)) goto __errout;

        tun_v6.GetAddressInfo(addressInfo);
        ss_key = convert_addinfo_to_session_key(instance, addressInfo);
    }

    /* 双向查询 */
    SessionEntry * ss_entry;
    if (instance->is_auto_reserve_tunnel)
    {
        ss_entry = th_instance->ss_table->QueryDualDirect(ss_key);
    }
    else
    {
        ss_entry = th_instance->ss_table->Query(ss_key);
    }

    //fprintf(stderr, "query src %s dst %s, result is %p\n", str_in_addr_src, str_in_addr_dst, ss_entry);
    if (ss_entry != nullptr)
    {
        goto session_ready;
    }
    
    /* 没有查询到，使用最近的一次隧道数据，随便找到一个专用设备回注 */
    if (instance->is_use_recent_tunnel)
    {
        ss_entry = th_instance->ss_table->QueryRecent(ss_key_type);
    }

    if (ss_entry != nullptr)
    {
        TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_ENCAP_USE_RECENT_SESSION, 1);
        goto session_ready;
    }

__errout:

    /* 还没有查询到，报告失败 */
    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_ENCAP_FAIL_NO_SESSION, 1);
    return RT_ERR;

session_ready:

    /* 已经查找到隧道上下文，构建隧道报文 */
    TunnelContainerArray & tun_array = ss_entry->tun_array;

    /* 逆序构建报文隧道，最内层隧道不构建 */
    for (int i = tun_array.sz_array - 2; i >= 0; i--)
    {
        TunnelContainer & __tun_container = tun_array.tun_array[i];
        __tun_container.MbufMetaConstruct(mbuf);

        auto __this_tun_length = static_cast<uint16_t>(__tun_container.ThisTunLength());
        const char * pkt_ptr = marsio_buff_prepend(mbuf, __this_tun_length);
        unsigned int pkt_len = marsio_buff_datalen(mbuf);
        __tun_container.PacketConstruct(pkt_ptr, pkt_len);
    }

    (void)__adj_check;
    return RT_SUCCESS;
}

static int __virt_to_phy_pkt_virtual_link_id(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    marsio_buff_t *& mbuf)
{
    mr_tunnat_ctrlzone * ctrlzone = static_cast<mr_tunnat_ctrlzone *>(
        marsio_buff_ctrlzone(mbuf, g_ctrlzone_id));

    assert(ctrlzone != nullptr);

    /* 封装标记，内外层都不封装 */
    if(!(ctrlzone->action & TUNNAT_CZ_ACTION_ENCAP_VIRTUAL_LINK_ID))
    {
        return RT_ERR;
    }

    /* 查虚拟链路表 */
    tunnat_link_id_t virtual_link_id = ctrlzone->virtual_link_id;
    LinkInfoTable * link_info_table = th_instance->link_info_table;
    TunnelContainerArray * tun_array = link_info_table->LookupTunnel(virtual_link_id);

    /* 报文找不到对应的隧道，发包失败 */
    if(!tun_array)
    {
        TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_VIRTUAL_LINK_ID_NO_SESSION, 1);
        return RT_ERR;
    }

    /* 去掉以太网头部，因为是应用自己造的包，所以没有备份的以太网类型数据 */
    /* 使用写时复制的报文修改接口，因为此时报文可能被其他人引用 */

    void * __adj_check;
    mbuf = marsio_buff_adj_cw(instance->mr_instance, mbuf, sizeof(struct ether_hdr), &__adj_check);

    assert(__adj_check != nullptr);
    assert(mbuf != nullptr);

    /* 逆序构建报文隧道，最内层隧道不构建 */
    for (int i = tun_array->sz_array - 2; i >= 0; i--)
    {
        TunnelContainer & __tun_container = tun_array->tun_array[i];
        __tun_container.MbufMetaConstruct(mbuf);

        auto __this_tun_length = static_cast<uint16_t>(__tun_container.ThisTunLength());

        const char * pkt_ptr = marsio_buff_prepend(mbuf, __this_tun_length);
        unsigned int pkt_len = marsio_buff_datalen(mbuf);
        tun_array->tun_array[i].PacketConstruct(pkt_ptr, pkt_len);
    }

    (void)__adj_check;
    return RT_SUCCESS;
}

static int __virt_to_phy_pkt_no_session(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    marsio_buff_t * mbuf)
{
    mr_tunnat_ctrlzone * ctrlzone = static_cast<mr_tunnat_ctrlzone *>(
        marsio_buff_ctrlzone(mbuf, g_ctrlzone_id));

    assert(ctrlzone != nullptr);

    /* 直接构建报文，不查会话表 */
    if (!(ctrlzone->action & TUNNAT_CZ_ACTION_ENCAP_NO_SESSION))
        return RT_ERR;

    /* 对于内层是以太报文的 */
    if (ctrlzone->g_device_inner_encap_type == TUNNAT_TUNNEL_TYPE_ETHER)
    {
        TunVxlan OuterVXLAN;
        int ret = OuterVXLAN.CtrlZoneParse(ctrlzone);
        if (ret < 0) return RT_ERR;

        char * pkt_ptr = marsio_buff_prepend(mbuf, OuterVXLAN.ThisTunLength());
        unsigned int pkt_len = marsio_buff_datalen(mbuf);

        assert(pkt_ptr != nullptr);
        OuterVXLAN.PacketConstruct(pkt_ptr, pkt_len);

        return RT_SUCCESS;
    }
    else if (ctrlzone->g_device_inner_encap_type == TUNNAT_TUNNEL_TYPE_PPP)
    {
        assert(0);
    }
    else if (ctrlzone->g_device_inner_encap_type == TUNNAT_TUNNEL_TYPE_HDLC)
    {
        assert(0);
    }

    return RT_ERR;
}

static void __virt_to_phy_one_device(TunnatInstance * instance, TunnatThreadInstance * th_instance,
    TunnatInstance::_devs * phy_device_hand, TunnatInstance::_devs * virt_device_hand)
{
    unsigned int tid = th_instance->thread_id;
    unsigned int nr_burst = instance->nr_burst;

    marsio_buff_t * mbufs[MR_BURST_MAX];

    int nr_mbufs = marsio_recv_burst(virt_device_hand->vdev_handler, tid, mbufs, nr_burst);
    if (nr_mbufs <= 0) return;

    marsio_buff_t * fwd_mbufs[MR_BURST_MAX];
    unsigned int nr_fwd_mbufs = 0;

    marsio_buff_t * encap_bufs[MR_BURST_MAX];
    unsigned int nr_encap_bufs = 0;

    marsio_buff_t * no_session_bufs[MR_BURST_MAX];
    unsigned int nr_no_session_bufs = 0;

    marsio_buff_t * direct_bufs[MR_BURST_MAX];
    unsigned int nr_direct_bufs = 0;

    marsio_buff_t * virtual_link_id_bufs[MR_BURST_MAX];
    unsigned int nr_virtual_link_id_bufs = 0;

    for (int i = 0; i < nr_mbufs; i++)
    {
        if(instance->is_use_link_info_table)
        {
            if (__virt_to_phy_pkt_virtual_link_id(instance, th_instance, mbufs[i]) == RT_SUCCESS)
            {
                virtual_link_id_bufs[nr_virtual_link_id_bufs++] = mbufs[i];
                continue;
            }
        }

	    if (__virt_to_phy_pkt_no_session(instance, th_instance, mbufs[i]) == RT_SUCCESS)
        {
            no_session_bufs[nr_no_session_bufs++] = mbufs[i];
            continue;
        }

        if (__virt_to_phy_pkt_forward(instance, th_instance, mbufs[i]) == RT_SUCCESS)
        {
            fwd_mbufs[nr_fwd_mbufs++] = mbufs[i];
            continue;
        }

        if (__virt_to_phy_pkt_encap(instance, th_instance, mbufs[i]) == RT_SUCCESS)
        {
            encap_bufs[nr_encap_bufs++] = mbufs[i];
            continue;
        }

        /* 其他情况，比如保活包，直接转发到专用设备 */
        direct_bufs[nr_direct_bufs++] = mbufs[i];
    }

    marsio_send_burst(phy_device_hand->vdev_sendpath, tid, fwd_mbufs, nr_fwd_mbufs);
    marsio_send_burst(phy_device_hand->vdev_sendpath, tid, encap_bufs, nr_encap_bufs);
    marsio_send_burst(phy_device_hand->vdev_sendpath, tid, virtual_link_id_bufs, nr_virtual_link_id_bufs);
    marsio_send_burst(phy_device_hand->vdev_sendpath, tid, no_session_bufs, nr_no_session_bufs);
    marsio_send_burst(phy_device_hand->vdev_sendpath, tid, direct_bufs, nr_direct_bufs);

    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_FORWARD_OUTPUT, nr_fwd_mbufs);
    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_ENCAP_OUTPUT, nr_encap_bufs);
    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_DIRECT_OUTPUT, nr_direct_bufs);
    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_NO_SESSION_OUTPUT, nr_no_session_bufs);
    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_VIRTUAL_LINK_ID_OUTPUT, nr_virtual_link_id_bufs);

    return;
}

void * tunnat_thread_loop(void * arg)
{
    TunnatThreadInstance * th_instance = (TunnatThreadInstance *)arg;
    TunnatInstance * instance = th_instance->instance;

    marsio_thread_init(instance->mr_instance);

    /* 注册统计对象 */
    TunnatThreadStat::ThreadStatObjectSet(th_instance->th_stat);

    /* 目前，只处理第一对物理设备、虚拟设备 */
    auto & virtdev = instance->virtdevs[0];
    auto & phydev = instance->phydevs[0];

    /* Idle Counter*/
    unsigned long idle_counter = 0;

    while (g_keep_running)
    {
        __phy_to_virt_one_device(instance, th_instance, &phydev, &virtdev);
        __virt_to_phy_one_device(instance, th_instance, &phydev, &virtdev);

        if(likely(instance->idle_threshold > 0) &&
           unlikely(idle_counter >= instance->idle_threshold))
        {
            marsio_send_burst_flush(virtdev.vdev_sendpath, th_instance->thread_id);
            marsio_send_burst_flush(phydev.vdev_sendpath, th_instance->thread_id);
            idle_counter = 0;
        }

        idle_counter++;
    }

    return (void *)nullptr;
}