第18周：DPDK 核心库解析

2026-07-02

字数统计: 2.6k字 | 阅读时长≈ 12分

第18周：DPDK 核心库解析

目标：深入理解 DPDK 三大核心数据结构——mbuf、ring、mempool。

1. Mbuf（报文缓冲区）

1.1 结构体图示

struct rte_mbuf {
    // ─── 缓存行 1（64 bytes）───
    struct rte_mbuf       *next;       // 下一个 mbuf（链式包）
    void                  *data_addr;  // 数据缓冲地址
    uint16_t              data_off;   // 数据偏移（从 mbuf 头算起）
    uint16_t              buf_len;     // 总缓冲区长度
    uint16_t              pkt_len;     // 整个包长度（链式时 = 所有 mbuf 的 data_len 之和）
    uint16_t              data_len;    // 当前 mbuf 的数据长度
    uint64_t              phys_addr;   // 数据缓冲的物理地址（DMA 用）
    uint16_t              refcnt;      // 引用计数
    uint8_t               nb_segs;     // 分段数量
    uint8_t               port;        // 接收端口
    uint64_t              ol_flags;    // offload 标志

    // ─── 缓存行 2 ──
    uint32_t              vlan_tci;    // VLAN 标签
    uint32_t              hash;        // RSS hash 结果
    uint16_t              vlan_tci_outer;
    uint16_t              buf_iova;    // IOMMU VA（仅 VFIO）
    uint32_t              timestamp;   // 时间戳

    // ─── 缓存行 3 ──
    union {
        struct {
            uint16_t  l2_len;    // L2 头长度（以太网）
            uint16_t  l3_len;    // L3 头长度（IP）
            uint16_t  l4_len;    // L4 头长度（TCP/UDP）
        };
        struct rte_mbuf_sched sched; // 调度信息（多队列）
    };

    // ─── 私有数据（可选）───
    char                 priv[64];     // 用户自定义数据
};

// 内存布局：
// [rte_mbuf head] [data buffer]
//  64 bytes        多页大小（默认 2048 + headroom）
//                  headroom（默认 128 bytes）
//                  [数据 payload ...]

1.2 Mbuf 池（Mempool）创建

#include <rte_mbuf.h>

#define NB_MBUF      8192       // mbuf 数量
#define MBUF_CACHE   256        // 每个核本地缓存
#define MBUF_SIZE    (2048 + RTE_PKTMBUF_HEADROOM + sizeof(struct rte_mbuf))

int create_mempool(void) {
    // 创建 mbuf pool
    struct rte_mempool *mp;

    // 在指定 socket 上创建
    mp = rte_pktmbuf_pool_create("MBUF_POOL",  // 名称
                                  NB_MBUF,      // mbuf 数量
                                  MBUF_CACHE,   // 每个核的缓存
                                  0,            // 私有数据大小
                                  RTE_MBUF_DEFAULT_BUF_SIZE,  // 数据缓冲区大小
                                  SOCKET_ID_ANY); // 内存 socket

    if (mp == NULL) {
        printf("Cannot create mbuf pool: %s\n", rte_strerror(rte_errno));
        return -1;
    }

    printf("Mempool created: %s, %d mbufs, cache=%d\n",
           "MBUF_POOL", NB_MBUF, MBUF_CACHE);
    return 0;
}

1.3 Mbuf 分配与释放

// 从 mempool 分配一个 mbuf
struct rte_mbuf *mbuf = rte_pktmbuf_alloc(mbuf_pool);
if (mbuf == NULL) {
    // mempool 耗尽
    return -1;
}

// 设置数据指针
rte_pktmbuf_append(mbuf, data_len);  // 扩展数据区域
mbuf->data_len = data_len;
mbuf->pkt_len = data_len;

// 重置 mbuf（回收后复用）
rte_pktmbuf_reset(mbuf);

// 释放回 mempool
rte_pktmbuf_free(mbuf);

// 释放链式 mbuf（多段包）
rte_pktmbuf_free_bulk(mbufs, n);

1.4 Mbuf 链式包

大包（超过 MTU 1500）可能跨越多个 mbuf：

mbuf1 → mbuf2 → mbuf3
 │        │        │
 ├─────── headroom
 ├── data ──┤  ├── data ──┤  ├── data ──┤
 │  L2+L3   │  │  L4+body │  │  body    │
 │  (60B)   │  │  (1400B) │  │  (548B)  │

总 pkt_len = data_len1 + data_len2 + data_len3 = 2048
mbuf1->pkt_len = 2048（整个包的长度）
mbuf1->data_len = 60（当前段的数据长度）

// 创建一个链式包（2 段）
struct rte_mbuf *mbuf1 = rte_pktmbuf_alloc(pool);
struct rte_mbuf *mbuf2 = rte_pktmbuf_alloc(pool);

// 第一段：IP 头
rte_pktmbuf_append(mbuf1, 60);
// 第二段：TCP payload
rte_pktmbuf_append(mbuf2, 1400);

// 链式连接
mbuf1->next = mbuf2;
mbuf1->nb_segs = 2;
mbuf1->pkt_len = 1560;

// 遍历所有段
struct rte_mbuf *cur = mbuf1;
int seg = 0;
while (cur) {
    printf("Seg %d: data_len=%u\n", seg, cur->data_len);
    cur = cur->next;
    seg++;
}

2. Ring 无锁队列

2.1 核心概念

Ring 是一个无锁的环形缓冲区（SPSC/MPMC）：

  ┌──────────────────────────────────────┐
  │  prod.head      cons.head            │
  │      ↓              ↓                │
  │  ┌──┬──┬──┬──┬──┬──┬──┬──┬──┬──┐  │
  │  │  │  │  │  │  │  │  │  │  │  │  │
  │  └──┴──┴──┴──┴──┴──┴──┴──┴──┴──┘  │
  │      ↑              ↑                │
  │  prod.tail      cons.tail            │
  └──────────────────────────────────────┘

  生产者（producer）修改 head/tail
  消费者（consumer）修改 head/tail

  关键保证：
  - 多个生产者可以并发写入（CAS 操作）
  - 多个消费者可以并发读取（CAS 操作）
  - 单个 ring 支持：
      SP/SC（单生产者/单消费者）— 最快
      MP/SC（多生产者/单消费者）
      SP/MC（单生产者/多消费者）
      MP/MC（多生产者/多消费者）— 最通用

2.2 Ring 操作

#include <rte_ring.h>

// 创建 Ring
struct rte_ring *ring = rte_ring_create("RING",    // 名称
                                        1024,       // 元素数量（2 的幂）
                                        SOCKET_ID_ANY,  // NUMA socket
                                        RING_F_SP_ENQ | RING_F_SC_DEQ);
// 标志：
// RING_F_SP_ENQ  — 单生产者入队（fast path）
// RING_F_SC_DEQ  — 单消费者出队（fast path）
// 0              — 多生产者/多消费者

// 入队（enqueue）
unsigned int count;
void *obj = some_object;

// 单元素入队
ret = rte_ring_enqueue(ring, obj);
if (ret == 0) {
    // 成功
}

// 批量入队（burst，更高效）
struct rte_mbuf *pkts[32];
ret = rte_ring_enqueue_bulk(ring, (void **)pkts, 32);
// 返回成功入队的数量

// 出队（dequeue）
void *deq_obj;
ret = rte_ring_dequeue(ring, &deq_obj);

// 批量出队（burst）
ret = rte_ring_dequeue_burst(ring, (void **)pkts, 32, NULL);
// 返回实际取出的数量（可能少于 32，如果 ring 不够满）

// 查看状态
printf("Free: %u, Used: %u, Capacity: %u\n",
       rte_ring_free_count(ring),
       rte_ring_count(ring),
       rte_ring_get_capacity(ring));

// 销毁
rte_ring_free(ring);

2.3 使用 Ring 实现线程间包传递

// ring_comm.c
#include <rte_ring.h>
#include <rte_mbuf.h>
#include <pthread.h>

struct rte_ring *rx_ring;     // RX → 处理 的 ring
struct rte_ring *tx_ring;     // 处理 → TX 的 ring

// === 生产者线程（收包线程）===
void *rx_thread(void *arg) {
    struct rte_mbuf *pkts[32];
    (void)arg;

    while (1) {
        // 从网卡收包
        int nb_rx = rte_eth_rx_burst(0, 0, pkts, 32);

        if (nb_rx > 0) {
            // 将包放入 ring 供处理线程消费
            int enq = rte_ring_enqueue_bulk(rx_ring,
                                             (void **)pkts, nb_rx);
            if (enq < nb_rx) {
                // Ring 满，释放溢出的包
                for (int i = enq; i < nb_rx; i++)
                    rte_pktmbuf_free(pkts[i]);
            }
        }
    }
    return NULL;
}

// === 消费者线程（处理线程）===
void *worker_thread(void *arg) {
    struct rte_mbuf *pkts[32];
    (void)arg;

    while (1) {
        // 从 ring 取出包
        int nb_deq = rte_ring_dequeue_burst(rx_ring,
                                             (void **)pkts, 32, NULL);
        if (nb_deq == 0) continue;

        // 处理包（例如查 MAC 表转发）
        for (int i = 0; i < nb_deq; i++) {
            process_packet(pkts[i]);
        }

        // 处理后放入 TX ring
        rte_ring_enqueue_bulk(tx_ring, (void **)pkts, nb_deq);
    }
    return NULL;
}

// === TX 线程（发包线程）===
void *tx_thread(void *arg) {
    struct rte_mbuf *pkts[32];
    (void)arg;

    while (1) {
        // 从 ring 取出待发包
        int nb_deq = rte_ring_dequeue_burst(tx_ring,
                                             (void **)pkts, 32, NULL);
        if (nb_deq == 0) continue;

        // 发送
        rte_eth_tx_burst(0, 0, pkts, nb_deq);

        // 释放未发送成功的
        for (int i = nb_deq; i < nb_deq; i++)
            rte_pktmbuf_free(pkts[i]);
    }
    return NULL;
}

3. Mempool（内存池）

3.1 Mempool 与 Ring 的关系

Mempool = Ring + 对象管理

  ┌─────────────────────────────────────────┐
  │            Mempool                       │
  │  ┌─────────────┐                        │
  │  │  free ring  │ ← 空闲 mbuf 指针       │
  │  └─────────────┘                        │
  │  ┌─────────────┐                        │
  │  │   ring      │ ← 用户可管理的缓存 ring │
  │  └─────────────┘                        │
  │                                         │
  │  [mbuf1] [mbuf2] [mbuf3] ... [mbufN]   │ ← 实际内存
  │  ↑ 每个 mbuf 指向一块物理内存           │
  │  (来自 hugepage)                        │
  └─────────────────────────────────────────┘

  申请：从 free ring 取一个 mbuf 指针
  释放：将 mbuf 指针放回 free ring

3.2 Mempool 创建与使用

#include <rte_mempool.h>

// 自定义对象（例如一个包含 mbuf 引用的上下文）
struct pkt_context {
    uint32_t flow_hash;
    uint16_t out_port;
    uint64_t timestamp;
    struct rte_mbuf *mbuf;
};

// 创建对象池
struct rte_mempool *ctx_pool = rte_mempool_create(
    "PKT_CTX",         // 名称
    8192,              // 对象数量
    sizeof(struct pkt_context),  // 对象大小
    256,               // 缓存大小
    0,                 // 私有数据大小
    NULL,              // 私有数据初始化
    NULL,              // 对象初始化
    NULL,              // 参数
    SOCKET_ID_ANY,     // NUMA socket
    0                  // 标志
);

// 分配对象
struct pkt_context *ctx = rte_mempool_get(ctx_pool, NULL);

// 使用对象
ctx->flow_hash = hash;
ctx->out_port = 1;
ctx->timestamp = rte_rdtsc();

// 释放对象
rte_mempool_put(ctx_pool, ctx);

3.3 Mempool 的底层布局

Mempool 内存布局（一个连续的 hugepage 区域）：

  ┌─────────────────────────────────────────┐
  │  Mempool Header                         │
  │  (rte_mempool struct + 元数据)          │
  ├─────────────────────────────────────────┤
  │  Free Ring 数据区                       │
  │  (存储 mbuf 指针数组)                   │
  ├─────────────────────────────────────────┤
  │  Cache Ring 数据区（per-lcore）          │
  │  (每个 lcore 有独立的本地缓存)           │
  ├─────────────────────────────────────────┤
  │  MBUF Array                             │
  │  [rte_mbuf][rte_mbuf][rte_mbuf]...       │
  │  ↑ 每个指向对应的 data buffer           │
  ├─────────────────────────────────────────┤
  │  Data Buffers (hugepage)                │
  │  [2048+128 bytes per mbuf]              │
  └─────────────────────────────────────────┘

内存来源：
  - 默认使用 hugepage 分配
  - 通过 DPDK 的 rte_malloc 从 hugepage 池中分配
  - 物理地址连续，DMA 友好

4. Memzone（内存域）

#include <rte_memzone.h>

// Memzone：大块共享内存区域
// 用途：跨进程/跨核共享数据结构

// 创建 memzone
const struct rte_memzone *mz = rte_memzone_reserve(
    "SHARED_STATS",     // 名称（必须唯一）
    sizeof(struct stats),  // 大小
    SOCKET_ID_ANY,      // NUMA socket
    RTE_MEMZONE_ATOMIC   // 原子创建（多进程安全）
);

if (mz == NULL) {
    printf("Failed to reserve memzone: %s\n", rte_strerror(rte_errno));
    return -1;
}

// 使用
struct stats *st = mz->addr;
st->total_packets = 0;

// 查找已有 memzone
const struct rte_memzone *mz2 = rte_memzone_lookup("SHARED_STATS");
if (mz2) {
    printf("Found memzone at %p\n", mz2->addr);
}

5. Timer（高精度定时器）

#include <rte_timer.h>

// DPDK 定时器（基于 TSC 时钟周期）

static struct rte_timer timer1, timer2;

// 定时器回调
static void timer_cb(struct rte_timer *tim, void *arg) {
    printf("Timer fired! arg=%s\n", (char *)arg);
}

void setup_timers(void) {
    // 初始化定时器
    rte_timer_init(&timer1);
    rte_timer_init(&timer2);

    // 配置：单次触发，在 lcore 0 上执行
    uint64_t hz = rte_get_timer_hz();  // 获取 TSC 频率
    uint64_t delay = hz;  // 1 秒后触发

    rte_timer_reset(&timer1,        // 定时器
                    delay,          // 到期时间（ticks）
                    SINGLE,         // 单次触发（PERIODICAL = 周期）
                    rte_get_master_lcore(),  // 执行的 lcore
                    timer_cb,       // 回调
                    "timer1");      // 回调参数

    // 周期触发（每秒）
    rte_timer_reset(&timer2, hz, PERIODICAL,
                    rte_get_master_lcore(), timer_cb, "periodic");
}

// 在主循环中处理定时器
void main_loop(void) {
    while (1) {
        // 处理定时器
        rte_timer_manage();

        // 其他工作
        // ...
    }
}

6. 实践：线程间包传递完整示例

// ring_demo.c
#include <stdio.h>
#include <rte_eal.h>
#include <rte_ring.h>
#include <rte_mbuf.h>
#include <rte_mempool.h>
#include <pthread.h>

#define RING_SIZE  1024
#define NB_MBUF    4096
#define MBUF_CACHE 256

static struct rte_ring *g_ring;
static struct rte_mempool *g_mbuf_pool;

// 生产者线程：模拟收包并放入 ring
static void *producer_thread(void *arg) {
    unsigned int lcore_id = (unsigned int)(uintptr_t)arg;
    printf("Producer thread on lcore %u\n", lcore_id);

    while (1) {
        // 模拟收包：分配 mbuf 并填充
        struct rte_mbuf *mbuf = rte_pktmbuf_alloc(g_mbuf_pool);
        if (!mbuf) continue;

        // 模拟 payload
        uint8_t *data = rte_pktmbuf_mtod(mbuf, uint8_t *);
        snprintf((char *)data, 64, "Hello from lcore %u", lcore_id);
        mbuf->data_len = 32;
        mbuf->pkt_len = 32;

        // 入队
        int ret = rte_ring_enqueue(g_ring, mbuf);
        if (ret != 0) {
            printf("Ring full! Dropping packet\n");
            rte_pktmbuf_free(mbuf);
        }
    }
    return NULL;
}

// 消费者线程：从 ring 取包处理
static void *consumer_thread(void *arg) {
    unsigned int lcore_id = (unsigned int)(uintptr_t)arg;
    printf("Consumer thread on lcore %u\n", lcore_id);

    uint64_t counter = 0;
    uint64_t start = rte_rdtsc();

    while (1) {
        struct rte_mbuf *mbuf;
        int ret = rte_ring_dequeue(g_ring, (void **)&mbuf);
        if (ret != 0) continue;

        // 处理
        uint8_t *data = rte_pktmbuf_mtod(mbuf, uint8_t *);
        counter++;

        // 每 100 万包打印一次速率
        if (counter % 1000000 == 0) {
            uint64_t now = rte_rdtsc();
            double secs = (double)(now - start) / rte_get_timer_hz();
            printf("Consumer %u: %lu packets (%.2f Mpps)\n",
                   lcore_id, counter, counter / secs / 1e6);
        }

        rte_pktmbuf_free(mbuf);
    }
    return NULL;
}

int main(int argc, char **argv) {
    int ret;
    unsigned int lcore_id;
    pthread_t prod_tid, cons_tid;

    // 1. 初始化 EAL
    ret = rte_eal_init(argc, argv);
    if (ret < 0) {
        printf("EAL init failed\n");
        return 1;
    }

    // 2. 创建 mbuf pool
    g_mbuf_pool = rte_pktmbuf_pool_create("DEMO_MBUF", NB_MBUF,
                                           MBUF_CACHE, 0,
                                           RTE_MBUF_DEFAULT_BUF_SIZE,
                                           SOCKET_ID_ANY);
    if (!g_mbuf_pool) {
        printf("Cannot create mbuf pool\n");
        return 1;
    }

    // 3. 创建 ring
    g_ring = rte_ring_create("DEMO_RING", RING_SIZE,
                              SOCKET_ID_ANY, 0);
    if (!g_ring) {
        printf("Cannot create ring\n");
        return 1;
    }

    printf("Ring capacity: %u\n", rte_ring_get_capacity(g_ring));

    // 4. 创建线程
    pthread_create(&prod_tid, NULL, producer_thread,
                   (void *)(uintptr_t)rte_lcore_id());
    pthread_create(&cons_tid, NULL, consumer_thread,
                   (void *)(uintptr_t)rte_get_master_lcore());

    // 等待
    pthread_join(prod_tid, NULL);
    pthread_join(cons_tid, NULL);

    return 0;
}

7. 关键数据结构速查

7.1 rte_mbuf 关键字段

字段	含义	说明
`buf_len`	总缓冲区大小	包括 headroom
`data_off`	数据偏移	从 mbuf 开头到 data 的偏移
`data_len`	当前段数据长度	链式包中的当前段
`pkt_len`	整个包长度	链式包 = 所有段之和
`nb_segs`	分段数量	链式包 > 1
`next`	下一个 mbuf	链式包
`ol_flags`	offload 标志	IP/TCP/UDP checksum offload 等
`l2_len`	L2 头长度	以太网头
`l3_len`	L3 头长度	IP 头
`l4_len`	L4 头长度	TCP/UDP 头

7.2 Mbuf 辅助宏

// 获取数据指针
rte_pktmbuf_mtod(mbuf, uint8_t *)  // data_off + mbuf 头

// 扩展/收缩数据区
rte_pktmbuf_append(mbuf, len)      // 扩展（只在尾部）
rte_pktmbuf_adj(mbuf, len)         // 收缩头部
rte_pktmbuf_trim(mbuf, len)        // 收缩尾部

// 克隆 mbuf（共享数据缓冲区）
rte_pktmbuf_clone(mbuf, pool)      // 浅拷贝
rte_pktmbuf_copy(mbuf, pool, offset, len)  // 深拷贝

// 预分配 headroom
rte_pktmbuf_reserve(mbuf, headroom) // 在 data_off 之前预留空间

// 获取物理地址（DMA）
rte_pktmbuf_iova(mbuf)             // IOMMU VA
rte_pktmbuf_physaddr(mbuf)         // 物理地址

// 判断是否为链式包
rte_pktmbuf_is_contiguous(mbuf)    // 1 = 非链式（单段）

7.3 Ring 关键函数速查

函数	功能
`rte_ring_create()`	创建 ring
`rte_ring_enqueue()`	入队一个元素
`rte_ring_enqueue_bulk()`	批量入队
`rte_ring_dequeue()`	出队一个元素
`rte_ring_dequeue_burst()`	批量出队
`rte_ring_free_count()`	空闲槽位数
`rte_ring_count()`	已用槽位数
`rte_ring_empty()`	是否为空
`rte_ring_full()`	是否满
`rte_ring_free()`	销毁 ring

本文作者： CoderSong
本文链接： https://jack-song-gif.github.io/2026/07/02/第18周：DPDK 核心库解析/
版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！