第19周：PMD 驱动与收发包

字数统计: 2.2k字   |   阅读时长≈ 11分

第19周：PMD 驱动与收发包

目标：理解 PMD（Poll Mode Driver）工作原理，掌握 DPDK 收发包核心 API。

1. PMD 驱动原理

1.1 PMD vs 内核驱动

1
2
3
4
5
6
7
8
9
10
11
内核驱动（传统）：
  中断驱动 — 每个包触发中断
  内核协议栈处理 — TCP/IP 栈
  软中断（ksoftirqd）— 批量处理
  sk_buff 管理 — 开销大

PMD 驱动（DPDK）：
  轮询模式 — 主动检查 descriptor
  无中断 — 独占 CPU
  用户态直接访问 — 无内核切换
  mbuf 管理 — 高效批量

1.2 Descriptor Ring

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
RX Descriptor Ring（接收描述符环）：

  ┌─────┬─────┬─────┬─────┬─────┬─────┐
  │  0  │  1  │  2  │  3  │ ... │ N-1 │  ← 硬件写（收到包）
  └─────┴─────┴─────┴─────┴─────┴─────┘
    ↑                            ↑
  rx_ring_head               rx_ring_tail
  （用户读）                  （硬件写）

每个 descriptor 包含：
  - 数据缓冲的物理地址（DMA）
  - 数据长度
  - 状态标志（已完成/错误等）

操作流程：
  1. 用户分配 mbuf，获取物理地址
  2. 将物理地址写入 descriptor
  3. 硬件收到包 → DMA 到 mbuf → 更新 descriptor 状态
  4. 用户轮询检查 descriptor 状态
  5. 取出已完成的 descriptor，处理包
  6. 将新 mbuf 放回 descriptor 供下次使用

1.3 收发流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
RX 流程（收包）：
  1. 分配一批 mbuf（或从 pool 取）
  2. 将 mbuf 的物理地址填入 RX descriptor
  3. 通知网卡有新 buffer（通过 head 指针）
  4. 轮询检查 descriptor 状态
  5. 取出已完成收包的 descriptor 对应 mbuf
  6. 处理包（查表/转发/释放）
  7. 将新 mbuf 重新填回 descriptor

TX 流程（发包）：
  1. 准备 mbuf（包含要发送的数据）
  2. 将 mbuf 指针填入 TX descriptor
  3. 通知网卡发送（通过 tail 指针）
  4. 轮询检查发送完成的 descriptor
  5. 释放已发送完成的 mbuf 回 mempool

2. 端口配置

2.1 初始化流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
#include <rte_ethdev.h>

#define RX_RINGS    1      // RX 队列数
#define TX_RINGS    1      // TX 队列数
#define RX_DESC     1024   // RX descriptor 数量
#define TX_DESC     1024   // TX descriptor 数量
#define NB_MBUF     8192   // mbuf 数量

int port_init(uint16_t port) {
    struct rte_eth_conf port_conf;
    struct rte_eth_dev_info dev_info;
    int ret;
    uint16_t q;

    // 1. 获取设备信息
    rte_eth_dev_info_get(port, &dev_info);
    printf("Port %u: %s, max_rx_queues=%u, max_tx_queues=%u\n",
           port, dev_info.driver_name,
           dev_info.max_rx_queues, dev_info.max_tx_queues);

    // 2. 配置端口
    memset(&port_conf, 0, sizeof(port_conf));
    port_conf.rxmode.mq_mode = ETH_MQ_RX_NONE;    // 无多队列
    port_conf.rxmode.split_hdr_size = 0;
    port_conf.rxmode.offloads = 0;
    port_conf.txmode.mq_mode = ETH_MQ_TX_NONE;
    port_conf.txmode.offloads = 0;

    // 3. 配置设备（RX/TX 队列数）
    ret = rte_eth_dev_configure(port, RX_RINGS, TX_RINGS, &port_conf);
    if (ret != 0) {
        printf("rte_eth_dev_configure failed: %d\n", ret);
        return ret;
    }

    // 4. 设置 RX 队列
    ret = rte_eth_rx_queue_setup(port, 0,    // port, queue_id
                                  RX_DESC,      // descriptor 数
                                  rte_eth_dev_socket_id(port),  // NUMA
                                  NULL,          // 配置（NULL=默认）
                                  mbuf_pool);    // mbuf pool
    if (ret != 0) {
        printf("rte_eth_rx_queue_setup failed: %d\n", ret);
        return ret;
    }

    // 5. 设置 TX 队列
    ret = rte_eth_tx_queue_setup(port, 0,    // port, queue_id
                                  TX_DESC,      // descriptor 数
                                  rte_eth_dev_socket_id(port),  // NUMA
                                  NULL);
    if (ret != 0) {
        printf("rte_eth_tx_queue_setup failed: %d\n", ret);
        return ret;
    }

    // 6. 启动设备
    ret = rte_eth_dev_start(port);
    if (ret != 0) {
        printf("rte_eth_dev_start failed: %d\n", ret);
        return ret;
    }

    // 7. 启用混杂模式（接收所有包）
    rte_eth_promiscuous_enable(port);

    // 8. 打印 MAC 地址
    struct rte_ether_addr mac;
    rte_eth_macaddr_get(port, &mac);
    printf("Port %u MAC: %02X:%02X:%02X:%02X:%02X:%02X\n",
           port, mac.addr_bytes[0], mac.addr_bytes[1], mac.addr_bytes[2],
           mac.addr_bytes[3], mac.addr_bytes[4], mac.addr_bytes[5]);

    return 0;
}

2.2 设备信息查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
struct rte_eth_dev_info {
    const char *driver_name;
    uint32_t min_rx_buf_size;      // 最小接收缓冲区
    uint32_t max_rx_pktlen;        // 最大包长度
    uint16_t max_rx_queues;        // 最大 RX 队列数
    uint16_t max_tx_queues;        // 最大 TX 队列数
    uint32_t max_mac_addrs;        // MAC 地址数量
    uint32_t max_vfs;               // 最大 VF 数（SR-IOV）
    uint32_t max_hash_mac_addrs;
    uint32_t max_vmdq_pools;
    uint32_t rx_offload_capa;      // 支持的 RX offload
    uint32_t tx_offload_capa;      // 支持的 TX offload
    uint32_t rx_queue_offload_capa;
    uint32_t tx_queue_offload_capa;
    uint32_t reta_size;            // RSS 重定向表大小
    uint16_t hash_key_size;        // RSS hash 密钥大小
    // ...
};

// 查询
rte_eth_dev_info_get(port, &dev_info);

3. 收发包 API

3.1 rte_eth_rx_burst

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
#include <rte_ethdev.h>

// 批量接收数据包
// 返回实际收到的包数（0 ~ nb_pkts）
uint16_t rte_eth_rx_burst(uint16_t port_id, uint16_t queue_id,
                          struct rte_mbuf **rx_pkts, uint16_t nb_pkts);

// 使用示例
struct rte_mbuf *pkts[32];
uint16_t nb_rx;

nb_rx = rte_eth_rx_burst(port_id, 0, pkts, 32);

if (nb_rx > 0) {
    printf("Received %d packets\n", nb_rx);

    for (int i = 0; i < nb_rx; i++) {
        struct rte_mbuf *mbuf = pkts[i];

        // 获取包信息
        uint16_t pkt_len = rte_pktmbuf_pkt_len(mbuf);
        uint16_t data_len = rte_pktmbuf_data_len(mbuf);
        uint8_t *data = rte_pktmbuf_mtod(mbuf, uint8_t *);

        printf("Packet %d: len=%d\n", i, pkt_len);

        // TODO: 处理包
    }

    // 处理完后释放所有 mbuf
    rte_pktmbuf_free_bulk(pkts, nb_rx);
}

3.2 rte_eth_tx_burst

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
// 批量发送数据包
// 返回实际发送的包数（0 ~ nb_pkts）
uint16_t rte_eth_tx_burst(uint16_t port_id, uint16_t queue_id,
                          struct rte_mbuf **tx_pkts, uint16_t nb_pkts);

// 使用示例
struct rte_mbuf *pkts_to_send[32];

// 准备要发送的 mbuf（填充数据）
for (int i = 0; i < n; i++) {
    pkts_to_send[i] = rte_pktmbuf_alloc(pool);
    // ... 填充数据
}

// 发送
uint16_t nb_tx = rte_eth_tx_burst(port_id, 0, pkts_to_send, n);

// 处理未发送的
if (nb_tx < n) {
    for (int i = nb_tx; i < n; i++)
        rte_pktmbuf_free(pkts_to_send[i]);
}

3.3 完整收发循环

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
// simple_fwd.c — 最简单的转发
#include <rte_eal.h>
#include <rte_ethdev.h>
#include <rte_mbuf.h>

#define RX_RINGS  1
#define TX_RINGS  1
#define RX_DESC   1024
#define TX_DESC   1024
#define NB_MBUF   8192

static struct rte_mempool *mbuf_pool;

// 端口初始化
int port_init(uint16_t port) {
    // ... 如前所述
}

// 主转发循环
void main_loop(void) {
    struct rte_mbuf *pkts[32];
    uint16_t nb_rx;

    printf("Starting main loop on lcore %u\n", rte_lcore_id());

    while (1) {
        // 从 port 0 接收
        nb_rx = rte_eth_rx_burst(0, 0, pkts, 32);

        if (nb_rx == 0) continue;

        for (int i = 0; i < nb_rx; i++) {
            // 简单修改 MAC 地址（示范）
            struct rte_ether_hdr *eth;
            eth = rte_pktmbuf_mtod(pkts[i], struct rte_ether_hdr *);

            // 交换源/目的 MAC
            struct rte_ether_addr tmp = eth->src_addr;
            eth->src_addr = eth->dst_addr;
            eth->dst_addr = tmp;
        }

        // 发送到 port 1
        uint16_t nb_tx = rte_eth_tx_burst(1, 0, pkts, nb_rx);

        // 释放未发送的
        if (nb_tx < nb_rx)
            rte_pktmbuf_free_bulk(&pkts[nb_tx], nb_rx - nb_tx);
    }
}

int main(int argc, char **argv) {
    int ret;
    uint16_t port_id;

    // 初始化 EAL
    ret = rte_eal_init(argc, argv);
    if (ret < 0) return 1;
    argc -= ret; argv += ret;

    // 创建 mbuf pool
    mbuf_pool = rte_pktmbuf_pool_create("MBUF_POOL", NB_MBUF,
                                        256, 0,
                                        RTE_MBUF_DEFAULT_BUF_SIZE,
                                        SOCKET_ID_ANY);
    if (!mbuf_pool) return 1;

    // 初始化端口
    RTE_ETH_FOREACH_DEV(port_id) {
        port_init(port_id);
    }

    // 在每个 lcore 上运行主循环
    rte_eal_mp_remote_launch(main_loop, NULL, SKIP_MASTER);
    rte_eal_mp_wait_lcore();

    return 0;
}

4. 统计与监控

4.1 端口统计

1
2
3
4
5
6
7
8
9
10
11
12
13
14
struct rte_eth_stats stats;

// 获取端口统计
rte_eth_stats_get(port_id, &stats);

printf("Port %u stats:\n", port_id);
printf("  RX packets:  %lu\n", stats.ipackets);
printf("  TX packets:  %lu\n", stats.opackets);
printf("  RX bytes:    %lu\n", stats.ibytes);
printf("  TX bytes:    %lu\n", stats.obytes);
printf("  RX errors:   %lu\n", stats.ierrors);
printf("  TX errors:   %lu\n", stats.oerrors);
printf("  RX missed:   %lu\n", stats.imissed);  // 接收缓冲区满丢包
printf("  RX no mbuf:  %lu\n", stats.rx_nombuf);  // mbuf 不足丢包

4.2 统计清零

1
2
3
4
5
// 清零统计
rte_eth_stats_reset(port_id);

// 持续统计（回调方式）
rte_eth_stats_callback_register(port_id, stats_callback, NULL);

4.3 PPS 计算示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// pps_counter.c
struct port_stats {
    uint64_t prev_packets;
    uint64_t prev_ts;
    double pps;
};

void update_pps(uint16_t port, struct port_stats *s) {
    struct rte_eth_stats stats;
    uint64_t now = rte_rdtsc();

    rte_eth_stats_get(port, &stats);

    uint64_t delta_packets = stats.ipackets - s->prev_packets;
    uint64_t delta_ts = now - s->prev_ts;
    double secs = (double)delta_ts / rte_get_timer_hz();

    s->pps = delta_packets / secs;
    s->prev_packets = stats.ipackets;
    s->prev_ts = now;

    printf("Port %u: %.2f Mpps, %.2f Gbps\n",
           port, s->pps / 1e6,
           (s->pps * 64 * 8) / 1e9);  // 假设 64B 包
}

5. 简单的收发包统计程序

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
// pktgen_simple.c
#include <rte_eal.h>
#include <rte_ethdev.h>
#include <rte_mbuf.h>
#include <rte_cycles.h>

#define NB_MBUF   65536
#define BURST_SIZE 32
#define PORT_ID   0

static struct rte_mempool *mbuf_pool;
static uint64_t prev_tsc;

static void print_stats(void) {
    struct rte_eth_stats stats;
    uint64_t now = rte_rdtsc();
    double secs = (double)(now - prev_tsc) / rte_get_timer_hz();

    rte_eth_stats_get(PORT_ID, &stats);

    printf("\n=== Stats (%.2f seconds) ===\n", secs);
    printf("  RX packets:  %lu (%.2f Mpps)\n",
           stats.ipackets,
           (stats.ipackets / secs) / 1e6);
    printf("  TX packets:  %lu (%.2f Mpps)\n",
           stats.opackets,
           (stats.opackets / secs) / 1e6);
    printf("  RX bytes:    %lu (%.2f Gbps)\n",
           stats.ibytes,
           (stats.ibytes * 8 / secs) / 1e9);
    printf("  RX missed:   %lu\n", stats.imissed);
    printf("  RX no mbuf:  %lu\n", stats.rx_nombuf);

    prev_tsc = now;
}

int main(int argc, char **argv) {
    struct rte_mbuf *pkts[BURST_SIZE];
    uint16_t nb_rx, nb_tx;
    int ret;
    uint16_t port_id = PORT_ID;

    // 1. 初始化
    ret = rte_eal_init(argc, argv);
    if (ret < 0) return 1;

    printf("DPDK Packet Counter\n");
    printf("Lcores: %u\n", rte_lcore_count());

    // 2. 创建 mbuf pool
    mbuf_pool = rte_pktmbuf_pool_create("MBUF_POOL", NB_MBUF,
                                        256, 0,
                                        RTE_MBUF_DEFAULT_BUF_SIZE,
                                        SOCKET_ID_ANY);
    if (!mbuf_pool) return 1;

    // 3. 初始化端口
    int fwd_ports[] = {port_id, port_id}; // RX=TX 同一个端口（loopback）
    // 实际应该配置两个端口进行收发

    // 4. 统计循环
    prev_tsc = rte_rdtsc();

    while (1) {
        // 接收
        nb_rx = rte_eth_rx_burst(port_id, 0, pkts, BURST_SIZE);

        if (nb_rx > 0) {
            // 可以在这里分析包内容
            for (int i = 0; i < nb_rx; i++) {
                struct rte_mbuf *m = pkts[i];
                uint8_t *data = rte_pktmbuf_mtod(m, uint8_t *);
                uint16_t len = rte_pktmbuf_pkt_len(m);

                // 打印以太网类型
                struct rte_ether_hdr *eth;
                eth = rte_pktmbuf_mtod(m, struct rte_ether_hdr *);
                printf("RX: ETH_TYPE=0x%04X, len=%u\n",
                       rte_be_to_cpu_16(eth->ether_type), len);
            }

            // 释放包（不做转发）
            rte_pktmbuf_free_bulk(pkts, nb_rx);
        }

        // 每秒打印一次统计
        uint64_t now = rte_rdtsc();
        if ((now - prev_tsc) > rte_get_timer_hz()) {
            print_stats();
        }
    }

    return 0;
}

6. 编译

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# Makefile
RTE_SDK := /path/to/dpdk
RTE_TARGET := x86_64-native-linuxapp-gcc

CFLAGS := -O3 -I$(RTE_SDK)/$(RTE_TARGET)/include
LDFLAGS := -L$(RTE_SDK)/$(RTE_TARGET)/lib -ldpdk

TARGET := pktgen_simple

$(TARGET): $(TARGET).c
	gcc $(CFLAGS) -o $@ $< $(LDFLAGS) -lpthread -lm -ldl

clean:
	rm -f $(TARGET)
1
2
3
4
5
6
7
# Meson 方式（DPDK 22.x+）
# dpdk/build 目录中有构建好的环境
gcc -o pktgen_simple pktgen_simple.c \
    -I dpdk/build/include \
    -L dpdk/build/lib \
    -Wl,--whole-archive -ldpdk -Wl,--no-whole-archive \
    -lpthread -lm -ldl

7. 关键 API 速查

API 功能
rte_eal_init() 初始化 EAL
rte_lcore_count() 可用 lcore 数量
rte_get_master_lcore() 获取主 lcore ID
rte_rdtsc() 读取 TSC 时间戳
rte_get_timer_hz() TSC 频率（Hz）
rte_pktmbuf_pool_create() 创建 mbuf 池
rte_pktmbuf_alloc() 分配 mbuf
rte_pktmbuf_free() 释放 mbuf
rte_pktmbuf_mtod() mbuf → 数据指针
rte_pktmbuf_pkt_len() 获取包长度
rte_pktmbuf_data_len() 获取当前段长度
rte_pktmbuf_append() 扩展数据区
rte_eth_dev_configure() 配置端口
rte_eth_rx_queue_setup() 设置 RX 队列
rte_eth_tx_queue_setup() 设置 TX 队列
rte_eth_dev_start() 启动端口
rte_eth_dev_stop() 停止端口
rte_eth_rx_burst() 批量接收
rte_eth_tx_burst() 批量发送
rte_eth_stats_get() 获取端口统计
rte_eth_macaddr_get() 获取 MAC 地址
rte_ring_create() 创建无锁 ring
rte_ring_enqueue_bulk() 批量入队
rte_ring_dequeue_burst() 批量出队
rte_mempool_create() 创建 mempool
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发、Linux开发、
公众号开发、开源爱好者、Linux

联系QQ:562054870