path: root/tunnat/src/runtime.cc


#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
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];

        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 (tun_array.sz_array == 0)
        return RT_ERR;

    tunnel_type inner_tun_type = tun_array.GetInnerTunContainer().ThisTunType();

    /* 内层不是IPv4和IPv6的，无法处理，转到应用解析 */
    if (inner_tun_type != TUNNEL_TYPE_IPV4 && inner_tun_type != TUNNEL_TYPE_IPV6)
        return RT_ERR;

    return RT_SUCCESS;
}

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

    int ret = tun_array.GetInnerTunContainer().GetAddressInfo(addr_info);
    if (ret < 0) return RT_ERR;
	switch(addr_info.type)
	{
	case IPV4_ADDR:
		session_key.type=KEY_TYPE_IPV4;
		session_key.ipv4_key.s_in_addr.s_addr=addr_info.addr.ipv4_addr.src_in_addr.s_addr;
		session_key.ipv4_key.d_in_addr.s_addr=addr_info.addr.ipv4_addr.dst_in_addr.s_addr;
		break;
	case IPV6_ADDR:
		session_key.type=KEY_TYPE_IPV6;
		memcpy(&session_key.ipv6_key.s_in6_addr,&addr_info.addr.ipv6_addr.src_in6_addr,sizeof(struct in6_addr));
		memcpy(&session_key.ipv6_key.d_in6_addr,&addr_info.addr.ipv6_addr.dst_in6_addr,sizeof(struct in6_addr));
		break;
	default:
		assert(0);
		break;
	}
   
    /* 查询当前会话表 */
    SessionEntry * ss_entry = th_instance->ss_table->Query(session_key);
    if (ss_entry != NULL) return RT_SUCCESS;

    SessionEntry _temp_entry;
    _temp_entry.tun_array = tun_array;

    th_instance->ss_table->Add(session_key, _temp_entry);
    return RT_SUCCESS;
}

static int __phy_to_virt_clear_cz(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));

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

    return RT_SUCCESS;
}

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

    assert(ctrlzone != NULL);

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

    AddressInfo outer_tunnel_addrinfo;
    if (outer_tunnel.GetAddressInfo(outer_tunnel_addrinfo) == RT_SUCCESS)
    {
    	assert(outer_tunnel_addrinfo.type==IPV4_ADDR);
		memcpy(ctrlzone->g_device_mac, outer_tunnel_addrinfo.src_mac_addr, sizeof(ctrlzone->g_device_mac));
        memcpy(ctrlzone->l_device_mac, outer_tunnel_addrinfo.dst_mac_addr, sizeof(ctrlzone->l_device_mac));
        ctrlzone->g_device_in_addr = outer_tunnel_addrinfo.addr.ipv4_addr.src_in_addr.s_addr;
        ctrlzone->l4_src_port = outer_tunnel_addrinfo.s_port;
        ctrlzone->g_device_vpn_id = outer_tunnel_addrinfo.vpn_id;
    }

    ctrlzone->__encap_type = outer_tunnel.ThisTunType();
    ctrlzone->__encap_len = tun_array.sz_total_len - inner_tunnel.ThisTunLength();
    ctrlzone->action = TUNNAT_CZ_ACTION_FORWARD;

    assert(ctrlzone->__encap_len != 0);

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


#if MR_TUNNAT_USE_PHONY_ETHER_HEADER   /* 构造伪以太网头部 */
    struct ether_hdr * p_ether_hdr = (struct ether_hdr *)
        marsio_buff_prepend(mbuf, sizeof(struct ether_hdr));

    assert(p_ether_hdr != NULL);

    ctrlzone->__p_ether_type = p_ether_hdr->ether_type;
    switch (tun_array.GetInnerTunContainer().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);
    }

#endif

    (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)
        {
            __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]);
        }
    }

    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)
{
    mr_tunnat_ctrlzone * ctrlzone = static_cast<mr_tunnat_ctrlzone *>(
        marsio_buff_ctrlzone(mbuf, g_ctrlzone_id));

    assert(ctrlzone != NULL);

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

    /* 对于没有封装的报文，如专用设备的保活报文，其隧道长度为0，不做任何处理 */
    if (ctrlzone->__encap_len == 0) return RT_ERR;

#if MR_TUNNAT_USE_PHONY_ETHER_HEADER
    
    /* 恢复伪以太网头部修改的数据，去掉以太网头部 */
    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 != NULL);

#endif

    marsio_buff_prepend(mbuf, ctrlzone->__encap_len);
    char * tun_start = marsio_buff_mtod(mbuf);
    tunnel_type tun_type = static_cast<tunnel_type>(ctrlzone->__encap_type);
    TunnelContainer::PacketForwardModify(tun_start, ctrlzone->__encap_len, tun_type);

    TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_FORWARD_TUNNEL_OUTPUT, 1);

    (void)__adj_check;
    return RT_SUCCESS;
}

static int __virt_to_phy_pkt_encap(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 != NULL);

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

#if MR_TUNNAT_USE_PHONY_ETHER_HEADER

	SessionKey ss_key;
	struct ether_hdr *ether_hdr=(struct ether_hdr *)marsio_buff_mtod(mbuf);
	switch(ntohs(ether_hdr->ether_type))
	{
	case ETHER_TYPE_IPv4:
		ss_key.type=KEY_TYPE_IPV4;
		break;
	case ETHER_TYPE_IPv6:
		ss_key.type=KEY_TYPE_IPV6;
		break;
	default:
		assert(0);
		break;
	}

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

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

#endif

	switch(ss_key.type)
	{
	case KEY_TYPE_IPV4:
	{
		struct ipv4_hdr *ipv4_hdr = (struct ipv4_hdr *)marsio_buff_mtod(mbuf);
		ss_key.ipv4_key.s_in_addr.s_addr=ipv4_hdr->src_addr;
		ss_key.ipv4_key.d_in_addr.s_addr=ipv4_hdr->dst_addr;
		break;
	}
	case KEY_TYPE_IPV6:
	{
		struct ipv6_hdr *ipv6_hdr = (struct ipv6_hdr *)marsio_buff_mtod(mbuf);
		memcpy(&ss_key.ipv6_key.s_in6_addr,ipv6_hdr->src_addr,sizeof(struct in6_addr));
		memcpy(&ss_key.ipv6_key.d_in6_addr,ipv6_hdr->dst_addr,sizeof(struct in6_addr));
		break;
	}
	default:
		assert(0);
		break;
	} 

    /* 双向查询 */
    SessionEntry * ss_entry = th_instance->ss_table->QueryDualDirect(ss_key);
    if (ss_entry != NULL) goto session_realy;
    
    /* 没有查询到，使用最近的一次隧道数据，随便找到一个专用设备回注 */
    ss_entry = th_instance->ss_table->QueryRecent(ss_key.type);
    if (ss_entry != NULL)
    {
        TUNNAT_THREAD_STAT_ADD(TUNNAT_STAT_ENCAP_USE_RECENT_SESSION, 1);
        goto session_realy;
    }

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

session_realy:

    /* 已经查找到隧道上下文，构建隧道报文 */
    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];

        const char * pkt_ptr = marsio_buff_prepend(mbuf, __tun_container.ThisTunLength());
        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 != NULL);

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

    /* 对于内层是以太报文的 */
    if (ctrlzone->g_device_inner_encap_type == 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 != NULL);
        OuterVXLAN.PacketConstruct(pkt_ptr, pkt_len);

        return RT_SUCCESS;
    }
    else if (ctrlzone->g_device_inner_encap_type == TUNNEL_TYPE_PPP)
    {
        assert(0);
    }
    else if (ctrlzone->g_device_inner_encap_type == 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;

    for (int i = 0; i < nr_mbufs; i++)
    {    
	    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, 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);

    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];

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

    return 0;
}