简介

dpdk编程之从给定的port/queue tuple捕获流量。

rte_lcore_id

rte_lcore_id()返回当前运行的lcore id。

lcore指的是EAL linux线程。

EAL创建和管理eal线程通过remote_launch来实现任务分配。

在每个EAL线程中，有一个称为_lcore_id TLS是线程的独一无二的id。

通常EAL线程绑定物理cpu，lcore_id通常等于cpu?id。

当使用多线程时，绑定不再在线程和指定物理cpu之间总是1:1，EAL线程设为对cpu集的亲和性，而lcore id不再和cpu?id一样。所以dpdk 提供了EAL选项-lcores，设置lcore的cpu亲和性，指定lcore id或组对EAL线程设置cpu set。

EAL为环境抽象层，DPDK运行时对linux操作系统进行初始化，并为用户态 DPDK App提供了一个通用接口，隐藏封装了与底层交互能力。做了一些实dpdk运行初始化工作，大页内存分配、numa亲和性、cpu绑定、内存/buffer/ring分配等，并通过 UIO 或 VFIO 技术将 PCI/PCIe 设备地址映射到用户空间。

rte_eth_rx_burst

nb_rx = rte_eth_rx_burst(arg0，arg1，arg2，arg3)从物理网口批量收包函数。

最大收包数量由args3决定。

arg0为port id，arg1为queue id。

arg2为数据包buffers数组类型为struct rte_mbuf *bufs[arg3];。

返回值nb_rx为实际接收的数据包数量，有效提供给bufs数组的指向rte_mbuf结构的指针数量。

rte_ring_enqueue_burst

nb_rx_enqueued = rte_ring_enqueue_burst(arg0，arg1，arg2)在ring上入队packet bufs。

根据ring创建时指定的默认行为，调用多生产者or单生产者。通常rte_ring支持多生产者入队和多消费者出队并且是非抢占的。

arg0为ring。

arg2为packet bufs。

arg3为nb_rx。

返回值nb_rx_enqueued为入队的实际数据包数

丢包统计

nb_rx与nb_rx_enqueued差值就是丢失数据包个数。

通过rte_pktmbuf_free把不能入队的数据包结构体free掉。

demo code

#include <stdbool.h>
#include <signal.h>
#include <string.h>

#include <dpdk/rte_log.h>
#include <dpdk/rte_lcore.h>
#include <dpdk/rte_ethdev.h>
#include <dpdk/rte_ether.h>
#include <dpdk/rte_ethdev.h>
#include <dpdk/rte_ip.h>
#include <arpa/inet.h>
#include "core_capture.h"


// 从给定的port/queue tuple捕获流量
int capture_core(const struct core_capture_config * config) {
  struct rte_mbuf *bufs[DPDKCAP_CAPTURE_BURST_SIZE];
  uint16_t nb_rx;
  int nb_rx_enqueued;

  RTE_LOG(INFO, DPDKCAP, "Core %u is capturing packets for port %u\n",
      rte_lcore_id(), config->port);

  /* Init stats */
  *(config->stats) = (struct core_capture_stats) {
    .core_id=rte_lcore_id(),
    .packets = 0,
    .missed_packets = 0,
  };

  /* Run until the application is quit or killed. */
  for (;;) {
    /* Stop condition */
    if (unlikely(*(config->stop_condition))) {
      break;
    }

    /* Retrieve packets and put them into the ring */
    nb_rx = rte_eth_rx_burst(config->port, config->queue,
        bufs, DPDKCAP_CAPTURE_BURST_SIZE);
    if (likely(nb_rx > 0)) {
      nb_rx_enqueued = rte_ring_enqueue_burst(config->ring, (void*) bufs,
          nb_rx);

      /* Update stats */
      config->stats->packets+=nb_rx_enqueued;
      config->stats->missed_packets+=nb_rx-nb_rx_enqueued;

      /* Free whatever we can't put in the write ring */
      for (; nb_rx_enqueued < nb_rx; nb_rx_enqueued++) {
        rte_pktmbuf_free(bufs[nb_rx_enqueued]);
      }
    }
  }

  RTE_LOG(INFO, DPDKCAP, "Closed capture core %d (port %d)\n",
      rte_lcore_id(), config->port);

  return 0;
}