/* Copyright (C) 2021-2025 Open Information Security Foundation
 *
 * You can copy, redistribute or modify this Program under the terms of
 * the GNU General Public License version 2 as published by the Free
 * Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 */

/**
 *  \defgroup dpdk DPDK running mode
 *
 *  @{
 */

/**
 * \file
 *
 * \author Lukas Sismis <lukas.sismis@gmail.com>
 *
 * DPDK capture interface
 *
 */

#include "suricata-common.h"
#include "runmodes.h"
#include "decode.h"
#include "packet.h"
#include "source-dpdk.h"
#include "suricata.h"
#include "threads.h"
#include "threadvars.h"
#include "tm-threads.h"
#include "tmqh-packetpool.h"
#include "util-privs.h"
#include "util-device-private.h"
#include "action-globals.h"

#ifndef HAVE_DPDK

TmEcode NoDPDKSupportExit(ThreadVars *, const void *, void **);

void TmModuleReceiveDPDKRegister(void)
{
    tmm_modules[TMM_RECEIVEDPDK].name = "ReceiveDPDK";
    tmm_modules[TMM_RECEIVEDPDK].ThreadInit = NoDPDKSupportExit;
    tmm_modules[TMM_RECEIVEDPDK].Func = NULL;
    tmm_modules[TMM_RECEIVEDPDK].ThreadExitPrintStats = NULL;
    tmm_modules[TMM_RECEIVEDPDK].ThreadDeinit = NULL;
    tmm_modules[TMM_RECEIVEDPDK].cap_flags = 0;
    tmm_modules[TMM_RECEIVEDPDK].flags = TM_FLAG_RECEIVE_TM;
}

/**
 * \brief Registration Function for DecodeDPDK.
 */
void TmModuleDecodeDPDKRegister(void)
{
    tmm_modules[TMM_DECODEDPDK].name = "DecodeDPDK";
    tmm_modules[TMM_DECODEDPDK].ThreadInit = NoDPDKSupportExit;
    tmm_modules[TMM_DECODEDPDK].Func = NULL;
    tmm_modules[TMM_DECODEDPDK].ThreadExitPrintStats = NULL;
    tmm_modules[TMM_DECODEDPDK].ThreadDeinit = NULL;
    tmm_modules[TMM_DECODEDPDK].cap_flags = 0;
    tmm_modules[TMM_DECODEDPDK].flags = TM_FLAG_DECODE_TM;
}

/**
 * \brief this function prints an error message and exits.
 */
TmEcode NoDPDKSupportExit(ThreadVars *tv, const void *initdata, void **data)
{
    FatalError("Error creating thread %s: you do not have "
               "support for DPDK enabled, on Linux host please recompile "
               "with --enable-dpdk",
            tv->name);
}

#else /* We have DPDK support */

#include "util-affinity.h"
#include "util-dpdk.h"
#include "util-dpdk-i40e.h"
#include "util-dpdk-ice.h"
#include "util-dpdk-ixgbe.h"
#include "util-dpdk-mlx5.h"
#include "util-dpdk-bonding.h"
#include "util-ring-buffer.h"
#include <numa.h>

// interrupt mode constants
#define MIN_ZERO_POLL_COUNT          10U
#define MIN_ZERO_POLL_COUNT_TO_SLEEP 10U
#define MINIMUM_SLEEP_TIME_US        1U
#define STANDARD_SLEEP_TIME_US       100U
#define MAX_EPOLL_TIMEOUT_MS         500U
static rte_spinlock_t intr_lock[RTE_MAX_ETHPORTS];

/**
 * \brief Structure to hold thread specific variables.
 */
typedef struct DPDKThreadVars_ {
    /* counters */
    uint64_t pkts;
    ThreadVars *tv;
    TmSlot *slot;
    LiveDevice *livedev;
    ChecksumValidationMode checksum_mode;
    bool intr_enabled;
    /* references to packet and drop counters */
    StatsCounterId capture_dpdk_packets;
    StatsCounterId capture_dpdk_rx_errs;
    StatsCounterId capture_dpdk_imissed;
    StatsCounterId capture_dpdk_rx_no_mbufs;
    StatsCounterId capture_dpdk_ierrors;
    StatsCounterId capture_dpdk_tx_errs;
    unsigned int flags;
    uint16_t threads;
    /* for IPS */
    DpdkCopyModeEnum copy_mode;
    uint16_t out_port_id;
    /* Entry in the peers_list */

    uint64_t bytes;
    uint64_t accepted;
    uint64_t dropped;
    uint16_t port_id;
    uint16_t queue_id;
    int32_t port_socket_id;
    struct rte_mbuf *received_mbufs[DPDK_RX_BURST_SIZE];
    DPDKWorkerSync *workers_sync;
    SCRingBuffer *rx_backlog;
    bool backlog_capacity_limited;
    StatsCounterId capture_dpdk_backlog_current;
    StatsCounterMaxId capture_dpdk_backlog_max;
    StatsCounterId capture_dpdk_backlog_enqueued;
    StatsCounterId capture_dpdk_backlog_capacity_hits;
    StatsCounterId capture_dpdk_backlog_drain_budget_hits;
} DPDKThreadVars;

static TmEcode ReceiveDPDKThreadInit(ThreadVars *, const void *, void **);
static TmEcode ReceiveDPDKThreadDeinit(ThreadVars *, void *);
static TmEcode ReceiveDPDKLoop(ThreadVars *tv, void *data, void *slot);

static TmEcode DecodeDPDKThreadInit(ThreadVars *, const void *, void **);
static TmEcode DecodeDPDKThreadDeinit(ThreadVars *tv, void *data);
static TmEcode DecodeDPDK(ThreadVars *, Packet *, void *);

static void DPDKFreeMbufArray(struct rte_mbuf **mbuf_array, uint16_t mbuf_cnt, uint16_t offset);
static bool InterruptsRXEnable(uint16_t port_id, uint16_t queue_id)
{
    uint32_t event_data = (uint32_t)port_id << UINT16_WIDTH | queue_id;
    int32_t ret = rte_eth_dev_rx_intr_ctl_q(port_id, queue_id, RTE_EPOLL_PER_THREAD,
            RTE_INTR_EVENT_ADD, (void *)((uintptr_t)event_data));

    if (ret != 0) {
        SCLogError("%s-Q%d: failed to enable interrupt mode: %s", DPDKGetPortNameByPortID(port_id),
                queue_id, rte_strerror(-ret));
        return false;
    }
    return true;
}

static inline uint32_t InterruptsSleepHeuristic(uint32_t no_pkt_polls_count)
{
    if (no_pkt_polls_count < MIN_ZERO_POLL_COUNT_TO_SLEEP)
        return MINIMUM_SLEEP_TIME_US;

    return STANDARD_SLEEP_TIME_US;
}

static inline void InterruptsTurnOnOff(uint16_t port_id, uint16_t queue_id, bool on)
{
    rte_spinlock_lock(&(intr_lock[port_id]));

    if (on)
        rte_eth_dev_rx_intr_enable(port_id, queue_id);
    else
        rte_eth_dev_rx_intr_disable(port_id, queue_id);

    rte_spinlock_unlock(&(intr_lock[port_id]));
}

static inline void DPDKFreeMbufArray(
        struct rte_mbuf **mbuf_array, uint16_t mbuf_cnt, uint16_t offset)
{
    for (int i = offset; i < mbuf_cnt; i++) {
        rte_pktmbuf_free(mbuf_array[i]);
    }
}

static void DevicePostStartPMDSpecificActions(DPDKThreadVars *ptv, const char *driver_name)
{
    if (strcmp(driver_name, "net_bonding") == 0)
        driver_name = BondingDeviceDriverGet(ptv->port_id);
    if (strcmp(driver_name, "net_i40e") == 0)
        i40eDeviceSetRSS(ptv->port_id, ptv->threads, ptv->livedev->dev);
    else if (strcmp(driver_name, "net_ixgbe") == 0)
        ixgbeDeviceSetRSS(ptv->port_id, ptv->threads, ptv->livedev->dev);
    else if (strcmp(driver_name, "net_ice") == 0)
        iceDeviceSetRSS(ptv->port_id, ptv->threads, ptv->livedev->dev);
    else if (strcmp(driver_name, "mlx5_pci") == 0)
        mlx5DeviceSetRSS(ptv->port_id, ptv->threads, ptv->livedev->dev);
}

static void DevicePreClosePMDSpecificActions(DPDKThreadVars *ptv, const char *driver_name)
{
    if (strcmp(driver_name, "net_bonding") == 0) {
        driver_name = BondingDeviceDriverGet(ptv->port_id);
    }

    if (strcmp(driver_name, "net_i40e") == 0 || strcmp(driver_name, "net_ixgbe") == 0 ||
            strcmp(driver_name, "net_ice") == 0 || strcmp(driver_name, "mlx5_pci") == 0) {
        // Flush the RSS rules that have been inserted in the post start section
        struct rte_flow_error flush_error = { 0 };
        int32_t retval = rte_flow_flush(ptv->port_id, &flush_error);
        if (retval != 0) {
            SCLogError("%s: unable to flush rte_flow rules: %s Flush error msg: %s",
                    ptv->livedev->dev, rte_strerror(-retval), flush_error.message);
        }
    }
}

/**
 * Attempts to retrieve NUMA node id on which the caller runs
 * @return NUMA id on success, -1 otherwise
 */
static int GetNumaNode(void)
{
    int cpu = 0;
    int node = -1;

#if defined(__linux__)
    cpu = sched_getcpu();
    node = numa_node_of_cpu(cpu);
#else
    SCLogWarning("NUMA node retrieval is not supported on this OS.");
#endif

    return node;
}

/**
 * \brief Registration Function for ReceiveDPDK.
 * \todo Unit tests are needed for this module.
 */
void TmModuleReceiveDPDKRegister(void)
{
    tmm_modules[TMM_RECEIVEDPDK].name = "ReceiveDPDK";
    tmm_modules[TMM_RECEIVEDPDK].ThreadInit = ReceiveDPDKThreadInit;
    tmm_modules[TMM_RECEIVEDPDK].Func = NULL;
    tmm_modules[TMM_RECEIVEDPDK].PktAcqLoop = ReceiveDPDKLoop;
    tmm_modules[TMM_RECEIVEDPDK].PktAcqBreakLoop = NULL;
    tmm_modules[TMM_RECEIVEDPDK].ThreadExitPrintStats = NULL;
    tmm_modules[TMM_RECEIVEDPDK].ThreadDeinit = ReceiveDPDKThreadDeinit;
    tmm_modules[TMM_RECEIVEDPDK].cap_flags = SC_CAP_NET_RAW;
    tmm_modules[TMM_RECEIVEDPDK].flags = TM_FLAG_RECEIVE_TM;
}

/**
 * \brief Registration Function for DecodeDPDK.
 * \todo Unit tests are needed for this module.
 */
void TmModuleDecodeDPDKRegister(void)
{
    tmm_modules[TMM_DECODEDPDK].name = "DecodeDPDK";
    tmm_modules[TMM_DECODEDPDK].ThreadInit = DecodeDPDKThreadInit;
    tmm_modules[TMM_DECODEDPDK].Func = DecodeDPDK;
    tmm_modules[TMM_DECODEDPDK].ThreadExitPrintStats = NULL;
    tmm_modules[TMM_DECODEDPDK].ThreadDeinit = DecodeDPDKThreadDeinit;
    tmm_modules[TMM_DECODEDPDK].cap_flags = 0;
    tmm_modules[TMM_DECODEDPDK].flags = TM_FLAG_DECODE_TM;
}

static inline void DPDKDumpBacklogCounters(DPDKThreadVars *ptv)
{
    if (ptv->rx_backlog == NULL) {
        return;
    }

    StatsCounterSetI64(
            &ptv->tv->stats, ptv->capture_dpdk_backlog_current, SCRingBufferCount(ptv->rx_backlog));
}

static void DPDKBacklogClear(DPDKThreadVars *ptv)
{
    if (ptv->rx_backlog == NULL) {
        return;
    }

    struct rte_mbuf *mbuf;
    while (SCRingBufferDequeue(ptv->rx_backlog, &mbuf, 1)) {
        rte_pktmbuf_free(mbuf);
    }
}

static void DPDKBacklogFree(DPDKThreadVars *ptv)
{
    DPDKBacklogClear(ptv);
    SCRingBufferFree(ptv->rx_backlog);
    ptv->rx_backlog = NULL;
}

static inline void DPDKDumpCounters(DPDKThreadVars *ptv)
{
    DPDKDumpBacklogCounters(ptv);
    /* Some NICs (e.g. Intel) do not support queue statistics and the drops can be fetched only on
     * the port level. Therefore setting it to the first worker to have at least continuous update
     * on the dropped packets. */
    if (ptv->queue_id == 0) {
        struct rte_eth_stats eth_stats;
        int retval = rte_eth_stats_get(ptv->port_id, &eth_stats);
        if (unlikely(retval != 0)) {
            SCLogError("%s: failed to get stats: %s", ptv->livedev->dev, rte_strerror(-retval));
            return;
        }

        StatsCounterSetI64(&ptv->tv->stats, ptv->capture_dpdk_packets,
                ptv->pkts + eth_stats.imissed + eth_stats.ierrors + eth_stats.rx_nombuf);
        SC_ATOMIC_SET(ptv->livedev->pkts,
                eth_stats.ipackets + eth_stats.imissed + eth_stats.ierrors + eth_stats.rx_nombuf);
        StatsCounterSetI64(&ptv->tv->stats, ptv->capture_dpdk_rx_errs,
                eth_stats.imissed + eth_stats.ierrors + eth_stats.rx_nombuf);
        StatsCounterSetI64(&ptv->tv->stats, ptv->capture_dpdk_imissed, eth_stats.imissed);
        StatsCounterSetI64(&ptv->tv->stats, ptv->capture_dpdk_rx_no_mbufs, eth_stats.rx_nombuf);
        StatsCounterSetI64(&ptv->tv->stats, ptv->capture_dpdk_ierrors, eth_stats.ierrors);
        StatsCounterSetI64(&ptv->tv->stats, ptv->capture_dpdk_tx_errs, eth_stats.oerrors);
        SC_ATOMIC_SET(
                ptv->livedev->drop, eth_stats.imissed + eth_stats.ierrors + eth_stats.rx_nombuf);
    } else {
        StatsCounterSetI64(&ptv->tv->stats, ptv->capture_dpdk_packets, ptv->pkts);
    }
}

static void DPDKReleasePacket(Packet *p)
{
    int retval;
    /* Need to be in copy mode and need to detect early release
       where Ethernet header could not be set (and pseudo packet)
       When enabling promiscuous mode on Intel cards, 2 ICMPv6 packets are generated.
       These get into the infinite cycle between the NIC and the switch in some cases */
    if ((p->dpdk_v.copy_mode == DPDK_COPY_MODE_TAP ||
                (p->dpdk_v.copy_mode == DPDK_COPY_MODE_IPS && !PacketCheckAction(p, ACTION_DROP)))
#if defined(RTE_LIBRTE_I40E_PMD) || defined(RTE_LIBRTE_IXGBE_PMD) || defined(RTE_LIBRTE_ICE_PMD)
            && !(PacketIsICMPv6(p) && PacketGetICMPv6(p)->type == 143)
#endif
    ) {
        BUG_ON(PKT_IS_PSEUDOPKT(p));
        retval =
                rte_eth_tx_burst(p->dpdk_v.out_port_id, p->dpdk_v.out_queue_id, &p->dpdk_v.mbuf, 1);
        // rte_eth_tx_burst can return only 0 (failure) or 1 (success) because we are only
        // transmitting burst of size 1 and the function rte_eth_tx_burst returns number of
        // successfully sent packets.
        if (unlikely(retval < 1)) {
            // sometimes a repeated transmit can help to send out the packet
            rte_delay_us(DPDK_BURST_TX_WAIT_US);
            retval = rte_eth_tx_burst(
                    p->dpdk_v.out_port_id, p->dpdk_v.out_queue_id, &p->dpdk_v.mbuf, 1);
            if (unlikely(retval < 1)) {
                SCLogDebug("Unable to transmit the packet on port %u queue %u",
                        p->dpdk_v.out_port_id, p->dpdk_v.out_queue_id);
                rte_pktmbuf_free(p->dpdk_v.mbuf);
                p->dpdk_v.mbuf = NULL;
            }
        }
    } else {
        rte_pktmbuf_free(p->dpdk_v.mbuf);
        p->dpdk_v.mbuf = NULL;
    }

    PacketFreeOrRelease(p);
}

static TmEcode ReceiveDPDKLoopInit(ThreadVars *tv, DPDKThreadVars *ptv)
{
    SCEnter();
    // Indicate that the thread is actually running its application level
    // code (i.e., it can poll packets)
    TmThreadsSetFlag(tv, THV_RUNNING);
    PacketPoolWait();

    rte_eth_stats_reset(ptv->port_id);
    rte_eth_xstats_reset(ptv->port_id);

    if (ptv->intr_enabled && !InterruptsRXEnable(ptv->port_id, ptv->queue_id))
        SCReturnInt(TM_ECODE_FAILED);

    SCReturnInt(TM_ECODE_OK);
}

static inline void LoopHandleTimeoutOnIdle(ThreadVars *tv)
{
    static thread_local uint64_t last_timeout_msec = 0;
    SCTime_t t = TimeGet();
    uint64_t msecs = SCTIME_MSECS(t);
    if (msecs > last_timeout_msec + 100) {
        TmThreadsCaptureHandleTimeout(tv, NULL);
        last_timeout_msec = msecs;
    }
}

/**
 * \brief Decides if it should retry the packet poll or continue with the packet processing
 * \return true if the poll should be retried, false otherwise
 */
static inline bool RXPacketCountHeuristic(ThreadVars *tv, DPDKThreadVars *ptv, uint16_t nb_rx)
{
    static thread_local uint32_t zero_pkt_polls_cnt = 0;

    if (nb_rx > 0) {
        zero_pkt_polls_cnt = 0;
        return false;
    }

    LoopHandleTimeoutOnIdle(tv);
    if (!ptv->intr_enabled)
        return true;

    zero_pkt_polls_cnt++;
    if (zero_pkt_polls_cnt <= MIN_ZERO_POLL_COUNT)
        return true;

    uint32_t pwd_idle_hint = InterruptsSleepHeuristic(zero_pkt_polls_cnt);
    if (pwd_idle_hint < STANDARD_SLEEP_TIME_US) {
        rte_delay_us(pwd_idle_hint);
    } else {
        InterruptsTurnOnOff(ptv->port_id, ptv->queue_id, true);
        struct rte_epoll_event event;
        rte_epoll_wait(RTE_EPOLL_PER_THREAD, &event, 1, MAX_EPOLL_TIMEOUT_MS);
        InterruptsTurnOnOff(ptv->port_id, ptv->queue_id, false);
        return true;
    }

    return false;
}

/**
 * \brief Initializes a packet from an mbuf
 * \return true if the packet was initialized successfully, false otherwise
 */
static inline Packet *PacketInitFromMbuf(
        DPDKThreadVars *ptv, struct rte_mbuf *mbuf, SCTime_t timestamp)
{
    Packet *p = PacketGetFromQueueOrAlloc();
    if (unlikely(p == NULL)) {
        return NULL;
    }
    PKT_SET_SRC(p, PKT_SRC_WIRE);
    p->datalink = LINKTYPE_ETHERNET;
    if (ptv->checksum_mode == CHECKSUM_VALIDATION_DISABLE) {
        p->flags |= PKT_IGNORE_CHECKSUM;
    }

    p->ts = timestamp;
    p->dpdk_v.mbuf = mbuf;
    p->ReleasePacket = DPDKReleasePacket;
    p->dpdk_v.copy_mode = ptv->copy_mode;
    p->dpdk_v.out_port_id = ptv->out_port_id;
    p->dpdk_v.out_queue_id = ptv->queue_id;
    p->livedev_id = ptv->livedev->id;

    if (ptv->checksum_mode == CHECKSUM_VALIDATION_DISABLE) {
        p->flags |= PKT_IGNORE_CHECKSUM;
    } else if (ptv->checksum_mode == CHECKSUM_VALIDATION_OFFLOAD) {
        uint64_t ol_flags = p->dpdk_v.mbuf->ol_flags;
        if ((ol_flags & RTE_MBUF_F_RX_IP_CKSUM_MASK) == RTE_MBUF_F_RX_IP_CKSUM_GOOD &&
                (ol_flags & RTE_MBUF_F_RX_L4_CKSUM_MASK) == RTE_MBUF_F_RX_L4_CKSUM_GOOD) {
            SCLogDebug("HW detected GOOD IP and L4 chsum, ignoring validation");
            p->flags |= PKT_IGNORE_CHECKSUM;
        } else {
            if ((ol_flags & RTE_MBUF_F_RX_IP_CKSUM_MASK) == RTE_MBUF_F_RX_IP_CKSUM_BAD) {
                SCLogDebug("HW detected BAD IP checksum");
                // chsum recalc will not be triggered but rule keyword check will be
                p->l3.csum_set = true;
                p->l3.csum = 0;
            }
            if ((ol_flags & RTE_MBUF_F_RX_L4_CKSUM_MASK) == RTE_MBUF_F_RX_L4_CKSUM_BAD) {
                SCLogDebug("HW detected BAD L4 chsum");
                p->l4.csum_set = true;
                p->l4.csum = 0;
            }
        }
    }

    return p;
}

static inline void DPDKSegmentedMbufWarning(struct rte_mbuf *mbuf)
{
    static thread_local bool segmented_mbufs_warned = false;
    if (!segmented_mbufs_warned && !rte_pktmbuf_is_contiguous(mbuf)) {
        char warn_s[] = "Segmented mbufs detected! Redmine Ticket #6012 "
                        "Check your configuration or report the issue";
        enum rte_proc_type_t eal_t = rte_eal_process_type();
        if (eal_t == RTE_PROC_SECONDARY) {
            SCLogWarning("%s. To avoid segmented mbufs, "
                         "try to increase mbuf size in your primary application",
                    warn_s);
        } else if (eal_t == RTE_PROC_PRIMARY) {
            SCLogWarning("%s. To avoid segmented mbufs, "
                         "try to increase MTU in your suricata.yaml",
                    warn_s);
        }

        segmented_mbufs_warned = true;
    }
}

static inline TmEcode DPDKProcessMbuf(
        DPDKThreadVars *ptv, struct rte_mbuf *mbuf, SCTime_t timestamp)
{
    Packet *p = PacketInitFromMbuf(ptv, mbuf, timestamp);
    if (p == NULL) {
        rte_pktmbuf_free(mbuf);
        return TM_ECODE_OK;
    }
    DPDKSegmentedMbufWarning(mbuf);
    PacketSetData(p, rte_pktmbuf_mtod(mbuf, uint8_t *), rte_pktmbuf_pkt_len(mbuf));
    /* The processing pipeline releases the Packet on both success and failure. */
    return TmThreadsSlotProcessPkt(ptv->tv, ptv->slot, p);
}

static inline TmEcode DPDKProcessMbufArray(DPDKThreadVars *ptv, uint16_t nb_rx)
{
    SCTime_t timestamp = TimeGet();
    for (uint16_t i = 0; i < nb_rx; i++) {
        if (DPDKProcessMbuf(ptv, ptv->received_mbufs[i], timestamp) != TM_ECODE_OK) {
            DPDKFreeMbufArray(ptv->received_mbufs, nb_rx, i + 1);
            return TM_ECODE_FAILED;
        }
    }
    return TM_ECODE_OK;
}

static inline void DPDKBacklogUpdateCapacity(DPDKThreadVars *ptv)
{
    uint32_t free_slots = SCRingBufferSpace(ptv->rx_backlog);
    if (free_slots >= DPDK_RX_BURST_SIZE) {
        // ends episode of the full backlog
        ptv->backlog_capacity_limited = false;
    } else if (free_slots < DPDK_RX_BURST_ALIGNMENT && !ptv->backlog_capacity_limited) {
        // starts episode of the full backlog
        StatsCounterIncr(&ptv->tv->stats, ptv->capture_dpdk_backlog_capacity_hits);
        ptv->backlog_capacity_limited = true;
    }
}

static inline void DPDKBacklogRecordBurst(DPDKThreadVars *ptv, uint16_t count)
{
    StatsCounterAddI64(&ptv->tv->stats, ptv->capture_dpdk_backlog_enqueued, count);
    StatsCounterMaxUpdateI64(
            &ptv->tv->stats, ptv->capture_dpdk_backlog_max, SCRingBufferCount(ptv->rx_backlog));
    DPDKDumpBacklogCounters(ptv);
    DPDKBacklogUpdateCapacity(ptv);
}

static inline void DPDKBacklogEnqueueBurst(DPDKThreadVars *ptv, uint16_t count)
{
    bool written = SCRingBufferEnqueue(ptv->rx_backlog, ptv->received_mbufs, count);
    BUG_ON(!written);
    DPDKBacklogRecordBurst(ptv, count);
}

/**
 * \brief Receive into backlog storage, using the receive array only for an unaligned wrap.
 * \param ptv The DPDK thread variables.
 * \param[in,out] request Pointer to the number of requested packets to receive.
 * \return The number of packets actually received into the backlog.
 */
static inline uint16_t DPDKBacklogReceive(DPDKThreadVars *ptv, uint16_t *request)
{
    uint32_t contiguous;
    struct rte_mbuf **rx = SCRingBufferEnqueueSpanGet(ptv->rx_backlog, &contiguous);
    uint16_t nb_rx = 0;
    if (contiguous >= DPDK_RX_BURST_ALIGNMENT) {
        // Bit-mask to find closest lower multiple of DPDK_RX_BURST_ALIGNMENT
        *request = MIN(*request, contiguous) & ~(DPDK_RX_BURST_ALIGNMENT - 1);
        nb_rx = rte_eth_rx_burst(ptv->port_id, ptv->queue_id, rx, *request);
        if (nb_rx > 0) {
            bool committed = SCRingBufferEnqueueSpanCommit(ptv->rx_backlog, nb_rx);
            BUG_ON(!committed);
            DPDKBacklogRecordBurst(ptv, nb_rx);
        }
    } else {
        // Keep RX requests aligned when only 1..7 slots remain before wrap.
        *request = DPDK_RX_BURST_ALIGNMENT;
        rx = ptv->received_mbufs;
        nb_rx = rte_eth_rx_burst(ptv->port_id, ptv->queue_id, rx, *request);
        if (nb_rx > 0) {
            DPDKBacklogEnqueueBurst(ptv, nb_rx);
        }
    }
    ptv->pkts += nb_rx;
    return nb_rx;
}

static inline void DPDKBacklogDrain(DPDKThreadVars *ptv)
{
    uint16_t drained = 0;
    while (drained < DPDK_RX_DRAIN_BUDGET) {
        uint32_t free_slots = SCRingBufferSpace(ptv->rx_backlog);
        if (free_slots < DPDK_RX_BURST_ALIGNMENT) {
            break;
        }
        uint16_t available = MIN(free_slots, DPDK_RX_DRAIN_BUDGET - drained);
        if (available < DPDK_RX_BURST_ALIGNMENT) {
            break;
        }
        uint16_t nb_rx = DPDKBacklogReceive(ptv, &available);
        drained += nb_rx;
        if (nb_rx < available) {
            break;
        }
    }
    if (drained == DPDK_RX_DRAIN_BUDGET) {
        StatsCounterIncr(&ptv->tv->stats, ptv->capture_dpdk_backlog_drain_budget_hits);
    }
}

static inline TmEcode DPDKBacklogProcess(DPDKThreadVars *ptv)
{
    uint16_t count = MIN(SCRingBufferCount(ptv->rx_backlog), DPDK_BACKLOG_PROCESS_QUANTUM);
    if (unlikely(!SCRingBufferDequeue(ptv->rx_backlog, ptv->received_mbufs, count))) {
        return TM_ECODE_FAILED;
    }
    TmEcode ret = DPDKProcessMbufArray(ptv, count);
    DPDKBacklogUpdateCapacity(ptv);
    DPDKDumpBacklogCounters(ptv);
    return ret;
}

static inline TmEcode DPDKBacklogPoll(DPDKThreadVars *ptv)
{
    DPDKBacklogDrain(ptv);
    return DPDKBacklogProcess(ptv);
}

static TmEcode DPDKBacklogInit(DPDKThreadVars *ptv, uint32_t size)
{
    if (size == 0) {
        return TM_ECODE_OK;
    }

    ptv->rx_backlog = SCRingBufferInit(size, sizeof(struct rte_mbuf *));
    if (ptv->rx_backlog == NULL) {
        SCLogError("%s-Q%u: failed to allocate RX backlog", ptv->livedev->dev, ptv->queue_id);
        return TM_ECODE_FAILED;
    }
    ptv->capture_dpdk_backlog_current =
            StatsRegisterCounter("capture.dpdk.backlog.current", &ptv->tv->stats);
    ptv->capture_dpdk_backlog_max =
            StatsRegisterMaxCounter("capture.dpdk.backlog.max", &ptv->tv->stats);
    ptv->capture_dpdk_backlog_enqueued =
            StatsRegisterCounter("capture.dpdk.backlog.enqueued", &ptv->tv->stats);
    ptv->capture_dpdk_backlog_capacity_hits =
            StatsRegisterCounter("capture.dpdk.backlog.capacity_hits", &ptv->tv->stats);
    ptv->capture_dpdk_backlog_drain_budget_hits =
            StatsRegisterCounter("capture.dpdk.backlog.drain_budget_hits", &ptv->tv->stats);
    return TM_ECODE_OK;
}

static void PrintDPDKPortXstats(uint16_t port_id, const char *port_name)
{
    int ret = rte_eth_xstats_get(port_id, NULL, 0);
    if (ret <= 0) {
        SCLogPerf("%s: unable to obtain rte_eth_xstats (%s)", port_name,
                ret == 0 ? "not supported" : rte_strerror(-ret));
        return;
    }
    unsigned int len = (unsigned int)ret;
    struct rte_eth_xstat_name *xstats_names = NULL;
    struct rte_eth_xstat *xstats = SCCalloc(len, sizeof(*xstats));
    if (xstats == NULL) {
        SCLogWarning("Failed to allocate memory for the rte_eth_xstat structure");
        return;
    }

    ret = rte_eth_xstats_get(port_id, xstats, len);
    if (ret < 0 || (unsigned int)ret > len) {
        SCLogPerf("%s: unable to obtain rte_eth_xstats (%s)", port_name,
                ret < 0 ? rte_strerror(-ret) : "table size too small");
        goto cleanup;
    }
    xstats_names = SCCalloc(len, sizeof(*xstats_names));
    if (xstats_names == NULL) {
        SCLogWarning("Failed to allocate memory for the rte_eth_xstat_name array");
        goto cleanup;
    }
    ret = rte_eth_xstats_get_names(port_id, xstats_names, len);
    if (ret < 0 || (unsigned int)ret > len) {
        SCLogPerf("%s: unable to obtain names of rte_eth_xstats (%s)", port_name,
                ret < 0 ? rte_strerror(-ret) : "table size too small");
        goto cleanup;
    }
    for (unsigned int i = 0; i < len; i++) {
        if (xstats[i].value > 0)
            SCLogPerf("Port %u (%s) - %s: %" PRIu64, port_id, port_name, xstats_names[i].name,
                    xstats[i].value);
    }

cleanup:
    if (xstats != NULL)
        SCFree(xstats);
    if (xstats_names != NULL)
        SCFree(xstats_names);
}

static void HandleShutdown(DPDKThreadVars *ptv)
{
    SCLogDebug("Stopping Suricata!");
    SC_ATOMIC_ADD(ptv->workers_sync->worker_checked_in, 1);
    while (SC_ATOMIC_GET(ptv->workers_sync->worker_checked_in) < ptv->workers_sync->worker_cnt) {
        rte_delay_us(10);
    }
    // Dump counters while device is still running - some drivers (e.g. BNXT) fail
    // to report stats after the device is stopped
    DPDKDumpCounters(ptv);
    StatsSyncCounters(&ptv->tv->stats);
    if (ptv->queue_id == 0) {
        PrintDPDKPortXstats(ptv->port_id, ptv->livedev->dev);
        rte_delay_us(20); // wait for all threads to get out of the sync loop
        SC_ATOMIC_SET(ptv->workers_sync->worker_checked_in, 0);
        // If Suricata runs in peered mode, the peer threads might still want to send
        // packets to our port. Instead, we know, that we are done with the peered port, so
        // we stop it. The peered threads will stop our port.
        if (ptv->copy_mode == DPDK_COPY_MODE_TAP || ptv->copy_mode == DPDK_COPY_MODE_IPS) {
            rte_eth_dev_stop(ptv->out_port_id);
        } else {
            // in IDS we stop our port - no peer threads are running
            rte_eth_dev_stop(ptv->port_id);
        }
    }
}

static void PeriodicDPDKDumpCounters(DPDKThreadVars *ptv)
{
    static thread_local SCTime_t last_dump = { 0 };
    SCTime_t current_time = TimeGet();
    /* Trigger one dump of stats every second */
    if (current_time.secs != last_dump.secs) {
        DPDKDumpCounters(ptv);
        last_dump = current_time;
    }
}

/**
 *  \brief Main DPDK reading Loop function
 */
static TmEcode ReceiveDPDKLoop(ThreadVars *tv, void *data, void *slot)
{
    SCEnter();
    DPDKThreadVars *ptv = (DPDKThreadVars *)data;
    ptv->slot = ((TmSlot *)slot)->slot_next;
    uint32_t mp_sz = ptv->livedev->dpdk_vars->pkt_mp[ptv->queue_id]->size;
    uint16_t burst_size = (uint16_t)MIN(DPDK_RX_BURST_SIZE, mp_sz);

    TmEcode ret = ReceiveDPDKLoopInit(tv, ptv);
    if (ret != TM_ECODE_OK) {
        SCReturnInt(ret);
    }
    while (true) {
        if (unlikely(suricata_ctl_flags != 0)) {
            HandleShutdown(ptv);
            break;
        }

        if (ptv->rx_backlog == NULL) {
            uint16_t nb_rx =
                    rte_eth_rx_burst(ptv->port_id, ptv->queue_id, ptv->received_mbufs, burst_size);
            if (RXPacketCountHeuristic(tv, ptv, nb_rx) == false) {
                ptv->pkts += nb_rx;
                if (nb_rx > 0 && DPDKProcessMbufArray(ptv, nb_rx) != TM_ECODE_OK) {
                    SCReturnInt(TM_ECODE_FAILED);
                }
            }
        } else {
            if (SCRingBufferCount(ptv->rx_backlog) == 0) {
                uint16_t nb_rx = rte_eth_rx_burst(
                        ptv->port_id, ptv->queue_id, ptv->received_mbufs, burst_size);
                if (RXPacketCountHeuristic(tv, ptv, nb_rx) == false) {
                    ptv->pkts += nb_rx;
                    if (nb_rx < burst_size) {
                        if (nb_rx > 0 && DPDKProcessMbufArray(ptv, nb_rx) != TM_ECODE_OK) {
                            SCReturnInt(TM_ECODE_FAILED);
                        }
                    } else {
                        DPDKBacklogEnqueueBurst(ptv, nb_rx);
                    }
                }
            }
            if (SCRingBufferCount(ptv->rx_backlog) > 0 && DPDKBacklogPoll(ptv) != TM_ECODE_OK) {
                DPDKBacklogClear(ptv);
                DPDKDumpBacklogCounters(ptv);
                StatsSyncCounters(&tv->stats);
                SCReturnInt(TM_ECODE_FAILED);
            }
        }

        PeriodicDPDKDumpCounters(ptv);
        StatsSyncCountersIfSignalled(&tv->stats);
    }

    SCReturnInt(TM_ECODE_OK);
}

/**
 * \brief Init function for ReceiveDPDK.
 *
 * \param tv pointer to ThreadVars
 * \param initdata pointer to the interface passed from the user
 * \param data pointer gets populated with DPDKThreadVars
 *
 */
static TmEcode ReceiveDPDKThreadInit(ThreadVars *tv, const void *initdata, void **data)
{
    SCEnter();
    int retval, thread_numa;
    DPDKThreadVars *ptv = NULL;
    DPDKIfaceConfig *dpdk_config = (DPDKIfaceConfig *)initdata;

    if (initdata == NULL) {
        SCLogError("DPDK configuration is NULL in thread initialization");
        goto fail;
    }

    ptv = SCCalloc(1, sizeof(DPDKThreadVars));
    if (unlikely(ptv == NULL)) {
        SCLogError("Unable to allocate memory");
        goto fail;
    }

    ptv->tv = tv;
    ptv->pkts = 0;
    ptv->bytes = 0;

    ptv->livedev = LiveGetDevice(dpdk_config->iface);
    if (unlikely(ptv->livedev == NULL)) {
        SCLogError("Unable to allocate memory for livedev %s", dpdk_config->iface);
        goto fail;
    }

    ptv->capture_dpdk_packets = StatsRegisterCounter("capture.packets", &ptv->tv->stats);
    ptv->capture_dpdk_rx_errs = StatsRegisterCounter("capture.rx_errors", &ptv->tv->stats);
    ptv->capture_dpdk_tx_errs = StatsRegisterCounter("capture.tx_errors", &ptv->tv->stats);
    ptv->capture_dpdk_imissed = StatsRegisterCounter("capture.dpdk.imissed", &ptv->tv->stats);
    ptv->capture_dpdk_rx_no_mbufs = StatsRegisterCounter("capture.dpdk.no_mbufs", &ptv->tv->stats);
    ptv->capture_dpdk_ierrors = StatsRegisterCounter("capture.dpdk.ierrors", &ptv->tv->stats);

    ptv->copy_mode = dpdk_config->copy_mode;
    ptv->checksum_mode = dpdk_config->checksum_mode;

    ptv->threads = dpdk_config->threads;
    ptv->intr_enabled = (dpdk_config->flags & DPDK_IRQ_MODE) != 0;
    ptv->port_id = dpdk_config->port_id;
    ptv->out_port_id = dpdk_config->out_port_id;
    ptv->port_socket_id = dpdk_config->socket_id;

    thread_numa = GetNumaNode();
    if (thread_numa >= 0 && ptv->port_socket_id != SOCKET_ID_ANY &&
            thread_numa != ptv->port_socket_id) {
        SC_ATOMIC_ADD(dpdk_config->inconsistent_numa_cnt, 1);
        SCLogPerf("%s: NIC is on NUMA %d, thread on NUMA %d", dpdk_config->iface,
                ptv->port_socket_id, thread_numa);
    }

    ptv->workers_sync = dpdk_config->workers_sync;
    uint16_t queue_id = SC_ATOMIC_ADD(dpdk_config->queue_id, 1);
    ptv->queue_id = queue_id;

    if (DPDKBacklogInit(ptv, dpdk_config->rx_backlog_size) != TM_ECODE_OK) {
        goto fail;
    }

    // the last thread starts the device
    if (queue_id == dpdk_config->threads - 1) {
        retval = rte_eth_dev_start(ptv->port_id);
        if (retval < 0) {
            SCLogError("%s: error (%s) during device startup", dpdk_config->iface,
                    rte_strerror(-retval));
            goto fail;
        }

        struct rte_eth_dev_info dev_info;
        retval = rte_eth_dev_info_get(ptv->port_id, &dev_info);
        if (retval != 0) {
            SCLogError("%s: error (%s) when getting device info", dpdk_config->iface,
                    rte_strerror(-retval));
            goto fail;
        }

        uint32_t timeout = dpdk_config->linkup_timeout * 10;
        while (timeout > 0) {
            struct rte_eth_link link = { 0 };
            retval = rte_eth_link_get_nowait(ptv->port_id, &link);
            if (retval != 0) {
                if (retval == -ENOTSUP) {
                    SCLogInfo("%s: link status not supported, skipping", dpdk_config->iface);
                } else {
                    SCLogInfo("%s: error (%s) when getting link status, skipping",
                            dpdk_config->iface, rte_strerror(-retval));
                }
                break;
            }
            if (link.link_status) {
                char link_status_str[RTE_ETH_LINK_MAX_STR_LEN];
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
                rte_eth_link_to_str(link_status_str, sizeof(link_status_str), &link);
#pragma GCC diagnostic pop

                SCLogInfo("%s: %s", dpdk_config->iface, link_status_str);
                break;
            }

            rte_delay_ms(100);
            timeout--;
        }

        if (dpdk_config->linkup_timeout && timeout == 0) {
            SCLogWarning("%s: link is down, trying to continue anyway", dpdk_config->iface);
        }

        // some PMDs requires additional actions only after the device has started
        DevicePostStartPMDSpecificActions(ptv, dev_info.driver_name);

        uint16_t inconsistent_numa_cnt = SC_ATOMIC_GET(dpdk_config->inconsistent_numa_cnt);
        if (inconsistent_numa_cnt > 0 && ptv->port_socket_id != SOCKET_ID_ANY) {
            SCLogWarning("%s: NIC is on NUMA %d, %u threads on different NUMA node(s)",
                    dpdk_config->iface, ptv->port_socket_id, inconsistent_numa_cnt);
        } else if (ptv->port_socket_id == SOCKET_ID_ANY && rte_socket_count() > 1) {
            SCLogNotice(
                    "%s: unable to determine NIC's NUMA node, degraded performance can be expected",
                    dpdk_config->iface);
        }
        if (ptv->intr_enabled) {
            rte_spinlock_init(&intr_lock[ptv->port_id]);
        }
    }

    *data = (void *)ptv;
    dpdk_config->DerefFunc(dpdk_config);
    SCReturnInt(TM_ECODE_OK);

fail:
    if (dpdk_config != NULL) {
        dpdk_config->DerefFunc(dpdk_config);
    }
    if (ptv != NULL) {
        DPDKBacklogFree(ptv);
        SCFree(ptv);
    }
    SCReturnInt(TM_ECODE_FAILED);
}

/**
 * \brief DeInit function closes dpdk at exit.
 * \param tv pointer to ThreadVars
 * \param data pointer that gets cast into DPDKThreadVars for ptv
 */
static TmEcode ReceiveDPDKThreadDeinit(ThreadVars *tv, void *data)
{
    SCEnter();
    DPDKThreadVars *ptv = (DPDKThreadVars *)data;

    DPDKBacklogFree(ptv);

    if (ptv->queue_id == 0) {
        struct rte_eth_dev_info dev_info;
        int retval = rte_eth_dev_info_get(ptv->port_id, &dev_info);
        if (retval != 0) {
            SCLogError("%s: error (%s) when getting device info", ptv->livedev->dev,
                    rte_strerror(-retval));
            SCReturnInt(TM_ECODE_FAILED);
        }

        DevicePreClosePMDSpecificActions(ptv, dev_info.driver_name);

        if (ptv->workers_sync) {
            SCFree(ptv->workers_sync);
        }
    }

    SCFree(ptv);
    SCReturnInt(TM_ECODE_OK);
}

/**
 * \brief This function passes off to link type decoders.
 *
 * DecodeDPDK decodes packets from DPDK and passes
 * them off to the proper link type decoder.
 *
 * \param t pointer to ThreadVars
 * \param p pointer to the current packet
 * \param data pointer that gets cast into DPDKThreadVars for ptv
 */
static TmEcode DecodeDPDK(ThreadVars *tv, Packet *p, void *data)
{
    SCEnter();
    DecodeThreadVars *dtv = (DecodeThreadVars *)data;

    BUG_ON(PKT_IS_PSEUDOPKT(p));

    /* update counters */
    DecodeUpdatePacketCounters(tv, dtv, p);

    /* If suri has set vlan during reading, we increase vlan counter */
    if (p->vlan_idx) {
        StatsCounterIncr(&tv->stats, dtv->counter_vlan);
    }

    /* call the decoder */
    DecodeLinkLayer(tv, dtv, p->datalink, p, GET_PKT_DATA(p), GET_PKT_LEN(p));

    PacketDecodeFinalize(tv, dtv, p);

    SCReturnInt(TM_ECODE_OK);
}

static TmEcode DecodeDPDKThreadInit(ThreadVars *tv, const void *initdata, void **data)
{
    SCEnter();
    DecodeThreadVars *dtv = NULL;

    dtv = DecodeThreadVarsAlloc(tv);

    if (dtv == NULL)
        SCReturnInt(TM_ECODE_FAILED);

    DecodeRegisterPerfCounters(dtv, tv);

    *data = (void *)dtv;

    SCReturnInt(TM_ECODE_OK);
}

static TmEcode DecodeDPDKThreadDeinit(ThreadVars *tv, void *data)
{
    SCEnter();
    if (data != NULL)
        DecodeThreadVarsFree(tv, data);
    SCReturnInt(TM_ECODE_OK);
}

#endif /* HAVE_DPDK */
/* eof */
/**
 * @}
 */