You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
suricata/src/source-dpdk.c

1071 lines
37 KiB
C

/* 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 <numa.h>
#define BURST_SIZE 32
// 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;
StatsCounterId capture_dpdk_segmented_drops;
StatsCounterId capture_dpdk_segmented_too_large;
unsigned int flags;
uint16_t threads;
/* for IPS */
DpdkCopyModeEnum copy_mode;
uint16_t out_port_id;
uint16_t out_tx_seg_max; // max segments per packet the copy interface can transmit
bool tx_segmented_mbufs; // forward packets received in multiple mbufs
/* 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[BURST_SIZE];
DPDKWorkerSync *workers_sync;
} 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 (
#if RTE_VERSION > RTE_VERSION_NUM(20, 0, 0, 0)
strcmp(driver_name, "net_i40e") == 0 ||
#endif /* RTE_VERSION > RTE_VERSION_NUM(20, 0, 0, 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 DPDKDumpCounters(DPDKThreadVars *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);
}
}
/**
* \brief Copy the packet data back to the segmented mbuf it was copied from
*
* Suricata modifies the packet data in place (inline stream normalization or
* the replace keyword) without changing the packet length.
*/
static inline void DPDKSegmentedMbufWriteBack(Packet *p)
{
const uint8_t *data = GET_PKT_DATA(p);
const uint32_t len = GET_PKT_LEN(p);
DEBUG_VALIDATE_BUG_ON(len != rte_pktmbuf_pkt_len(p->dpdk_v.mbuf));
uint32_t offset = 0;
for (struct rte_mbuf *seg = p->dpdk_v.mbuf; seg != NULL && offset < len; seg = seg->next) {
uint32_t seg_len = MIN(rte_pktmbuf_data_len(seg), len - offset);
memcpy(rte_pktmbuf_mtod(seg, uint8_t *), data + offset, seg_len);
offset += seg_len;
}
}
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.mbuf != NULL &&
(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));
// packets of segmented mbufs are copies, modifications must reach the mbufs
if (unlikely((p->flags & PKT_STREAM_MODIFIED) && !(p->flags & PKT_ZERO_COPY))) {
DPDKSegmentedMbufWriteBack(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)
{
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 = TimeGet();
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(void)
{
static thread_local bool segmented_mbufs_warned = false;
if (unlikely(!segmented_mbufs_warned)) {
enum rte_proc_type_t eal_t = rte_eal_process_type();
if (eal_t == RTE_PROC_SECONDARY) {
SCLogWarning("Segmented mbufs detected! Suricata processes them when possible, but "
"for better performance, try to increase mbuf size in your primary "
"application");
} else {
SCLogWarning("Segmented mbufs detected! Suricata processes them when possible, but "
"for better performance, try to increase MTU in your suricata.yaml");
}
segmented_mbufs_warned = true;
}
}
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 inline void DPDKSegmentedMbufTxDisabledWarning(void)
{
static thread_local bool disabled_warned = false;
if (unlikely(!disabled_warned)) {
SCLogWarning("Segmented mbufs cannot be forwarded with tx-segmented-mbufs disabled, "
"dropping such packets, see capture.dpdk.segmented_drops");
disabled_warned = true;
}
}
static inline void DPDKSegmentedMbufTooLargeWarning(uint32_t pkt_len)
{
static thread_local bool size_warned = false;
if (unlikely(!size_warned)) {
SCLogWarning("Segmented mbuf larger than max payload size (%u > %d), dropping such "
"packets, see capture.dpdk.segmented_too_large",
pkt_len, MAX_PAYLOAD_SIZE);
size_warned = true;
}
}
static inline void DPDKSegmentedMbufTooManySegmentsWarning(uint16_t nb_segs, uint16_t seg_max)
{
static thread_local bool segs_warned = false;
if (unlikely(!segs_warned)) {
SCLogWarning("Segmented mbuf of %u segments exceeds the %u segments the copy interface "
"can transmit, dropping such packets, see capture.dpdk.segmented_drops",
nb_segs, seg_max);
segs_warned = true;
}
}
static inline void DPDKSegmentedMbufMalformedWarning(void)
{
static thread_local bool malformed_warned = false;
if (unlikely(!malformed_warned)) {
SCLogWarning("Segmented mbuf whose segments do not match its packet length, dropping "
"such packets, see capture.dpdk.segmented_drops");
malformed_warned = true;
}
}
static inline void DPDKSegmentedMbufCopyFailedWarning(void)
{
static thread_local bool copy_warned = false;
if (unlikely(!copy_warned)) {
SCLogWarning("Failed to copy a segmented mbuf into the packet buffer, dropping such "
"packets, see capture.dpdk.segmented_drops");
copy_warned = true;
}
}
/**
* \brief Check that the segments of a chained mbuf add up to its packet length
*/
static inline bool DPDKSegmentedMbufIsValid(const struct rte_mbuf *mbuf)
{
const struct rte_mbuf *seg = mbuf;
uint32_t len = 0;
for (uint16_t i = 0; i < mbuf->nb_segs; i++) {
if (seg == NULL) {
return false;
}
len += rte_pktmbuf_data_len(seg);
seg = seg->next;
}
return seg == NULL && len == rte_pktmbuf_pkt_len(mbuf);
}
/**
* \brief Decide whether a segmented (chained) mbuf can be processed
*
* Drops malformed chains. Then tries rte_pktmbuf_linearize() which copies all
* segment data into the first segment. If that fails (not enough tailroom or a
* shared first segment), the data is later copied into the packet's buffer,
* which holds at most MAX_PAYLOAD_SIZE bytes. In the copy modes, the mbuf chain
* itself is forwarded, so tx-segmented-mbufs must be enabled and the chain must
* not have more segments than the copy interface can transmit.
*
* \param ptv Pointer to the thread variables
* \param mbuf Pointer to the DPDK mbuf
* \return true if the mbuf can be processed, false if it must be dropped
*/
static inline bool DPDKSegmentedMbufAccept(DPDKThreadVars *ptv, struct rte_mbuf *mbuf)
{
DPDKSegmentedMbufWarning();
if (unlikely(!DPDKSegmentedMbufIsValid(mbuf))) {
DPDKSegmentedMbufMalformedWarning();
StatsCounterIncr(&ptv->tv->stats, ptv->capture_dpdk_segmented_drops);
return false;
}
// linearize writes into the first segment, which must not be shared
if (RTE_MBUF_DIRECT(mbuf) && rte_mbuf_refcnt_read(mbuf) == 1 &&
rte_pktmbuf_linearize(mbuf) == 0) {
return true;
}
if (unlikely(rte_pktmbuf_pkt_len(mbuf) > MAX_PAYLOAD_SIZE)) {
DPDKSegmentedMbufTooLargeWarning(rte_pktmbuf_pkt_len(mbuf));
StatsCounterIncr(&ptv->tv->stats, ptv->capture_dpdk_segmented_too_large);
return false;
}
if (ptv->copy_mode != DPDK_COPY_MODE_NONE) {
if (!ptv->tx_segmented_mbufs) {
DPDKSegmentedMbufTxDisabledWarning();
StatsCounterIncr(&ptv->tv->stats, ptv->capture_dpdk_segmented_drops);
return false;
}
if (mbuf->nb_segs > ptv->out_tx_seg_max) {
DPDKSegmentedMbufTooManySegmentsWarning(mbuf->nb_segs, ptv->out_tx_seg_max);
StatsCounterIncr(&ptv->tv->stats, ptv->capture_dpdk_segmented_drops);
return false;
}
}
return true;
}
/**
* \brief Copy the data of a segmented (chained) mbuf into the packet's buffer
*
* \param p Pointer to the Packet structure
* \param mbuf Pointer to the DPDK mbuf
* \return 0 on success, -1 on failure
*/
static inline int DPDKSegmentedMbufCopy(Packet *p, struct rte_mbuf *mbuf)
{
uint32_t offset = 0;
for (struct rte_mbuf *seg = mbuf; seg != NULL; seg = seg->next) {
uint32_t seg_len = rte_pktmbuf_data_len(seg);
if (PacketCopyDataOffset(p, offset, rte_pktmbuf_mtod(seg, uint8_t *), seg_len) != 0) {
return -1;
}
offset += seg_len;
}
SET_PKT_LEN(p, offset);
return 0;
}
/**
* \brief Set the data of a packet received in a segmented (chained) mbuf
*
* \param ptv Pointer to the thread variables
* \param p Pointer to the Packet structure
* \return true on success, false if the packet must be dropped
*/
static inline bool DPDKSegmentedMbufSetData(DPDKThreadVars *ptv, Packet *p)
{
struct rte_mbuf *mbuf = p->dpdk_v.mbuf;
if (!DPDKSegmentedMbufAccept(ptv, mbuf)) {
return false;
}
if (rte_pktmbuf_is_contiguous(mbuf)) {
// linearized
PacketSetData(p, rte_pktmbuf_mtod(mbuf, uint8_t *), rte_pktmbuf_pkt_len(mbuf));
} else if (DPDKSegmentedMbufCopy(p, mbuf) != 0) {
DPDKSegmentedMbufCopyFailedWarning();
StatsCounterIncr(&ptv->tv->stats, ptv->capture_dpdk_segmented_drops);
return false;
}
return true;
}
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);
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(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;
}
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)) {
continue;
}
ptv->pkts += (uint64_t)nb_rx;
for (uint16_t i = 0; i < nb_rx; i++) {
Packet *p = PacketInitFromMbuf(ptv, ptv->received_mbufs[i]);
if (p == NULL) {
rte_pktmbuf_free(ptv->received_mbufs[i]);
continue;
}
if (likely(rte_pktmbuf_is_contiguous(p->dpdk_v.mbuf))) {
PacketSetData(p, rte_pktmbuf_mtod(p->dpdk_v.mbuf, uint8_t *),
rte_pktmbuf_pkt_len(p->dpdk_v.mbuf));
} else if (!DPDKSegmentedMbufSetData(ptv, p)) {
// drop the uninspected packet, freeing the mbuf also keeps TAP mode, which
// ignores verdicts, from forwarding it
PacketDrop(p, ACTION_DROP, PKT_DROP_REASON_CAPTURE_ERROR);
rte_pktmbuf_free(p->dpdk_v.mbuf);
p->dpdk_v.mbuf = NULL;
TmqhOutputPacketpool(ptv->tv, p);
continue;
}
if (TmThreadsSlotProcessPkt(ptv->tv, ptv->slot, p) != TM_ECODE_OK) {
TmqhOutputPacketpool(ptv->tv, p);
DPDKFreeMbufArray(ptv->received_mbufs, nb_rx - i - 1, i + 1);
SCReturnInt(EXIT_FAILURE);
}
}
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->capture_dpdk_segmented_drops =
StatsRegisterCounter("capture.dpdk.segmented_drops", &ptv->tv->stats);
ptv->capture_dpdk_segmented_too_large =
StatsRegisterCounter("capture.dpdk.segmented_too_large", &ptv->tv->stats);
ptv->copy_mode = dpdk_config->copy_mode;
ptv->checksum_mode = dpdk_config->checksum_mode;
ptv->tx_segmented_mbufs = dpdk_config->tx_segmented_mbufs;
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->out_tx_seg_max = dpdk_config->out_tx_seg_max;
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;
// 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];
#if RTE_VERSION >= RTE_VERSION_NUM(20, 11, 0, 0)
#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
#else
snprintf(link_status_str, sizeof(link_status_str),
"Link Up, speed %u Mbps, %s", // 22 chars + 10 for digits + 11 for duplex
link.link_speed,
(link.link_duplex == ETH_LINK_FULL_DUPLEX) ? "full-duplex" : "half-duplex");
#endif
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)
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;
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 */
/**
* @}
*/