80 FatalError(
"Error creating thread %s: you do not have "
81 "support for DPDK enabled, on Linux host please recompile "
99 #define MIN_ZERO_POLL_COUNT 10U
100 #define MIN_ZERO_POLL_COUNT_TO_SLEEP 10U
101 #define MINIMUM_SLEEP_TIME_US 1U
102 #define STANDARD_SLEEP_TIME_US 100U
103 #define MAX_EPOLL_TIMEOUT_MS 500U
104 static rte_spinlock_t intr_lock[RTE_MAX_ETHPORTS];
109 typedef struct DPDKThreadVars_ {
118 uint16_t capture_dpdk_packets;
119 uint16_t capture_dpdk_rx_errs;
120 uint16_t capture_dpdk_imissed;
121 uint16_t capture_dpdk_rx_no_mbufs;
122 uint16_t capture_dpdk_ierrors;
123 uint16_t capture_dpdk_tx_errs;
128 uint16_t out_port_id;
136 int32_t port_socket_id;
137 struct rte_mbuf *received_mbufs[BURST_SIZE];
142 static void ReceiveDPDKThreadExitStats(
ThreadVars *,
void *);
150 static void DPDKFreeMbufArray(
struct rte_mbuf **mbuf_array, uint16_t mbuf_cnt, uint16_t
offset);
151 static bool InterruptsRXEnable(uint16_t port_id, uint16_t queue_id)
153 uint32_t event_data = port_id << UINT16_WIDTH | queue_id;
154 int32_t ret = rte_eth_dev_rx_intr_ctl_q(port_id, queue_id, RTE_EPOLL_PER_THREAD,
155 RTE_INTR_EVENT_ADD, (
void *)((uintptr_t)event_data));
158 SCLogError(
"%s-Q%d: failed to enable interrupt mode: %s", DPDKGetPortNameByPortID(port_id),
159 queue_id, rte_strerror(-ret));
165 static inline uint32_t InterruptsSleepHeuristic(uint32_t no_pkt_polls_count)
167 if (no_pkt_polls_count < MIN_ZERO_POLL_COUNT_TO_SLEEP)
168 return MINIMUM_SLEEP_TIME_US;
170 return STANDARD_SLEEP_TIME_US;
173 static inline void InterruptsTurnOnOff(uint16_t port_id, uint16_t queue_id,
bool on)
175 rte_spinlock_lock(&(intr_lock[port_id]));
178 rte_eth_dev_rx_intr_enable(port_id, queue_id);
180 rte_eth_dev_rx_intr_disable(port_id, queue_id);
182 rte_spinlock_unlock(&(intr_lock[port_id]));
185 static inline void DPDKFreeMbufArray(
186 struct rte_mbuf **mbuf_array, uint16_t mbuf_cnt, uint16_t
offset)
188 for (
int i =
offset; i < mbuf_cnt; i++) {
189 rte_pktmbuf_free(mbuf_array[i]);
193 static void DevicePostStartPMDSpecificActions(DPDKThreadVars *ptv,
const char *driver_name)
195 if (strcmp(driver_name,
"net_bonding") == 0)
196 driver_name = BondingDeviceDriverGet(ptv->port_id);
197 if (strcmp(driver_name,
"net_i40e") == 0)
198 i40eDeviceSetRSS(ptv->port_id, ptv->threads, ptv->livedev->dev);
199 else if (strcmp(driver_name,
"net_ixgbe") == 0)
200 ixgbeDeviceSetRSS(ptv->port_id, ptv->threads, ptv->livedev->dev);
201 else if (strcmp(driver_name,
"net_ice") == 0)
202 iceDeviceSetRSS(ptv->port_id, ptv->threads, ptv->livedev->dev);
203 else if (strcmp(driver_name,
"mlx5_pci") == 0)
204 mlx5DeviceSetRSS(ptv->port_id, ptv->threads, ptv->livedev->dev);
207 static void DevicePreClosePMDSpecificActions(DPDKThreadVars *ptv,
const char *driver_name)
209 if (strcmp(driver_name,
"net_bonding") == 0) {
210 driver_name = BondingDeviceDriverGet(ptv->port_id);
214 #
if RTE_VERSION > RTE_VERSION_NUM(20, 0, 0, 0)
215 strcmp(driver_name,
"net_i40e") == 0 ||
217 strcmp(driver_name,
"net_ixgbe") == 0 || strcmp(driver_name,
"net_ice") == 0 ||
218 strcmp(driver_name,
"mlx5_pci") == 0) {
220 struct rte_flow_error flush_error = { 0 };
221 int32_t retval = rte_flow_flush(ptv->port_id, &flush_error);
223 SCLogError(
"%s: unable to flush rte_flow rules: %s Flush error msg: %s",
224 ptv->livedev->dev, rte_strerror(-retval), flush_error.message);
233 static int GetNumaNode(
void)
238 #if defined(__linux__)
239 cpu = sched_getcpu();
240 node = numa_node_of_cpu(cpu);
242 SCLogWarning(
"NUMA node retrieval is not supported on this OS.");
280 static inline void DPDKDumpCounters(DPDKThreadVars *ptv)
285 if (ptv->queue_id == 0) {
286 struct rte_eth_stats eth_stats;
287 int retval = rte_eth_stats_get(ptv->port_id, ð_stats);
289 SCLogError(
"%s: failed to get stats: %s", ptv->livedev->dev, rte_strerror(-retval));
294 ptv->pkts + eth_stats.imissed + eth_stats.ierrors + eth_stats.rx_nombuf);
296 eth_stats.ipackets + eth_stats.imissed + eth_stats.ierrors + eth_stats.rx_nombuf);
298 eth_stats.imissed + eth_stats.ierrors + eth_stats.rx_nombuf);
299 StatsSetUI64(ptv->tv, ptv->capture_dpdk_imissed, eth_stats.imissed);
300 StatsSetUI64(ptv->tv, ptv->capture_dpdk_rx_no_mbufs, eth_stats.rx_nombuf);
301 StatsSetUI64(ptv->tv, ptv->capture_dpdk_ierrors, eth_stats.ierrors);
302 StatsSetUI64(ptv->tv, ptv->capture_dpdk_tx_errs, eth_stats.oerrors);
304 ptv->livedev->drop, eth_stats.imissed + eth_stats.ierrors + eth_stats.rx_nombuf);
306 StatsSetUI64(ptv->tv, ptv->capture_dpdk_packets, ptv->pkts);
310 static void DPDKReleasePacket(
Packet *p)
319 #
if defined(RTE_LIBRTE_I40E_PMD) || defined(RTE_LIBRTE_IXGBE_PMD) || defined(RTE_LIBRTE_ICE_PMD)
320 && !(PacketIsICMPv6(p) && PacketGetICMPv6(p)->
type == 143)
325 rte_eth_tx_burst(p->dpdk_v.out_port_id, p->dpdk_v.out_queue_id, &p->dpdk_v.mbuf, 1);
332 retval = rte_eth_tx_burst(
333 p->dpdk_v.out_port_id, p->dpdk_v.out_queue_id, &p->dpdk_v.mbuf, 1);
335 SCLogDebug(
"Unable to transmit the packet on port %u queue %u",
336 p->dpdk_v.out_port_id, p->dpdk_v.out_queue_id);
337 rte_pktmbuf_free(p->dpdk_v.mbuf);
338 p->dpdk_v.mbuf = NULL;
342 rte_pktmbuf_free(p->dpdk_v.mbuf);
343 p->dpdk_v.mbuf = NULL;
357 rte_eth_stats_reset(ptv->port_id);
358 rte_eth_xstats_reset(ptv->port_id);
360 if (ptv->intr_enabled && !InterruptsRXEnable(ptv->port_id, ptv->queue_id))
366 static inline void LoopHandleTimeoutOnIdle(
ThreadVars *
tv)
368 static thread_local uint64_t last_timeout_msec = 0;
371 if (msecs > last_timeout_msec + 100) {
372 TmThreadsCaptureHandleTimeout(
tv, NULL);
373 last_timeout_msec = msecs;
381 static inline bool RXPacketCountHeuristic(
ThreadVars *
tv, DPDKThreadVars *ptv, uint16_t nb_rx)
383 static thread_local uint32_t zero_pkt_polls_cnt = 0;
386 zero_pkt_polls_cnt = 0;
390 LoopHandleTimeoutOnIdle(
tv);
391 if (!ptv->intr_enabled)
394 zero_pkt_polls_cnt++;
395 if (zero_pkt_polls_cnt <= MIN_ZERO_POLL_COUNT)
398 uint32_t pwd_idle_hint = InterruptsSleepHeuristic(zero_pkt_polls_cnt);
399 if (pwd_idle_hint < STANDARD_SLEEP_TIME_US) {
400 rte_delay_us(pwd_idle_hint);
402 InterruptsTurnOnOff(ptv->port_id, ptv->queue_id,
true);
403 struct rte_epoll_event event;
404 rte_epoll_wait(RTE_EPOLL_PER_THREAD, &event, 1, MAX_EPOLL_TIMEOUT_MS);
405 InterruptsTurnOnOff(ptv->port_id, ptv->queue_id,
false);
416 static inline Packet *PacketInitFromMbuf(DPDKThreadVars *ptv,
struct rte_mbuf *mbuf)
429 p->dpdk_v.mbuf = mbuf;
431 p->dpdk_v.copy_mode = ptv->copy_mode;
432 p->dpdk_v.out_port_id = ptv->out_port_id;
433 p->dpdk_v.out_queue_id = ptv->queue_id;
439 uint64_t ol_flags = p->dpdk_v.mbuf->ol_flags;
440 if ((ol_flags & RTE_MBUF_F_RX_IP_CKSUM_MASK) == RTE_MBUF_F_RX_IP_CKSUM_GOOD &&
441 (ol_flags & RTE_MBUF_F_RX_L4_CKSUM_MASK) == RTE_MBUF_F_RX_L4_CKSUM_GOOD) {
442 SCLogDebug(
"HW detected GOOD IP and L4 chsum, ignoring validation");
445 if ((ol_flags & RTE_MBUF_F_RX_IP_CKSUM_MASK) == RTE_MBUF_F_RX_IP_CKSUM_BAD) {
451 if ((ol_flags & RTE_MBUF_F_RX_L4_CKSUM_MASK) == RTE_MBUF_F_RX_L4_CKSUM_BAD) {
462 static inline void DPDKSegmentedMbufWarning(
struct rte_mbuf *mbuf)
464 static thread_local
bool segmented_mbufs_warned =
false;
465 if (!segmented_mbufs_warned && !rte_pktmbuf_is_contiguous(mbuf)) {
466 char warn_s[] =
"Segmented mbufs detected! Redmine Ticket #6012 "
467 "Check your configuration or report the issue";
468 enum rte_proc_type_t eal_t = rte_eal_process_type();
469 if (eal_t == RTE_PROC_SECONDARY) {
471 "try to increase mbuf size in your primary application",
473 }
else if (eal_t == RTE_PROC_PRIMARY) {
475 "try to increase MTU in your suricata.yaml",
479 segmented_mbufs_warned =
true;
483 static void HandleShutdown(DPDKThreadVars *ptv)
487 while (
SC_ATOMIC_GET(ptv->workers_sync->worker_checked_in) < ptv->workers_sync->worker_cnt) {
490 if (ptv->queue_id == 0) {
497 rte_eth_dev_stop(ptv->out_port_id);
500 rte_eth_dev_stop(ptv->port_id);
503 DPDKDumpCounters(ptv);
506 static void PeriodicDPDKDumpCounters(DPDKThreadVars *ptv)
508 static thread_local
SCTime_t last_dump = { 0 };
511 if (current_time.
secs != last_dump.secs) {
512 DPDKDumpCounters(ptv);
513 last_dump = current_time;
523 DPDKThreadVars *ptv = (DPDKThreadVars *)data;
524 ptv->slot = ((
TmSlot *)slot)->slot_next;
525 TmEcode ret = ReceiveDPDKLoopInit(
tv, ptv);
536 rte_eth_rx_burst(ptv->port_id, ptv->queue_id, ptv->received_mbufs, BURST_SIZE);
537 if (RXPacketCountHeuristic(
tv, ptv, nb_rx)) {
541 ptv->pkts += (uint64_t)nb_rx;
542 for (uint16_t i = 0; i < nb_rx; i++) {
543 Packet *p = PacketInitFromMbuf(ptv, ptv->received_mbufs[i]);
545 rte_pktmbuf_free(ptv->received_mbufs[i]);
548 DPDKSegmentedMbufWarning(ptv->received_mbufs[i]);
549 PacketSetData(p, rte_pktmbuf_mtod(p->dpdk_v.mbuf, uint8_t *),
550 rte_pktmbuf_pkt_len(p->dpdk_v.mbuf));
551 if (TmThreadsSlotProcessPkt(ptv->tv, ptv->slot, p) !=
TM_ECODE_OK) {
553 DPDKFreeMbufArray(ptv->received_mbufs, nb_rx - i - 1, i + 1);
558 PeriodicDPDKDumpCounters(ptv);
576 int retval, thread_numa;
577 DPDKThreadVars *ptv = NULL;
580 if (initdata == NULL) {
581 SCLogError(
"DPDK configuration is NULL in thread initialization");
585 ptv =
SCCalloc(1,
sizeof(DPDKThreadVars));
603 ptv->copy_mode = dpdk_config->copy_mode;
604 ptv->checksum_mode = dpdk_config->checksum_mode;
606 ptv->threads = dpdk_config->threads;
607 ptv->intr_enabled = (dpdk_config->flags &
DPDK_IRQ_MODE) ?
true :
false;
608 ptv->port_id = dpdk_config->port_id;
609 ptv->out_port_id = dpdk_config->out_port_id;
610 ptv->port_socket_id = dpdk_config->socket_id;
612 thread_numa = GetNumaNode();
613 if (thread_numa >= 0 && ptv->port_socket_id != SOCKET_ID_ANY &&
614 thread_numa != ptv->port_socket_id) {
616 SCLogPerf(
"%s: NIC is on NUMA %d, thread on NUMA %d", dpdk_config->iface,
617 ptv->port_socket_id, thread_numa);
620 ptv->workers_sync = dpdk_config->workers_sync;
622 ptv->queue_id = queue_id;
625 if (queue_id == dpdk_config->threads - 1) {
626 retval = rte_eth_dev_start(ptv->port_id);
628 SCLogError(
"%s: error (%s) during device startup", dpdk_config->iface,
629 rte_strerror(-retval));
633 struct rte_eth_dev_info dev_info;
634 retval = rte_eth_dev_info_get(ptv->port_id, &dev_info);
636 SCLogError(
"%s: error (%s) when getting device info", dpdk_config->iface,
637 rte_strerror(-retval));
641 uint32_t timeout = dpdk_config->linkup_timeout * 10;
642 while (timeout > 0) {
643 struct rte_eth_link link = { 0 };
644 retval = rte_eth_link_get_nowait(ptv->port_id, &link);
646 if (retval == -ENOTSUP) {
647 SCLogInfo(
"%s: link status not supported, skipping", dpdk_config->iface);
649 SCLogInfo(
"%s: error (%s) when getting link status, skipping",
650 dpdk_config->iface, rte_strerror(-retval));
654 if (link.link_status) {
655 char link_status_str[RTE_ETH_LINK_MAX_STR_LEN];
656 #if RTE_VERSION >= RTE_VERSION_NUM(20, 11, 0, 0)
657 #pragma GCC diagnostic push
658 #pragma GCC diagnostic ignored "-Wdeprecated-declarations"
659 rte_eth_link_to_str(link_status_str,
sizeof(link_status_str), &link);
660 #pragma GCC diagnostic pop
662 snprintf(link_status_str,
sizeof(link_status_str),
663 "Link Up, speed %u Mbps, %s",
665 (link.link_duplex == ETH_LINK_FULL_DUPLEX) ?
"full-duplex" :
"half-duplex");
668 SCLogInfo(
"%s: %s", dpdk_config->iface, link_status_str);
676 if (dpdk_config->linkup_timeout && timeout == 0) {
677 SCLogWarning(
"%s: link is down, trying to continue anyway", dpdk_config->iface);
681 DevicePostStartPMDSpecificActions(ptv, dev_info.driver_name);
683 uint16_t inconsistent_numa_cnt =
SC_ATOMIC_GET(dpdk_config->inconsistent_numa_cnt);
684 if (inconsistent_numa_cnt > 0 && ptv->port_socket_id != SOCKET_ID_ANY) {
685 SCLogWarning(
"%s: NIC is on NUMA %d, %u threads on different NUMA node(s)",
686 dpdk_config->iface, ptv->port_socket_id, inconsistent_numa_cnt);
687 }
else if (ptv->port_socket_id == SOCKET_ID_ANY && rte_socket_count() > 1) {
689 "%s: unable to determine NIC's NUMA node, degraded performance can be expected",
692 if (ptv->intr_enabled) {
693 rte_spinlock_init(&intr_lock[ptv->port_id]);
698 dpdk_config->DerefFunc(dpdk_config);
702 if (dpdk_config != NULL)
703 dpdk_config->DerefFunc(dpdk_config);
709 static void PrintDPDKPortXstats(uint32_t port_id,
const char *port_name)
711 struct rte_eth_xstat *xstats;
712 struct rte_eth_xstat_name *xstats_names;
714 int32_t
len = rte_eth_xstats_get(port_id, NULL, 0);
716 FatalError(
"Error (%s) getting count of rte_eth_xstats failed on port %s",
717 rte_strerror(-
len), port_name);
721 FatalError(
"Failed to allocate memory for the rte_eth_xstat structure");
723 int32_t ret = rte_eth_xstats_get(port_id, xstats,
len);
724 if (ret < 0 || ret >
len) {
726 FatalError(
"Error (%s) getting rte_eth_xstats failed on port %s", rte_strerror(-ret),
729 xstats_names =
SCCalloc(
len,
sizeof(*xstats_names));
730 if (xstats_names == NULL) {
732 FatalError(
"Failed to allocate memory for the rte_eth_xstat_name array");
734 ret = rte_eth_xstats_get_names(port_id, xstats_names,
len);
735 if (ret < 0 || ret >
len) {
738 FatalError(
"Error (%s) getting names of rte_eth_xstats failed on port %s",
739 rte_strerror(-ret), port_name);
741 for (int32_t i = 0; i <
len; i++) {
742 if (xstats[i].value > 0)
743 SCLogPerf(
"Port %u (%s) - %s: %" PRIu64, port_id, port_name, xstats_names[i].
name,
756 static void ReceiveDPDKThreadExitStats(
ThreadVars *
tv,
void *data)
760 DPDKThreadVars *ptv = (DPDKThreadVars *)data;
762 if (ptv->queue_id == 0) {
763 struct rte_eth_stats eth_stats;
764 PrintDPDKPortXstats(ptv->port_id, ptv->livedev->dev);
765 retval = rte_eth_stats_get(ptv->port_id, ð_stats);
767 SCLogError(
"%s: failed to get stats (%s)", ptv->livedev->dev, strerror(-retval));
770 SCLogPerf(
"%s: total RX stats: packets %" PRIu64
" bytes: %" PRIu64
" missed: %" PRIu64
771 " errors: %" PRIu64
" nombufs: %" PRIu64,
772 ptv->livedev->dev, eth_stats.ipackets, eth_stats.ibytes, eth_stats.imissed,
773 eth_stats.ierrors, eth_stats.rx_nombuf);
775 SCLogPerf(
"%s: total TX stats: packets %" PRIu64
" bytes: %" PRIu64
" errors: %" PRIu64,
776 ptv->livedev->dev, eth_stats.opackets, eth_stats.obytes, eth_stats.oerrors);
779 DPDKDumpCounters(ptv);
791 DPDKThreadVars *ptv = (DPDKThreadVars *)data;
793 if (ptv->queue_id == 0) {
794 struct rte_eth_dev_info dev_info;
795 int retval = rte_eth_dev_info_get(ptv->port_id, &dev_info);
797 SCLogError(
"%s: error (%s) when getting device info", ptv->livedev->dev,
798 rte_strerror(-retval));
802 DevicePreClosePMDSpecificActions(ptv, dev_info.driver_name);
804 if (ptv->workers_sync) {
805 SCFree(ptv->workers_sync);