O-RAN E Maintenance Release contribution for ODULOW
[o-du/phy.git] / fhi_lib / lib / ethernet / ethdi.c
index bbde7a1..b6ba257 100644 (file)
-/******************************************************************************\r
-*\r
-*   Copyright (c) 2019 Intel.\r
-*\r
-*   Licensed under the Apache License, Version 2.0 (the "License");\r
-*   you may not use this file except in compliance with the License.\r
-*   You may obtain a copy of the License at\r
-*\r
-*       http://www.apache.org/licenses/LICENSE-2.0\r
-*\r
-*   Unless required by applicable law or agreed to in writing, software\r
-*   distributed under the License is distributed on an "AS IS" BASIS,\r
-*   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
-*   See the License for the specific language governing permissions and\r
-*   limitations under the License.\r
-*\r
-*******************************************************************************/\r
-\r
-/**\r
- * @brief This file has all definitions for the Ethernet Data Interface Layer\r
- * @file ethdi.c\r
- * @ingroup group_lte_source_auxlib\r
- * @author Intel Corporation\r
- **/\r
-\r
-#define _GNU_SOURCE\r
-#include <stdio.h>\r
-#include <string.h>\r
-#include <stdint.h>\r
-#include <errno.h>\r
-#include <sys/queue.h>\r
-#include <err.h>\r
-#include <assert.h>\r
-#include <linux/limits.h>\r
-#include <sys/types.h>\r
-#include <stdlib.h>\r
-#include <sys/time.h>\r
-#include <time.h>\r
-#include <unistd.h>\r
-\r
-#include <rte_config.h>\r
-#include <rte_common.h>\r
-#include <rte_log.h>\r
-#include <rte_memory.h>\r
-#include <rte_memcpy.h>\r
-#include <rte_memzone.h>\r
-#include <rte_eal.h>\r
-#include <rte_per_lcore.h>\r
-#include <rte_launch.h>\r
-#include <rte_atomic.h>\r
-#include <rte_cycles.h>\r
-#include <rte_prefetch.h>\r
-#include <rte_lcore.h>\r
-#include <rte_per_lcore.h>\r
-#include <rte_branch_prediction.h>\r
-#include <rte_interrupts.h>\r
-#include <rte_pci.h>\r
-#include <rte_debug.h>\r
-#include <rte_ethdev.h>\r
-#include <rte_ring.h>\r
-#include <rte_mbuf.h>\r
-#include <rte_timer.h>\r
-\r
-#include "ethernet.h"\r
-#include "ethdi.h"\r
-#include "xran_fh_o_du.h"\r
-#include "xran_mlog_lnx.h"\r
-#include "xran_printf.h"\r
-\r
-#include "../src/xran_lib_mlog_tasks_id.h"\r
-\r
-#define BURST_RX_IO_SIZE 48\r
-\r
-struct xran_ethdi_ctx g_ethdi_ctx = { 0 };\r
-enum xran_if_state xran_if_current_state = XRAN_STOPPED;\r
-\r
-struct rte_mbuf *xran_ethdi_mbuf_alloc(void)\r
-{\r
-    return rte_pktmbuf_alloc(_eth_mbuf_pool);\r
-}\r
-\r
-int xran_ethdi_mbuf_send(struct rte_mbuf *mb, uint16_t ethertype)\r
-{\r
-    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();\r
-    int res = 0;\r
-\r
-    mb->port = ctx->io_cfg.port[ETHDI_UP_VF];\r
-    xran_add_eth_hdr_vlan(&ctx->entities[ID_RU], ethertype, mb, ctx->up_vtag);\r
-\r
-    res = xran_enqueue_mbuf(mb, ctx->tx_ring[ETHDI_UP_VF]);\r
-    return res;\r
-}\r
-\r
-int xran_ethdi_mbuf_send_cp(struct rte_mbuf *mb, uint16_t ethertype)\r
-{\r
-    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();\r
-    int res = 0;\r
-\r
-    mb->port = ctx->io_cfg.port[ETHDI_CP_VF];\r
-    xran_add_eth_hdr_vlan(&ctx->entities[ID_RU], ethertype, mb, ctx->cp_vtag);\r
-\r
-    res = xran_enqueue_mbuf(mb, ctx->tx_ring[ETHDI_CP_VF]);\r
-    return res;\r
-}\r
-#if 0\r
-void xran_ethdi_stop_tx()\r
-{\r
-    struct xran_ethdi_ctx *const ctx = xran_ethdi_get_ctx();\r
-    rte_timer_stop_sync(&ctx->timer_tx);\r
-}\r
-#endif\r
-\r
-struct {\r
-    uint16_t ethertype;\r
-    ethertype_handler fn;\r
-} xran_ethertype_handlers[] = {\r
-    { ETHER_TYPE_ETHDI, NULL },\r
-    { ETHER_TYPE_ECPRI, NULL },\r
-    { ETHER_TYPE_START_TX, NULL }\r
-};\r
-\r
-\r
-\r
-int xran_register_ethertype_handler(uint16_t ethertype, ethertype_handler callback)\r
-{\r
-    int i;\r
-\r
-    for (i = 0; i < RTE_DIM(xran_ethertype_handlers); ++i)\r
-        if (xran_ethertype_handlers[i].ethertype == ethertype) {\r
-            xran_ethertype_handlers[i].fn = callback;\r
-\r
-            return 1;\r
-        }\r
-\r
-    elog("support for ethertype %u not found", ethertype);\r
-\r
-    return 0;\r
-}\r
-\r
-int xran_handle_ether(uint16_t ethertype, struct rte_mbuf *pkt, uint64_t rx_time)\r
-{\r
-    int i;\r
-\r
-    for (i = 0; i < RTE_DIM(xran_ethertype_handlers); ++i)\r
-        if (xran_ethertype_handlers[i].ethertype == ethertype)\r
-            if (xran_ethertype_handlers[i].fn)\r
-                return xran_ethertype_handlers[i].fn(pkt, rx_time);\r
-\r
-    wlog("Packet with unrecognized ethertype '%.4X' dropped", ethertype);\r
-\r
-    return 0;\r
-};\r
-\r
-\r
-/* Process vlan tag. Cut the ethernet header. Call the etherype handlers. */\r
-int xran_ethdi_filter_packet(struct rte_mbuf *pkt, uint64_t rx_time)\r
-{\r
-    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();\r
-\r
-#ifdef VLAN_SUPPORT\r
-    if (rte_vlan_strip(pkt) == 0) {\r
-        if (pkt->vlan_tci == ctx->cp_vtag) {\r
-            dlog("VLAN tci matches %d", pkt->vlan_tci);\r
-        } else {\r
-            wlog("packet with wrong VLAN tag %d, dropping",\r
-                    pkt->vlan_tci);\r
-            return 0;\r
-        }\r
-    } else\r
-        dlog("Packet not vlan tagged");\r
-#endif\r
-\r
-    const struct ether_hdr *eth_hdr = rte_pktmbuf_mtod(pkt, void *);\r
-\r
-#if defined(DPDKIO_DEBUG) && DPDKIO_DEBUG > 1\r
-    nlog("*** processing RX'ed packet of size %d ***",\r
-            rte_pktmbuf_data_len(pkt));\r
-    /* TODO: just dump ethernet header in readable format? */\r
-#endif\r
-\r
-#if defined(DPDKIO_DEBUG) && DPDKIO_DEBUG > 1\r
-    {\r
-        char dst[ETHER_ADDR_FMT_SIZE] = "(empty)";\r
-        char src[ETHER_ADDR_FMT_SIZE] = "(empty)";\r
-\r
-        ether_format_addr(dst, sizeof(dst), &eth_hdr->d_addr);\r
-        ether_format_addr(src, sizeof(src), &eth_hdr->s_addr);\r
-        nlog("src: %s dst: %s ethertype: %.4X", dst, src,\r
-                rte_be_to_cpu_16(eth_hdr->ether_type));\r
-    }\r
-#endif\r
-\r
-    /* Cut out the ethernet header. It's not needed anymore. */\r
-    if (rte_pktmbuf_adj(pkt, sizeof(*eth_hdr)) == NULL) {\r
-        wlog("Packet too short, dropping");\r
-        return 0;\r
-    }\r
-\r
-\r
-    return xran_handle_ether(rte_be_to_cpu_16(eth_hdr->ether_type), pkt, rx_time);\r
-}\r
-\r
-#if 0\r
-//-------------------------------------------------------------------------------------------\r
-/** @ingroup xran\r
- *\r
- *  @param[in]  port - DPDK ETH port id\r
- *\r
- *  @return  void\r
- *\r
- *  @description\r
- *  Prints statistics of usage of DPDK port\r
- *\r
-**/\r
-//-------------------------------------------------------------------------------------------\r
-void xran_ethdi_ports_stats(void)\r
-{\r
-    struct rte_eth_stats stats;\r
-    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();\r
-    int32_t i = 0;\r
-\r
-    for(i = 0; i < ETHDI_VF_MAX; i++){\r
-        /* Get stats (extended stats includes common stats) */\r
-        rte_eth_stats_get(ctx->io_cfg.port[i], &stats);\r
-        printf("DPDK stats:\n");\r
-        printf("** Port %hhu **\n", ctx->io_cfg.port[i]);\r
-        printf("ierrors:\t%lu\n",   stats.ierrors);\r
-        printf("oerrors:\t%lu\n",   stats.oerrors);\r
-        printf("ipackets:\t%lu\n",  stats.ipackets);\r
-        printf("opackets:\t%lu\n",  stats.opackets);\r
-        printf("imissed:\t%lu\n",   stats.imissed);\r
-        printf("rx_nombuf:\t%lu\n", stats.rx_nombuf);\r
-    }\r
-    return ;\r
-}\r
-#endif\r
-/* Check the link status of all ports in up to 9s, and print them finally */\r
-static void check_port_link_status(uint8_t portid)\r
-{\r
-#define CHECK_INTERVAL 100 /* 100ms */\r
-#define MAX_CHECK_TIME 90 /* 9s (90 * 100ms) in total */\r
-    uint8_t count, all_ports_up, print_flag = 0;\r
-    struct rte_eth_link link;\r
-\r
-    printf("\nChecking link status");\r
-    fflush(stdout);\r
-    for (count = 0; count <= MAX_CHECK_TIME; count++) {\r
-        all_ports_up = 1;\r
-        memset(&link, 0, sizeof(link));\r
-        rte_eth_link_get_nowait(portid, &link);\r
-\r
-        /* print link status if flag set */\r
-        if (print_flag == 1) {\r
-            if (link.link_status)\r
-                printf("Port %d Link Up - speed %u "\r
-                        "Mbps - %s\n", (uint8_t)portid,\r
-                        (unsigned)link.link_speed,\r
-                        (link.link_duplex == ETH_LINK_FULL_DUPLEX) ?\r
-                        ("full-duplex") : ("half-duplex\n"));\r
-            else\r
-                printf("Port %d Link Down\n",\r
-                        (uint8_t)portid);\r
-        }\r
-        /* clear all_ports_up flag if any link down */\r
-        if (link.link_status == ETH_LINK_DOWN) {\r
-            all_ports_up = 0;\r
-            break;\r
-        }\r
-        /* after finally printing all link status, get out */\r
-        if (print_flag == 1)\r
-            break;\r
-\r
-        if (all_ports_up == 0) {\r
-            printf(".");\r
-            fflush(stdout);\r
-            rte_delay_ms(CHECK_INTERVAL);\r
-        }\r
-\r
-        /* set the print_flag if all ports up or timeout */\r
-        if (all_ports_up == 1 || count == (MAX_CHECK_TIME - 1)) {\r
-            print_flag = 1;\r
-            printf(" ... done\n");\r
-        }\r
-    }\r
-}\r
-\r
-\r
-int xran_ethdi_init_dpdk_io(char *name, const struct xran_io_loop_cfg *io_cfg,\r
-    int *lcore_id, struct ether_addr *p_lls_cu_addr, struct ether_addr *p_ru_addr,\r
-    uint16_t cp_vlan, uint16_t up_vlan)\r
-{\r
-    uint16_t port[2] = {0xffff, 0xffff};\r
-    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();\r
-    int i;\r
-    char core_mask[64];\r
-    long c_mask=0;\r
-    char bbdev_wdev[32] = "";\r
-    char bbdev_vdev[32] = "";\r
-\r
-    char *argv[] = { name, /*"-c 0xFFFFF00000FFFFF"*/core_mask, "-n2", "--socket-mem=8192", "--proc-type=auto",\r
-        "--file-prefix", name, "-w", "0000:00:00.0", bbdev_wdev, bbdev_vdev};\r
-\r
-    if (io_cfg == NULL)\r
-        return 0;\r
-    if(io_cfg->bbdev_mode != XRAN_BBDEV_NOT_USED){\r
-        printf("BBDEV_FEC_ACCL_NR5G\n");\r
-        if (io_cfg->bbdev_mode == XRAN_BBDEV_MODE_HW_ON){\r
-            // hw-accelerated bbdev\r
-            printf("hw-accelerated bbdev %s\n", io_cfg->bbdev_dev[0]);\r
-            sprintf(bbdev_wdev, "-w %s", io_cfg->bbdev_dev[0]);\r
-        } else if (io_cfg->bbdev_mode == XRAN_BBDEV_MODE_HW_OFF){\r
-            // hw-accelerated bbdev disable\r
-            if(io_cfg->bbdev_dev[0]){\r
-                printf("hw-accelerated bbdev disable %s\n", io_cfg->bbdev_dev[0]);\r
-                sprintf(bbdev_wdev, "-b %s", io_cfg->bbdev_dev[0]);\r
-            }\r
-            sprintf(bbdev_wdev, "%s", "--vdev=baseband_turbo_sw");\r
-        } else {\r
-            rte_panic("Cannot init DPDK incorrect [bbdev_mode %d]\n", io_cfg->bbdev_mode);\r
-        }\r
-    }\r
-\r
-    c_mask = (long)(1L << io_cfg->core) |\r
-            (long)(1L << io_cfg->system_core) |\r
-            (long)(1L << io_cfg->pkt_proc_core) |\r
-            (long)(1L << io_cfg->pkt_aux_core) |\r
-            (long)(1L << io_cfg->timing_core);\r
-\r
-    printf("c_mask 0x%lx core %d system_core %d pkt_proc_core %d pkt_aux_core %d timing_core %d\n",\r
-        c_mask, io_cfg->core, io_cfg->system_core, io_cfg->pkt_proc_core, io_cfg->pkt_aux_core, io_cfg->timing_core);\r
-\r
-    snprintf(core_mask, sizeof(core_mask), "-c 0x%lx", c_mask);\r
-\r
-    ctx->io_cfg = *io_cfg;\r
-    ctx->ping_state           = PING_IDLE;\r
-    ctx->known_peers          = 1;\r
-    ctx->busy_poll_till = rte_rdtsc();\r
-    ctx->cp_vtag = cp_vlan;\r
-    ctx->up_vtag = up_vlan;\r
-\r
-    for (i = 0; i <= ID_BROADCAST; i++)     /* Initialize all as broadcast */\r
-        memset(&ctx->entities[i], 0xFF, sizeof(ctx->entities[0]));\r
-\r
-    printf("%s: Calling rte_eal_init:", __FUNCTION__);\r
-    for (i = 0; i < RTE_DIM(argv); i++)\r
-    {\r
-        printf("%s ", argv[i]);\r
-    }\r
-    printf("\n");\r
-\r
-\r
-    /* This will return on system_core, which is not necessarily the\r
-     * one we're on right now. */\r
-    if (rte_eal_init(RTE_DIM(argv), argv) < 0)\r
-        rte_panic("Cannot init EAL: %s\n", rte_strerror(rte_errno));\r
-\r
-    xran_init_mbuf_pool();\r
-\r
-#ifdef RTE_LIBRTE_PDUMP\r
-    /* initialize packet capture framework */\r
-    rte_pdump_init(NULL);\r
-#endif\r
-\r
-    /* Timers. */\r
-    rte_timer_subsystem_init();\r
-    rte_timer_init(&ctx->timer_ping);\r
-    rte_timer_init(&ctx->timer_sync);\r
-    rte_timer_init(&ctx->timer_tx);\r
-\r
-    *lcore_id = rte_get_next_lcore(rte_lcore_id(), 0, 0);\r
-\r
-    PANIC_ON(*lcore_id == RTE_MAX_LCORE, "out of lcores for io_loop()");\r
-\r
-    if (rte_eal_process_type() == RTE_PROC_PRIMARY) {\r
-        for (i = 0; i < ETHDI_VF_MAX; i ++){\r
-            if(io_cfg->dpdk_dev[i]){\r
-                if (rte_eth_dev_attach(io_cfg->dpdk_dev[i], &port[i]) != 0 ||\r
-                    rte_eth_dev_count_avail() == 0)\r
-                    errx(1, "Network port doesn't exist.");\r
-                xran_init_port(port[i], p_lls_cu_addr);\r
-            } else {\r
-                printf("no DPDK port provided\n");\r
-            }\r
-            if(i==0){\r
-                ctx->tx_ring[i] = rte_ring_create("tx_ring_up", NUM_MBUFS,\r
-                    rte_lcore_to_socket_id(*lcore_id), RING_F_SC_DEQ);\r
-                ctx->rx_ring[i] = rte_ring_create("rx_ring_up", NUM_MBUFS,\r
-                    rte_lcore_to_socket_id(*lcore_id), RING_F_SC_DEQ);\r
-                ctx->pkt_dump_ring[i] = rte_ring_create("pkt_dump_ring_up", NUM_MBUFS,\r
-                    rte_lcore_to_socket_id(*lcore_id), RING_F_SC_DEQ);\r
-            }else {\r
-                ctx->tx_ring[i] = rte_ring_create("tx_ring_cp", NUM_MBUFS,\r
-                    rte_lcore_to_socket_id(*lcore_id), RING_F_SC_DEQ);\r
-                ctx->rx_ring[i] = rte_ring_create("rx_ring_cp", NUM_MBUFS,\r
-                    rte_lcore_to_socket_id(*lcore_id), RING_F_SC_DEQ);\r
-                ctx->pkt_dump_ring[i] = rte_ring_create("pkt_dump_ring_cp", NUM_MBUFS,\r
-                    rte_lcore_to_socket_id(*lcore_id), RING_F_SC_DEQ);\r
-            }\r
-            if(io_cfg->dpdk_dev[i])\r
-                check_port_link_status(port[i]);\r
-        }\r
-    } else {\r
-        rte_panic("ethdi_dpdk_io_loop() failed to start  with RTE_PROC_SECONDARY\n");\r
-    }\r
-    PANIC_ON(ctx->tx_ring == NULL, "failed to allocate tx ring");\r
-    PANIC_ON(ctx->rx_ring == NULL, "failed to allocate rx ring");\r
-    PANIC_ON(ctx->pkt_dump_ring == NULL, "failed to allocate pkt dumping ring");\r
-    for (i = 0; i < ETHDI_VF_MAX; i++){\r
-        ctx->io_cfg.port[i] = port[i];\r
-        print_dbg("port_id 0x%04x\n", ctx->io_cfg.port[i]);\r
-    }\r
-\r
-    if(io_cfg->dpdk_dev[ETHDI_UP_VF]){\r
-        rte_eth_macaddr_get(port[ETHDI_UP_VF], &ctx->entities[io_cfg->id]);\r
-        ether_addr_copy(p_ru_addr,  &ctx->entities[ID_RU]);\r
-    }\r
-\r
-    return 1;\r
-}\r
-\r
-static inline uint16_t xran_tx_from_ring(int port, struct rte_ring *r)\r
-{\r
-    struct rte_mbuf *mbufs[BURST_SIZE];\r
-    uint16_t dequeued, sent = 0;\r
-    uint32_t remaining;\r
-    int i;\r
-    long t1 = MLogTick();\r
-\r
-    dequeued = rte_ring_dequeue_burst(r, (void **)mbufs, BURST_SIZE,\r
-            &remaining);\r
-    if (!dequeued)\r
-        return 0;   /* Nothing to send. */\r
-\r
-    while (1) {     /* When tx queue is full it is trying again till succeed */\r
-        t1 = MLogTick();\r
-        sent += rte_eth_tx_burst(port, 0, &mbufs[sent], dequeued - sent);\r
-\r
-        MLogTask(PID_RADIO_ETH_TX_BURST, t1, MLogTick());\r
-\r
-        if (sent == dequeued)\r
-            return remaining;\r
-    }\r
-}\r
-\r
-int32_t process_dpdk_io(void)\r
-{\r
-    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();\r
-    const struct xran_io_loop_cfg *const cfg = &(xran_ethdi_get_ctx()->io_cfg);\r
-    const int port[ETHDI_VF_MAX] = {cfg->port[ETHDI_UP_VF], cfg->port[ETHDI_CP_VF]};\r
-    int port_id = 0;\r
-\r
-    for (port_id = 0; port_id < ETHDI_VF_MAX; port_id++){\r
-        struct rte_mbuf *mbufs[BURST_RX_IO_SIZE];\r
-        if(port[port_id] == 0xFF)\r
-            return 0;\r
-        /* RX */\r
-        const uint16_t rxed = rte_eth_rx_burst(port[port_id], 0, mbufs, BURST_RX_IO_SIZE);\r
-        if (rxed != 0){\r
-            unsigned enq_n = 0;\r
-            long t1 = MLogTick();\r
-            enq_n =  rte_ring_enqueue_burst(ctx->rx_ring[port_id], (void*)mbufs, rxed, NULL);\r
-            if(rxed - enq_n)\r
-                rte_panic("error enq\n");\r
-            MLogTask(PID_RADIO_RX_VALIDATE, t1, MLogTick());\r
-        }\r
-\r
-        /* TX */\r
-        const uint16_t sent = xran_tx_from_ring(port[port_id], ctx->tx_ring[port_id]);\r
-\r
-        if (XRAN_STOPPED == xran_if_current_state)\r
-            return -1;\r
-    }\r
-\r
-    if (XRAN_STOPPED == xran_if_current_state)\r
-            return -1;\r
-\r
-    return 0;\r
-}\r
-\r
-#if 0\r
-static inline void xran_process_rx_burst(struct rte_mbuf *mbufs[], uint16_t n_mbufs,\r
-        uint64_t rx_time)\r
-{\r
-        int i;\r
-\r
-        if (!n_mbufs)\r
-            return;\r
-\r
-        for (i = 0; i < n_mbufs; ++i)\r
-        {\r
-            if (xran_ethdi_filter_packet(mbufs[i], rx_time) == MBUF_FREE)\r
-                rte_pktmbuf_free(mbufs[i]);\r
-        }\r
-\r
-#ifdef DPDKIO_LATENCY_DEBUG\r
-       struct timeval tv_now, tv_diff;\r
-\r
-       gettimeofday(&tv_now, NULL);\r
-       if (n_mbufs > 1)\r
-           nlog("Warning - received %d mbufs in a row", n_mbufs);\r
-\r
-       timersub(&tv_now, &rx_time, &tv_diff);\r
-       nlog("rx processing took %d usec", tv_diff.tv_usec);\r
-#endif\r
-}\r
-\r
-/*\r
- * This is the main DPDK-IO loop.\r
- * This will sleep if there's no packets incoming and there's\r
- * no work enqueued, sleep lenth is defined in IDLE_SLEEP_MICROSECS\r
- */\r
-int xran_ethdi_dpdk_io_loop(void *io_loop_cfg)\r
-{\r
-    struct sched_param sched_param;\r
-    int res = 0;\r
-    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();\r
-    const struct xran_io_loop_cfg *const cfg = &(xran_ethdi_get_ctx()->io_cfg);\r
-    const int port[ETHDI_VF_MAX] = {cfg->port[ETHDI_UP_VF], cfg->port[ETHDI_CP_VF]};\r
-\r
-    printf("%s [PORT: %d %d] [CPU %2d] [PID: %6d]\n", __FUNCTION__, port[ETHDI_UP_VF], port[ETHDI_CP_VF] , rte_lcore_id(), getpid());\r
-\r
-    printf("%s [CPU %2d] [PID: %6d]\n", __FUNCTION__,  rte_lcore_id(), getpid());\r
-    sched_param.sched_priority = XRAN_THREAD_DEFAULT_PRIO;\r
-    if ((res = pthread_setschedparam(pthread_self(), SCHED_FIFO, &sched_param))) {\r
-        printf("priority is not changed: coreId = %d, result1 = %d\n",rte_lcore_id(), res);\r
-    }\r
-\r
-    for (;;){\r
-        if(process_dpdk_io()!=0)\r
-            break;\r
-    }\r
-\r
-    fflush(stderr);\r
-    fflush(stdout);\r
-    puts("IO loop finished");\r
-\r
-    return 0;\r
-}\r
-#endif\r
-\r
+/******************************************************************************
+*
+*   Copyright (c) 2020 Intel.
+*
+*   Licensed under the Apache License, Version 2.0 (the "License");
+*   you may not use this file except in compliance with the License.
+*   You may obtain a copy of the License at
+*
+*       http://www.apache.org/licenses/LICENSE-2.0
+*
+*   Unless required by applicable law or agreed to in writing, software
+*   distributed under the License is distributed on an "AS IS" BASIS,
+*   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+*   See the License for the specific language governing permissions and
+*   limitations under the License.
+*
+*******************************************************************************/
+
+/**
+ * @brief This file has all definitions for the Ethernet Data Interface Layer
+ * @file ethdi.c
+ * @ingroup group_lte_source_auxlib
+ * @author Intel Corporation
+ **/
+
+#define _GNU_SOURCE
+#include <stdio.h>
+#include <string.h>
+#include <stdint.h>
+#include <errno.h>
+#include <sys/queue.h>
+#include <err.h>
+#include <assert.h>
+#include <linux/limits.h>
+#include <sys/types.h>
+#include <stdlib.h>
+#include <sys/time.h>
+#include <time.h>
+#include <unistd.h>
+#include <immintrin.h>
+#include <rte_config.h>
+#include <rte_common.h>
+#include <rte_log.h>
+#include <rte_memory.h>
+#include <rte_malloc.h>
+#include <rte_memzone.h>
+#include <rte_eal.h>
+#include <rte_per_lcore.h>
+#include <rte_launch.h>
+#include <rte_atomic.h>
+#include <rte_cycles.h>
+#include <rte_prefetch.h>
+#include <rte_lcore.h>
+#include <rte_per_lcore.h>
+#include <rte_branch_prediction.h>
+#include <rte_interrupts.h>
+#include <rte_pci.h>
+#include <rte_debug.h>
+#include <rte_ethdev.h>
+#include <rte_ring.h>
+#include <rte_mbuf.h>
+#include <rte_timer.h>
+
+#include "ethernet.h"
+#include "ethdi.h"
+#include "xran_fh_o_du.h"
+#include "xran_mlog_lnx.h"
+#include "xran_printf.h"
+#include "xran_common.h"
+
+#include "xran_lib_mlog_tasks_id.h"
+
+#define BURST_RX_IO_SIZE 48
+
+//#define ORAN_OWD_DEBUG_TX_LOOP
+
+struct xran_ethdi_ctx g_ethdi_ctx = { 0 };
+enum xran_if_state xran_if_current_state = XRAN_STOPPED;
+
+struct rte_mbuf *xran_ethdi_mbuf_alloc(void)
+{
+    return rte_pktmbuf_alloc(_eth_mbuf_pool);
+}
+
+struct rte_mbuf *xran_ethdi_mbuf_indir_alloc(void)
+{
+    return rte_pktmbuf_alloc(socket_indirect_pool);
+}
+
+int32_t xran_ethdi_mbuf_send(struct rte_mbuf *mb, uint16_t ethertype, uint16_t vf_id)
+{
+    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();
+    int res = 0;
+
+    mb->port = ctx->io_cfg.port[vf_id];
+    xran_add_eth_hdr_vlan(&ctx->entities[vf_id][ID_O_RU], ethertype, mb);
+
+    res = xran_enqueue_mbuf(mb, ctx->tx_ring[vf_id]);
+    return res;
+}
+
+int32_t xran_ethdi_mbuf_send_cp(struct rte_mbuf *mb, uint16_t ethertype, uint16_t vf_id)
+{
+    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();
+    int res = 0;
+
+    mb->port = ctx->io_cfg.port[vf_id];
+    xran_add_eth_hdr_vlan(&ctx->entities[vf_id][ID_O_RU], ethertype, mb);
+
+    res = xran_enqueue_mbuf(mb, ctx->tx_ring[vf_id]);
+    return res;
+}
+
+struct {
+    uint16_t ethertype;
+    ethertype_handler fn;
+} xran_ethertype_handlers[] = {
+    { ETHER_TYPE_ECPRI, NULL },
+};
+
+int32_t xran_register_ethertype_handler(uint16_t ethertype, ethertype_handler callback)
+{
+    int i;
+
+    for (i = 0; i < RTE_DIM(xran_ethertype_handlers); ++i)
+        if (xran_ethertype_handlers[i].ethertype == ethertype) {
+            xran_ethertype_handlers[i].fn = callback;
+
+            return 1;
+        }
+
+    print_err("support for ethertype %u not found", ethertype);
+
+    return 0;
+}
+
+int xran_handle_ether(uint16_t ethertype, struct rte_mbuf* pkt_q[], uint16_t xport_id, struct xran_eaxc_info *p_cid,  uint16_t num)
+{
+    int i;
+
+    for (i = 0; i < RTE_DIM(xran_ethertype_handlers); ++i)
+        if (xran_ethertype_handlers[i].ethertype == ethertype)
+            if (xran_ethertype_handlers[i].fn){
+//                rte_prefetch0(rte_pktmbuf_mtod(pkt, void *));
+                return xran_ethertype_handlers[i].fn(pkt_q, xport_id, p_cid, num);
+            }
+
+    print_err("Packet with unrecognized ethertype '%.4X' dropped", ethertype);
+
+    return MBUF_FREE;
+};
+
+
+/* Process vlan tag. Cut the ethernet header. Call the etherype handlers. */
+int xran_ethdi_filter_packet(struct rte_mbuf *pkt_q[], uint16_t vf_id, uint16_t q_id, uint16_t num)
+{
+    int ret;
+    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();
+    struct rte_ether_hdr* eth_hdr;
+    uint16_t port_id = ctx->vf2xran_port[vf_id];
+    struct xran_eaxc_info *p_cid = &ctx->vf_and_q2cid[vf_id][q_id];
+
+    ret = xran_handle_ether(ETHER_TYPE_ECPRI, pkt_q, port_id, p_cid, num);
+
+    return MBUF_FREE;
+}
+
+/* Check the link status of all ports in up to 9s, and print them finally */
+static void check_port_link_status(uint8_t portid)
+{
+#define CHECK_INTERVAL 100 /* 100ms */
+#define MAX_CHECK_TIME 90 /* 9s (90 * 100ms) in total */
+    uint8_t count, all_ports_up, print_flag = 0;
+    struct rte_eth_link link;
+
+    printf("\nChecking link status portid [%d]  ",  portid);
+    fflush(stdout);
+    for (count = 0; count <= MAX_CHECK_TIME; count++) {
+        all_ports_up = 1;
+        memset(&link, 0, sizeof(link));
+        rte_eth_link_get_nowait(portid, &link);
+
+        /* print link status if flag set */
+        if (print_flag == 1) {
+            if (link.link_status)
+                printf("Port %d Link Up - speed %u "
+                        "Mbps - %s\n", (uint8_t)portid,
+                        (unsigned)link.link_speed,
+                        (link.link_duplex == ETH_LINK_FULL_DUPLEX) ?
+                        ("full-duplex") : ("half-duplex\n"));
+            else
+                printf("Port %d Link Down\n",
+                        (uint8_t)portid);
+        }
+        /* clear all_ports_up flag if any link down */
+        if (link.link_status == ETH_LINK_DOWN) {
+            all_ports_up = 0;
+            break;
+        }
+        /* after finally printing all link status, get out */
+        if (print_flag == 1)
+            break;
+
+        if (all_ports_up == 0) {
+            printf(".");
+            fflush(stdout);
+            rte_delay_ms(CHECK_INTERVAL);
+        }
+
+        /* set the print_flag if all ports up or timeout */
+        if (all_ports_up == 1 || count == (MAX_CHECK_TIME - 1)) {
+            print_flag = 1;
+            printf(" ... done\n");
+        }
+    }
+}
+
+/**
+ * create a flow rule that sends packets with matching pc_id
+ * to selected queue.
+ *
+ * @param port_id
+ *   The selected port.
+ * @param rx_q
+ *   The selected target queue.
+ * @param pc_id_be
+ *   The value to apply to the pc_id.
+ * @param[out] error
+ *   Perform verbose error reporting if not NULL.
+ *
+ * @return
+ *   A flow if the rule could be created else return NULL.
+ */
+struct rte_flow *
+generate_ecpri_flow(uint16_t port_id, uint16_t rx_q, uint16_t pc_id_be, struct rte_flow_error *error)
+{
+    struct rte_flow *flow = NULL;
+#if (RTE_VER_YEAR >= 21)
+#define MAX_PATTERN_NUM                3
+#define MAX_ACTION_NUM         2
+    struct rte_flow_attr attr;
+    struct rte_flow_item pattern[MAX_PATTERN_NUM];
+    struct rte_flow_action action[MAX_ACTION_NUM];
+
+    struct rte_flow_action_queue queue = { .index = rx_q };
+    struct rte_flow_item_ecpri ecpri_spec;
+    struct rte_flow_item_ecpri ecpri_mask;
+
+    int res;
+    print_dbg("%s\n", __FUNCTION__);
+    memset(pattern, 0, sizeof(pattern));
+    memset(action, 0, sizeof(action));
+
+    /*
+     * set the rule attribute.
+     * in this case only ingress packets will be checked.
+     */
+    memset(&attr, 0, sizeof(struct rte_flow_attr));
+    attr.ingress = 1;
+
+    /*
+     * create the action sequence.
+     * one action only,  move packet to queue
+     */
+    action[0].type = RTE_FLOW_ACTION_TYPE_QUEUE;
+    action[0].conf = &queue;
+    action[1].type = RTE_FLOW_ACTION_TYPE_END;
+
+    /*
+     * set the first level of the pattern (ETH).
+     * since in this example we just want to get the
+     * eCPRI we set this level to allow all.
+     */
+    pattern[0].type = RTE_FLOW_ITEM_TYPE_ETH;
+
+    memset(&ecpri_spec, 0, sizeof(struct rte_flow_item_ecpri));
+    memset(&ecpri_mask, 0, sizeof(struct rte_flow_item_ecpri));
+
+    ecpri_spec.hdr.common.type = RTE_ECPRI_MSG_TYPE_IQ_DATA;
+    ecpri_spec.hdr.type0.pc_id = pc_id_be;
+
+    ecpri_mask.hdr.common.type  = 0xff;
+    ecpri_mask.hdr.type0.pc_id  = 0xffff;
+
+    ecpri_spec.hdr.common.u32   = rte_cpu_to_be_32(ecpri_spec.hdr.common.u32);
+
+    pattern[1].type = RTE_FLOW_ITEM_TYPE_ECPRI;
+    pattern[1].spec = &ecpri_spec;
+    pattern[1].mask = &ecpri_mask;
+
+    struct rte_flow_item_ecpri *pecpri_spec = (struct rte_flow_item_ecpri *)pattern[1].spec;
+    struct rte_flow_item_ecpri *pecpri_mask = (struct rte_flow_item_ecpri *)pattern[1].mask;
+    print_dbg("RTE_FLOW_ITEM_TYPE_ECPRI\n");
+    print_dbg("spec type %x pc_id %x\n", pecpri_spec->hdr.common.type, pecpri_spec->hdr.type0.pc_id);
+    print_dbg("mask type %x pc_id %x\n", pecpri_mask->hdr.common.type, pecpri_mask->hdr.type0.pc_id);
+
+    /* the final level must be always type end */
+    pattern[2].type = RTE_FLOW_ITEM_TYPE_END;
+
+    res = rte_flow_validate(port_id, &attr, pattern, action, error);
+    if (!res)
+        flow = rte_flow_create(port_id, &attr, pattern, action, error);
+    else {
+        rte_panic("Flow can't be created %d message: %s\n",
+                    error->type,
+                    error->message ? error->message : "(no stated reason)");
+    }
+#endif
+    return flow;
+}
+
+
+int32_t
+xran_ethdi_init_dpdk_io(char *name, const struct xran_io_cfg *io_cfg,
+    int *lcore_id, struct rte_ether_addr *p_o_du_addr,
+    struct rte_ether_addr *p_ru_addr,  uint32_t mtu)
+{
+    uint16_t port[XRAN_VF_MAX];
+    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();
+    int i,ivf;
+    char core_mask[64];
+    uint64_t c_mask         = 0L;
+    uint64_t c_mask_64_127  = 0L;
+    uint64_t nWorkerCore = 1;
+    uint32_t coreNum = sysconf(_SC_NPROCESSORS_CONF);
+    char bbdev_wdev[32]   = "";
+    char bbdev_vdev[32]   = "";
+    char iova_mode[32]    = "--iova-mode=pa";
+    char socket_mem[32]   = "--socket-mem=8192";
+    char socket_limit[32] = "--socket-limit=8192";
+    char ring_name[32]    = "";
+    int32_t xran_port = -1;
+    portid_t port_id;
+    queueid_t qi = 0;
+    uint16_t count;
+
+    char *argv[] = { name, core_mask, "-n2", iova_mode, socket_mem, socket_limit, "--proc-type=auto",
+        "--file-prefix", name, "-a0000:00:00.0", bbdev_wdev, bbdev_vdev};
+
+
+    if (io_cfg == NULL)
+        return 0;
+    if(io_cfg->bbdev_mode != XRAN_BBDEV_NOT_USED){
+        printf("BBDEV_FEC_ACCL_NR5G\n");
+        if (io_cfg->bbdev_mode == XRAN_BBDEV_MODE_HW_ON){
+            // hw-accelerated bbdev
+            printf("hw-accelerated bbdev %s\n", io_cfg->bbdev_dev[0]);
+
+            snprintf(bbdev_wdev, RTE_DIM(bbdev_wdev), "-a%s", io_cfg->bbdev_dev[0]);
+
+        } else if (io_cfg->bbdev_mode == XRAN_BBDEV_MODE_HW_OFF){
+
+            snprintf(bbdev_wdev, RTE_DIM(bbdev_wdev), "%s", "--vdev=baseband_turbo_sw");
+        } else if (io_cfg->bbdev_mode == XRAN_BBDEV_MODE_HW_SW){
+            printf("software and hw-accelerated bbdev %s\n", io_cfg->bbdev_dev[0]);
+
+            snprintf(bbdev_wdev, RTE_DIM(bbdev_wdev), "-a%s", io_cfg->bbdev_dev[0]);
+
+            snprintf(bbdev_vdev, RTE_DIM(bbdev_vdev), "%s", "--vdev=baseband_turbo_sw");
+        } else {
+            rte_panic("Cannot init DPDK incorrect [bbdev_mode %d]\n", io_cfg->bbdev_mode);
+        }
+    }
+
+    if (io_cfg->dpdkIoVaMode == 1){
+        snprintf(iova_mode, RTE_DIM(iova_mode), "%s", "--iova-mode=va");
+    }
+
+    if (io_cfg->dpdkMemorySize){
+        snprintf(socket_mem, RTE_DIM(socket_mem), "--socket-mem=%d", io_cfg->dpdkMemorySize);
+        snprintf(socket_limit, RTE_DIM(socket_limit), "--socket-limit=%d", io_cfg->dpdkMemorySize);
+    }
+
+    if (io_cfg->core < 64)
+        c_mask |= (long)(1L << io_cfg->core);
+    else
+        c_mask_64_127 |= (long)(1L << (io_cfg->core - 64));
+
+    if (io_cfg->system_core < 64)
+        c_mask |= (long)(1L << io_cfg->system_core);
+    else
+        c_mask_64_127 |= (long)(1L << (io_cfg->system_core - 64));
+
+    if (io_cfg->timing_core < 64)
+       c_mask |=  (long)(1L << io_cfg->timing_core);
+    else
+       c_mask_64_127 |= (long)(1L << (io_cfg->timing_core - 64));
+
+    nWorkerCore = 1L;
+    for (i = 0; i < coreNum && i < 64; i++) {
+        if (nWorkerCore & (uint64_t)io_cfg->pkt_proc_core) {
+            c_mask |= nWorkerCore;
+        }
+        nWorkerCore = nWorkerCore << 1;
+    }
+
+    nWorkerCore = 1L;
+    for (i = 64; i < coreNum && i < 128; i++) {
+        if (nWorkerCore & (uint64_t)io_cfg->pkt_proc_core_64_127) {
+            c_mask_64_127 |= nWorkerCore;
+        }
+        nWorkerCore = nWorkerCore << 1;
+    }
+
+    printf("total cores %d c_mask 0x%lx%016lx core %d [id] system_core %d [id] pkt_proc_core 0x%lx%016lx [mask] pkt_aux_core %d [id] timing_core %d [id]\n",
+        coreNum, c_mask_64_127, c_mask, io_cfg->core, io_cfg->system_core, io_cfg->pkt_proc_core_64_127, io_cfg->pkt_proc_core, io_cfg->pkt_aux_core, io_cfg->timing_core);
+
+    snprintf(core_mask, sizeof(core_mask), "-c 0x%lx%016lx",c_mask_64_127,c_mask);
+
+    ctx->io_cfg = *io_cfg;
+
+    for (ivf = 0; ivf < XRAN_VF_MAX; ivf++){
+        for (i = 0; i < ID_MAX; i++)     /* Initialize all as broadcast */
+            memset(&ctx->entities[ivf][i], 0xFF, sizeof(ctx->entities[0][0]));
+    }
+
+    printf("%s: Calling rte_eal_init:", __FUNCTION__);
+    for (i = 0; i < RTE_DIM(argv); i++)
+    {
+        printf("%s ", argv[i]);
+    }
+    printf("\n");
+
+    /* This will return on system_core, which is not necessarily the
+     * one we're on right now. */
+    if (rte_eal_init(RTE_DIM(argv), argv) < 0)
+        rte_panic("Cannot init EAL: %s\n", rte_strerror(rte_errno));
+
+    if (rte_eal_process_type() == RTE_PROC_SECONDARY)
+        rte_exit(EXIT_FAILURE,
+                "Secondary process type not supported.\n");
+
+    xran_init_mbuf_pool(mtu);
+
+#ifdef RTE_LIBRTE_PDUMP
+    /* initialize packet capture framework */
+    rte_pdump_init();
+#endif
+
+    /* Timers. */
+    rte_timer_subsystem_init();
+
+    *lcore_id = rte_get_next_lcore(rte_lcore_id(), 0, 0);
+
+    PANIC_ON(*lcore_id == RTE_MAX_LCORE, "out of lcores for io_loop()");
+
+    for (i = 0; i < XRAN_VF_MAX; i++)
+        port[i] = 0xffff;
+
+    if (rte_eal_process_type() == RTE_PROC_PRIMARY) {
+        for (i = 0; i < XRAN_VF_MAX && i < io_cfg->num_vfs; i++){
+            if(io_cfg->dpdk_dev[i]){
+                struct rte_dev_iterator iterator;
+                uint16_t port_id;
+
+                if (rte_dev_probe(io_cfg->dpdk_dev[i]) != 0 ||
+                    rte_eth_dev_count_avail() == 0) {
+                        errx(1, "Network port doesn't exist\n");
+                }
+
+                RTE_ETH_FOREACH_MATCHING_DEV(port_id, io_cfg->dpdk_dev[i], &iterator){
+                        port[i] = port_id;
+                        xran_init_port(port[i], io_cfg->num_rxq, mtu);
+            }
+
+                if(!(i & 1) || io_cfg->one_vf_cu_plane){
+                snprintf(ring_name, RTE_DIM(ring_name), "%s_%d", "tx_ring_up", i);
+                    ctx->tx_ring[i] = rte_ring_create(ring_name, NUM_MBUFS_RING_TRX,
+                    rte_lcore_to_socket_id(*lcore_id), RING_F_SC_DEQ);
+                    PANIC_ON(ctx->tx_ring[i] == NULL, "failed to allocate tx ring");
+                    for(qi = 0; qi < io_cfg->num_rxq; qi++) {
+                        snprintf(ring_name, RTE_DIM(ring_name), "%s_%d_%d", "rx_ring_up", i, qi);
+                        ctx->rx_ring[i][qi] = rte_ring_create(ring_name, NUM_MBUFS_RING_TRX,
+                            rte_lcore_to_socket_id(*lcore_id), RING_F_SP_ENQ);
+                        PANIC_ON(ctx->rx_ring[i][qi] == NULL, "failed to allocate rx ring");
+                    }
+            }else {
+                snprintf(ring_name, RTE_DIM(ring_name), "%s_%d", "tx_ring_cp", i);
+                    ctx->tx_ring[i] = rte_ring_create(ring_name, NUM_MBUFS_RING_TRX,
+                    rte_lcore_to_socket_id(*lcore_id), RING_F_SC_DEQ);
+                    PANIC_ON(ctx->tx_ring[i] == NULL, "failed to allocate rx ring");
+                    for(qi = 0; qi < io_cfg->num_rxq; qi++) {
+                        snprintf(ring_name, RTE_DIM(ring_name), "%s_%d_%d", "rx_ring_cp", i, qi);
+                        ctx->rx_ring[i][qi] = rte_ring_create(ring_name, NUM_MBUFS_RING_TRX,
+                            rte_lcore_to_socket_id(*lcore_id), RING_F_SP_ENQ);
+                        PANIC_ON(ctx->rx_ring[i][qi] == NULL, "failed to allocate rx ring");
+                    }
+                }
+            } else {
+                printf("no DPDK port provided\n");
+                xran_init_port_mempool(i, mtu);
+            }
+
+            if(io_cfg->dpdk_dev[i]){
+                check_port_link_status(port[i]);
+            }
+        }
+    } else {
+        rte_panic("ethdi_dpdk_io_loop() failed to start  with RTE_PROC_SECONDARY\n");
+    }
+
+    for (i = 0; i < XRAN_VF_MAX && i < io_cfg->num_vfs; i++){
+        ctx->io_cfg.port[i] = port[i];
+        print_dbg("port_id 0x%04x\n", ctx->io_cfg.port[i]);
+    }
+
+    for (i = 0; i < XRAN_VF_MAX; i++){
+        ctx->vf2xran_port[i] = 0xFFFF;
+        ctx->rxq_per_port[i] = 1;
+        for (qi = 0; qi < XRAN_VF_QUEUE_MAX; qi++){
+            ctx->vf_and_q2pc_id[i][qi] = 0xFFFF;
+
+            ctx->vf_and_q2cid[i][qi].cuPortId     = 0xFF;
+            ctx->vf_and_q2cid[i][qi].bandSectorId = 0xFF;
+            ctx->vf_and_q2cid[i][qi].ccId         = 0xFF;
+            ctx->vf_and_q2cid[i][qi].ruPortId     = 0xFF;
+        }
+    }
+
+    for (i = 0; i < XRAN_VF_MAX && i < io_cfg->num_vfs; i++){
+        if(io_cfg->dpdk_dev[i]){
+            struct rte_ether_addr *p_addr;
+
+            if(i % (io_cfg->nEthLinePerPort * (2 - 1*ctx->io_cfg.one_vf_cu_plane)) == 0) /* C-p and U-p VFs per line */
+                xran_port +=1;
+
+            rte_eth_macaddr_get(port[i], &ctx->entities[i][io_cfg->id]);
+
+            p_addr = &ctx->entities[i][io_cfg->id];
+            printf("[%2d] vf %2u local  SRC MAC: %02"PRIx8" %02"PRIx8" %02"PRIx8
+                   " %02"PRIx8" %02"PRIx8" %02"PRIx8"\n",
+                   (unsigned)xran_port,
+                   (unsigned)i,
+                   p_addr->addr_bytes[0], p_addr->addr_bytes[1], p_addr->addr_bytes[2],
+                   p_addr->addr_bytes[3], p_addr->addr_bytes[4], p_addr->addr_bytes[5]);
+
+            p_addr = &p_ru_addr[i];
+            printf("[%2d] vf %2u remote DST MAC: %02"PRIx8" %02"PRIx8" %02"PRIx8
+                   " %02"PRIx8" %02"PRIx8" %02"PRIx8"\n",
+                   (unsigned)xran_port,
+                   (unsigned)i,
+                   p_addr->addr_bytes[0], p_addr->addr_bytes[1], p_addr->addr_bytes[2],
+                   p_addr->addr_bytes[3], p_addr->addr_bytes[4], p_addr->addr_bytes[5]);
+
+            rte_ether_addr_copy(&p_ru_addr[i],  &ctx->entities[i][ID_O_RU]);
+            ctx->vf2xran_port[i] = xran_port;
+            ctx->rxq_per_port[i] = io_cfg->num_rxq;
+        }
+    }
+
+    for(i = 0; i < xran_port + 1 && i < XRAN_PORTS_NUM; i++) {
+        snprintf(ring_name, RTE_DIM(ring_name), "%s_%d", "dl_gen_ring_up", i);
+        ctx->up_dl_pkt_gen_ring[i] = rte_ring_create(ring_name, NUM_MBUFS_RING,
+        rte_lcore_to_socket_id(*lcore_id), /*RING_F_SC_DEQ*/0);
+        PANIC_ON(ctx->up_dl_pkt_gen_ring[i] == NULL, "failed to allocate dl gen ring");
+        printf("created %s\n", ring_name);
+    }
+
+    return 1;
+}
+
+static inline uint16_t xran_tx_from_ring(int port, struct rte_ring *r)
+{
+    struct rte_mbuf *mbufs[BURST_SIZE];
+    uint16_t dequeued, sent = 0;
+    uint32_t remaining;
+    int i;
+    long t1 = MLogTick();
+
+    dequeued = rte_ring_dequeue_burst(r, (void **)mbufs, BURST_SIZE,
+            &remaining);
+    if (!dequeued)
+        return 0;   /* Nothing to send. */
+
+    while (1) {     /* When tx queue is full it is trying again till succeed */
+        sent += rte_eth_tx_burst(port, 0, &mbufs[sent], dequeued - sent);
+        if (sent == dequeued){
+        MLogTask(PID_RADIO_ETH_TX_BURST, t1, MLogTick());
+            return remaining;
+    }
+}
+}
+
+int32_t process_dpdk_io(void* args)
+{
+    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();
+    struct xran_io_cfg * cfg = &(xran_ethdi_get_ctx()->io_cfg);
+    int32_t* port = &cfg->port[0];
+    int port_id = 0;
+    int qi      = 0;
+
+    rte_timer_manage();
+
+    for (port_id = 0; port_id < XRAN_VF_MAX && port_id < ctx->io_cfg.num_vfs; port_id++){
+        struct rte_mbuf *mbufs[BURST_RX_IO_SIZE];
+        if(port[port_id] == 0xFF)
+            return 0;
+
+        /* RX */
+        for(qi = 0; qi < ctx->rxq_per_port[port_id]; qi++) {
+            const uint16_t rxed = rte_eth_rx_burst(port[port_id], qi, mbufs, BURST_RX_IO_SIZE);
+        if (rxed != 0){
+            unsigned enq_n = 0;
+            long t1 = MLogTick();
+                ctx->rx_vf_queue_cnt[port[port_id]][qi] += rxed;
+                enq_n =  rte_ring_enqueue_burst(ctx->rx_ring[port_id][qi], (void*)mbufs, rxed, NULL);
+            if(rxed - enq_n)
+                rte_panic("error enq\n");
+            MLogTask(PID_RADIO_RX_VALIDATE, t1, MLogTick());
+        }
+        }
+
+        /* TX */
+
+        const uint16_t sent = xran_tx_from_ring(port[port_id], ctx->tx_ring[port_id]);
+        /* One way Delay Measurements */
+        if ((cfg->eowd_cmn[cfg->id].owdm_enable != 0) && (cfg->eowd_cmn[cfg->id].measVf == port_id))
+        {
+          if (!xran_ecpri_port_update_required(cfg, (uint16_t)port_id))
+            {
+#ifdef ORAN_OWD_DEBUG_TX_LOOP
+                printf("going to owd tx for port %d\n", port_id);
+#endif
+                if (xran_ecpri_one_way_delay_measurement_transmitter((uint16_t) port_id, (void*)xran_dev_get_ctx()) != OK)
+                {
+                    errx(1,"Exit pdio port_id %d", port_id);
+                }
+            }
+        }
+
+        if (XRAN_STOPPED == xran_if_current_state)
+            return -1;
+    }
+
+    if (XRAN_STOPPED == xran_if_current_state)
+            return -1;
+
+    return 0;
+}
+
+int32_t process_dpdk_io_tx(void* args)
+{
+    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();
+    struct xran_io_cfg * cfg = &(xran_ethdi_get_ctx()->io_cfg);
+    int32_t* port = &cfg->port[0];
+    int port_id = 0;
+
+    //rte_timer_manage();
+
+    for (port_id = 0; port_id < XRAN_VF_MAX && port_id < ctx->io_cfg.num_vfs; port_id++){
+        struct rte_mbuf *mbufs[BURST_RX_IO_SIZE];
+        if(port[port_id] == 0xFF)
+            return 0;
+        /* TX */
+        const uint16_t sent = xran_tx_from_ring(port[port_id], ctx->tx_ring[port_id]);
+
+        if (XRAN_STOPPED == xran_if_current_state)
+            return -1;
+    }
+
+    if (XRAN_STOPPED == xran_if_current_state)
+            return -1;
+
+    return 0;
+}
+
+int32_t process_dpdk_io_rx(void* args)
+{
+    struct xran_ethdi_ctx *ctx = xran_ethdi_get_ctx();
+    struct xran_io_cfg * cfg = &(xran_ethdi_get_ctx()->io_cfg);
+    int32_t* port = &cfg->port[0];
+    int port_id = 0;
+    int qi     = 0;
+
+    rte_timer_manage();
+
+    for (port_id = 0; port_id < XRAN_VF_MAX && port_id < ctx->io_cfg.num_vfs; port_id++){
+        struct rte_mbuf *mbufs[BURST_RX_IO_SIZE];
+        if(port[port_id] == 0xFF)
+            return 0;
+
+        /* RX */
+        for(qi = 0; qi < ctx->rxq_per_port[port_id]; qi++){
+            const uint16_t rxed = rte_eth_rx_burst(port[port_id], qi, mbufs, BURST_RX_IO_SIZE);
+            if (rxed != 0){
+                unsigned enq_n = 0;
+                long t1 = MLogTick();
+                ctx->rx_vf_queue_cnt[port[port_id]][qi] += rxed;
+                enq_n =  rte_ring_enqueue_burst(ctx->rx_ring[port_id][qi], (void*)mbufs, rxed, NULL);
+                if(rxed - enq_n)
+                    rte_panic("error enq\n");
+                MLogTask(PID_RADIO_RX_VALIDATE, t1, MLogTick());
+            }
+        }
+        if (XRAN_STOPPED == xran_if_current_state)
+            return -1;
+    }
+
+    if (XRAN_STOPPED == xran_if_current_state)
+            return -1;
+
+    return 0;
+}
+