1
# 方式一：-l 指定 lcore 列表（lcore 0→CPU 0, lcore 1→CPU 2, lcore 2→CPU 4, ...）
2
dpdk-app -l 0,2,4,6 -n 4 -- -p 0x3
3

4
# 方式二：-c 指定 CPU 位掩码（bit 0=CPU 0, bit 1=CPU 1, ...）
5
dpdk-app -c 0x55 -n 4 -- -p 0x3

1
// EAL 初始化中的 lcore 绑核逻辑（简化）
2
int rte_eal_init(int argc, char **argv)
3
{
4
    // 解析 -l/-c 参数，建立 lcore 映射表
5
    eal_parse_args(argc, argv);
6

7
    // 为每个 worker lcore 创建线程并绑核
8
    RTE_LCORE_FOREACH_WORKER(lcore_id) {
9
        // 设置 CPU 亲和性
10
        CPU_ZERO(&cpuset);
11
        CPU_SET(lcore_config[lcore_id].cpu_id, &cpuset);
12

13
        // 创建线程
14
        pthread_create(&lcore_config[lcore_id].thread_id,
15
                       NULL, eal_thread_loop, (void *)(uintptr_t)lcore_id);
16

17
        // 绑定到指定 CPU 核心
18
        pthread_setaffinity_np(lcore_config[lcore_id].thread_id,
19
                               sizeof(cpuset), &cpuset);
20
    }
21
    return 0;
22
}

1
// Master lcore 启动 Worker lcore
2
static int lcore_main(void *arg)
3
{
4
    uint16_t port_id = (uint16_t)(uintptr_t)arg;
5
    struct rte_mbuf *bufs[BURST_SIZE];
6

7
    printf("lcore %u 处理 port %u\n", rte_lcore_id(), port_id);
8

9
    while (!force_quit) {
10
        // 收包
11
        uint16_t nb_rx = rte_eth_rx_burst(port_id, 0, bufs, BURST_SIZE);
12
        if (unlikely(nb_rx == 0))
13
            continue;
14

15
        // 处理 + 发包
16
        for (uint16_t i = 0; i < nb_rx; i++) {
17
            // 数据包处理逻辑...
18
            process_packet(bufs[i]);
19
        }
20

21
        rte_eth_tx_burst(port_id, 0, bufs, nb_rx);
22

23
        // 释放已发送的 mbuf
24
        for (uint16_t i = 0; i < nb_rx; i++)
25
            rte_pktmbuf_free(bufs[i]);
26
    }
27
    return 0;
28
}
29

30
int main(int argc, char **argv)
31
{
32
    rte_eal_init(argc, argv);
33

34
    // 在所有 Worker lcore 上启动 lcore_main
35
    rte_eal_mp_remote_launch(lcore_main, NULL, CALL_MASTER);
36

37
    // 等待所有 lcore 完成
38
    rte_eal_mp_wait_lcore();
39
    return 0;
40
}

1
// rte_eal_mp_remote_launch 的内部实现（简化）
2
int rte_eal_mp_remote_launch(lcore_function_t *f, void *arg,
3
                              enum rte_rmt_call_master_t call_master)
4
{
5
    int lcore_id;
6

7
    // 遍历所有 Worker lcore，设置待执行函数
8
    RTE_LCORE_FOREACH_WORKER(lcore_id) {
9
        lcore_config[lcore_id].f = f;
10
        lcore_config[lcore_id].arg = arg;
11
        // 唤醒 Worker lcore（通过管道写触发）
12
        rte_atomic_store_explicit(&lcore_config[lcore_id].state,
13
                                   WAIT, rte_memory_order_release);
14
    }
15

16
    // 如果需要，Master lcore 也执行该函数
17
    if (call_master == CALL_MASTER) {
18
        f(arg);
19
    }
20

21
    return 0;
22
}

1
// Worker lcore 的主循环（简化）
2
static __attribute__((noreturn)) void *
3
eal_thread_loop(__attribute__((unused)) void *arg)
4
{
5
    unsigned lcore_id = rte_lcore_id();
6

7
    while (1) {
8
        // 等待 Master 分配任务
9
        while (lcore_config[lcore_id].state != WAIT)
10
            pthread_cond_wait(...);
11

12
        // 执行分配的函数
13
        lcore_function_t *f = lcore_config[lcore_id].f;
14
        void *f_arg = lcore_config[lcore_id].arg;
15
        lcore_config[lcore_id].ret = f(f_arg);
16

17
        // 标记完成
18
        lcore_config[lcore_id].state = FINISHED;
19
    }
20
}

1
              ┌──────────────────────────────────────────┐
2
              │              网卡 (NIC)                   │
3
              │  Queue 0  Queue 1  Queue 2  Queue 3      │
4
              └────┬────────┬────────┬────────┬──────────┘
5
                   │        │        │        │
6
         ┌─────────▼──┐ ┌──▼────────┐ ┌──────▼───┐ ┌──▼─────────┐
7
         │  lcore 0   │ │  lcore 1  │ │  lcore 2 │ │  lcore 3   │
8
         │            │ │           │ │          │ │            │
9
         │ rx_burst() │ │ rx_burst()│ │rx_burst()│ │ rx_burst() │
10
         │    ↓       │ │    ↓      │ │   ↓      │ │    ↓       │
11
         │ parse()    │ │ parse()   │ │ parse()  │ │ parse()    │
12
         │    ↓       │ │    ↓      │ │   ↓      │ │    ↓       │
13
         │ lookup()   │ │ lookup()  │ │ lookup() │ │ lookup()   │
14
         │    ↓       │ │    ↓      │ │   ↓      │ │    ↓       │
15
         │ modify()   │ │ modify()  │ │ modify() │ │ modify()   │
16
         │    ↓       │ │    ↓      │ │   ↓      │ │    ↓       │
17
         │ tx_burst() │ │ tx_burst()│ │tx_burst()│ │ tx_burst() │
18
         └────────────┘ └───────────┘ └──────────┘ └────────────┘

1
#define BURST_SIZE 32
2

3
// Run-to-Completion 主循环
4
static int rtc_lcore_main(void *arg)
5
{
6
    uint16_t port_id = (uint16_t)(uintptr_t)arg;
7
    uint16_t queue_id = rte_lcore_id() - 1;  // 每个 lcore 使用自己的队列
8
    struct rte_mbuf *bufs[BURST_SIZE];
9
    struct rte_mbuf *tx_bufs[BURST_SIZE];
10
    uint16_t nb_tx;
11

12
    printf("RTC: lcore %u 处理 port %u queue %u\n",
13
           rte_lcore_id(), port_id, queue_id);
14

15
    while (!force_quit) {
16
        // 1. 收包
17
        uint16_t nb_rx = rte_eth_rx_burst(port_id, queue_id,
18
                                           bufs, BURST_SIZE);
19
        if (unlikely(nb_rx == 0))
20
            continue;
21

22
        uint16_t tx_count = 0;
23

24
        // 2. 逐包处理：解析 → 查表 → 修改
25
        for (uint16_t i = 0; i < nb_rx; i++) {
26
            struct rte_ether_hdr *eth = rte_pktmbuf_mtod(bufs[i],
27
                                                          struct rte_ether_hdr *);
28

29
            // 简单的 L2 转发：交换源/目的 MAC
30
            struct rte_ether_addr tmp;
31
            rte_ether_addr_copy(&eth->src_addr, &tmp);
32
            rte_ether_addr_copy(&eth->dst_addr, &eth->src_addr);
33
            rte_ether_addr_copy(&tmp, &eth->dst_addr);
34

35
            tx_bufs[tx_count++] = bufs[i];
36
        }
37

38
        // 3. 发包
39
        nb_tx = rte_eth_tx_burst(port_id, queue_id, tx_bufs, tx_count);
40

41
        // 4. 释放未发送的 mbuf
42
        if (unlikely(nb_tx < tx_count)) {
43
            for (uint16_t i = nb_tx; i < tx_count; i++)
44
                rte_pktmbuf_free(tx_bufs[i]);
45
        }
46
    }
47
    return 0;
48
}

1
// Pipeline 模型：三阶段流水线
2
// Stage 1: 收包 lcore
3
static int pipeline_rx_stage(void *arg)
4
{
5
    uint16_t port_id = (uint16_t)(uintptr_t)arg;
6
    struct rte_mbuf *bufs[BURST_SIZE];
7
    struct rte_ring *rx_ring = rte_ring_lookup("rx_ring");
8

9
    while (!force_quit) {
10
        uint16_t nb_rx = rte_eth_rx_burst(port_id, 0, bufs, BURST_SIZE);
11
        if (nb_rx == 0)
12
            continue;
13

14
        // 将收到的 mbuf 指针放入 ring，传递给下一阶段
15
        uint16_t enqueued = rte_ring_sp_enqueue_burst(
16
            rx_ring, (void *const *)bufs, nb_rx, NULL);
17

18
        // 释放未入队的 mbuf（ring 满时背压）
19
        for (uint16_t i = enqueued; i < nb_rx; i++)
20
            rte_pktmbuf_free(bufs[i]);
21
    }
22
    return 0;
23
}
24

25
// Stage 2: 解析 + 查表 lcore
26
static int pipeline_process_stage(void *arg)
27
{
28
    struct rte_mbuf *bufs[BURST_SIZE];
29
    struct rte_ring *rx_ring = rte_ring_lookup("rx_ring");
30
    struct rte_ring *tx_ring = rte_ring_lookup("tx_ring");
31

32
    while (!force_quit) {
33
        // 从 rx_ring 取出数据包
34
        uint16_t nb_deq = rte_ring_sc_dequeue_burst(
35
            rx_ring, (void **)bufs, BURST_SIZE, NULL);
36
        if (nb_deq == 0)
37
            continue;
38

39
        // 处理每个数据包
40
        for (uint16_t i = 0; i < nb_deq; i++) {
41
            process_packet(bufs[i]);
42
        }
43

44
        // 将处理后的数据包放入 tx_ring
45
        uint16_t enqueued = rte_ring_sp_enqueue_burst(
46
            tx_ring, (void *const *)bufs, nb_deq, NULL);
47

48
        for (uint16_t i = enqueued; i < nb_deq; i++)
49
            rte_pktmbuf_free(bufs[i]);
50
    }
51
    return 0;
52
}
53

54
// Stage 3: 发包 lcore
55
static int pipeline_tx_stage(void *arg)
56
{
57
    uint16_t port_id = (uint16_t)(uintptr_t)arg;
58
    struct rte_mbuf *bufs[BURST_SIZE];
59
    struct rte_ring *tx_ring = rte_ring_lookup("tx_ring");
60

61
    while (!force_quit) {
62
        uint16_t nb_deq = rte_ring_sc_dequeue_burst(
63
            tx_ring, (void **)bufs, BURST_SIZE, NULL);
64
        if (nb_deq == 0)
65
            continue;
66

67
        uint16_t nb_tx = rte_eth_tx_burst(port_id, 0, bufs, nb_deq);
68

69
        for (uint16_t i = nb_tx; i < nb_deq; i++)
70
            rte_pktmbuf_free(bufs[i]);
71
    }
72
    return 0;
73
}

1
// 创建 SPSC ring
2
struct rte_ring *ring;
3
ring = rte_ring_create("pipeline_ring", 1024,
4
                        rte_socket_id(), RING_F_SP_ENQ | RING_F_SC_DEQ);
5

6
// 生产者 lcore：批量入队
7
static int producer_lcore(void *arg)
8
{
9
    struct rte_ring *ring = rte_ring_lookup("pipeline_ring");
10
    struct rte_mbuf *bufs[BURST_SIZE];
11

12
    while (!force_quit) {
13
        uint16_t nb_rx = rte_eth_rx_burst(port_id, queue_id,
14
                                           bufs, BURST_SIZE);
15
        if (nb_rx == 0)
16
            continue;
17

18
        // SP 批量入队：无需 CAS，无需重试
19
        uint16_t enqueued = rte_ring_sp_enqueue_burst(
20
            ring, (void *const *)bufs, nb_rx, NULL);
21

22
        // 处理未入队的包（背压）
23
        for (uint16_t i = enqueued; i < nb_rx; i++)
24
            rte_pktmbuf_free(bufs[i]);
25
    }
26
    return 0;
27
}
28

29
// 消费者 lcore：批量出队
30
static int consumer_lcore(void *arg)
31
{
32
    struct rte_ring *ring = rte_ring_lookup("pipeline_ring");
33
    struct rte_mbuf *bufs[BURST_SIZE];
34

35
    while (!force_quit) {
36
        // SC 批量出队：无需 CAS，无需重试
37
        uint16_t nb_deq = rte_ring_sc_dequeue_burst(
38
            ring, (void **)bufs, BURST_SIZE, NULL);
39
        if (nb_deq == 0)
40
            continue;
41

42
        // 处理数据包
43
        for (uint16_t i = 0; i < nb_deq; i++)
44
            process_packet(bufs[i]);
45
    }
46
    return 0;
47
}

1
// 批量入队的内部优化（简化）
2
static __rte_always_inline unsigned int
3
rte_ring_sp_enqueue_burst(struct rte_ring *r, void *const *obj_table,
4
                           unsigned int n, unsigned int *free_space)
5
{
6
    uint32_t prod_head = r->prod.head;
7
    uint32_t cons_tail = r->cons.tail;
8
    uint32_t free_count = r->size + cons_tail - prod_head;
9

10
    // 实际入队数量 = min(请求数量, 可用空间)
11
    uint32_t actual = (n > free_count) ? free_count : n;
12
    if (actual == 0) {
13
        if (free_space) *free_space = 0;
14
        return 0;
15
    }
16

17
    // 批量拷贝指针——连续内存写入，缓存行友好
18
    const uint32_t mask = r->mask;
19
    uint32_t idx = prod_head & mask;
20

21
    // 处理环形回绕：可能需要分两段拷贝
22
    if (idx + actual <= r->size) {
23
        // 无回绕：一次 memcpy
24
        memcpy(&r->ring[idx], obj_table, actual * sizeof(void *));
25
    } else {
26
        // 有回绕：分两段拷贝
27
        uint32_t first_part = r->size - idx;
28
        memcpy(&r->ring[idx], obj_table, first_part * sizeof(void *));
29
        memcpy(&r->ring[0], &obj_table[first_part],
30
               (actual - first_part) * sizeof(void *));
31
    }
32

33
    // 内存屏障 + 更新指针
34
    rte_smp_wmb();
35
    r->prod.head = prod_head + actual;
36
    r->prod.tail = prod_head + actual;
37

38
    if (free_space) *free_space = free_count - actual;
39
    return actual;
40
}

1
// rte_ring 结构体中的 cache-line 对齐（DPDK 源码简化）
2
struct rte_ring {
3
    // ... 其他字段 ...
4

5
    alignas(RTE_CACHE_LINE_SIZE) struct rte_ring_headtail prod;
6
    // prod 独占一个 cache-line（64 字节）
7

8
    alignas(RTE_CACHE_LINE_SIZE) struct rte_ring_headtail cons;
9
    // cons 独占一个 cache-line（64 字节）
10
};

1
// 非原子的计数器更新——多核下会丢失更新
2
uint64_t packet_count;  // 全局计数器
3

4
// lcore 0 和 lcore 1 同时执行：
5
packet_count++;  // 编译为：load → add → store，不是原子操作
6
// 可能的结果：两个核都 load 了相同的值，各自 +1 后 store，丢失一次更新

1
#include <rte_atomic.h>
2

3
// 原子计数器
4
static rte_atomic64_t total_packets;
5

6
// 原子递增
7
rte_atomic64_inc(&total_packets);
8

9
// 原子读取
10
uint64_t count = rte_atomic64_read(&total_packets);
11

12
// 原子比较并交换（CAS）
13
rte_atomic64_t version;
14
int64_t expected = 0;
15
int success = rte_atomic64_cmpset(
16
    (volatile uint64_t *)&version,  // 目标地址
17
    (uint64_t)expected,              // 期望值
18
    (uint64_t)1                      // 新值
19
);
20
// 如果 version == expected，则 version = 1，返回 1
21
// 否则不做任何操作，返回 0

1
// 编译器屏障：阻止编译器重排序，但不影响 CPU
2
rte_compiler_barrier();
3

4
// SMP 内存屏障：在所有核心之间建立全局顺序
5
rte_smp_mb();    // 全屏障（读+写）
6
rte_smp_rmb();   // 读屏障
7
rte_smp_wmb();   // 写屏障
8

9
// I/O 内存屏障：用于设备寄存器访问
10
rte_io_mb();
11
rte_io_rmb();
12
rte_io_wmb();

1
// rte_ring MPMC enqueue 中的屏障使用
2
static __rte_always_inline unsigned int
3
__rte_ring_move_prod_head(struct rte_ring *r, unsigned int is_sp,
4
                           unsigned int n, uint32_t *old_head,
5
                           uint32_t *new_head, uint32_t *free_entries)
6
{
7
    do {
8
        // 1. 读取当前 prod.head
9
        *old_head = r->prod.head;
10

11
        // 2. 编译器屏障：确保 old_head 的读取不会被重排到后面
12
        rte_smp_rmb();
13

14
        // 3. 读取 cons.tail（消费者已确认的出队位置）
15
        *free_entries = r->size + r->cons.tail - *old_head;
16

17
        if (n > *free_entries)
18
            return 0;
19

20
        *new_head = *old_head + n;
21

22
    // 4. CAS 更新 prod.head（包含完整的内存屏障）
23
    } while (rte_atomic32_cmpset(&r->prod.head, *old_head, *new_head) == 0);
24

25
    return n;
26
}

1
#include <stdatomic.h>
2

3
// C11 原子变量
4
static atomic_uint_fast64_t total_packets;
5

6
// 原子递增（relaxed 序：无屏障，最快）
7
atomic_fetch_add_explicit(&total_packets, 1, memory_order_relaxed);
8

9
// 原子读取（relaxed 序：仅保证原子性，不保证顺序）
10
uint64_t count = atomic_load_explicit(&total_packets, memory_order_relaxed);
11

12
// 原子 CAS（acquire-release 序：保证 CAS 前后的读写顺序）
13
uint64_t expected = 0;
14
bool success = atomic_compare_exchange_strong_explicit(
15
    &version, &expected, 1,
16
    memory_order_acq_rel,   // 成功时的内存序
17
    memory_order_acquire    // 失败时的内存序
18
);

1
// Per-lcore 计数器——完全无锁
2
static RTE_DEFINE_PER_LCORE(uint64_t, local_packet_count);
3

4
// 每个 lcore 独立更新自己的计数器（无原子操作，无屏障）
5
local_packet_count++;
6

7
// 汇总：Master lcore 读取所有 Worker 的计数器
8
uint64_t get_total_packets(void)
9
{
10
    uint64_t total = 0;
11
    unsigned int lcore_id;
12
    RTE_LCORE_FOREACH_WORKER(lcore_id) {
13
        total += RTE_PER_LCORE_BY_ID(lcore_id, local_packet_count);
14
    }
15
    total += local_packet_count;  // 加上 Master 的
16
    return total;
17
}

1
#include <rte_rcu_qsbr.h>
2

3
#define MAX_LCORES 64
4

5
// RCU 变量
6
struct rte_rcu_qsbr *qsv;
7

8
// 初始化 RCU
9
int init_rcu(void)
10
{
11
    size_t sz = rte_rcu_qsbr_get_memsize(MAX_LCORES);
12
    qsv = rte_zmalloc("rcu", sz, RTE_CACHE_LINE_SIZE);
13
    if (qsv == NULL)
14
        return -ENOMEM;
15

16
    return rte_rcu_qsbr_init(qsv, MAX_LCORES);
17
}
18

19
// 读者 lcore：注册 + 报告静默状态
20
static int reader_lcore(void *arg)
21
{
22
    unsigned int lcore_id = rte_lcore_id();
23

24
    // 注册当前 lcore 为 RCU 读者
25
    rte_rcu_qsbr_thread_register(qsv, lcore_id);
26
    rte_rcu_qsbr_thread_online(qsv, lcore_id);
27

28
    while (!force_quit) {
29
        // 读取 RCU 保护的数据（无锁）
30
        struct flow_entry *flow = lookup_flow_table(packet);
31
        if (flow) {
32
            // 使用 flow 数据——在此期间，写者不会回收 flow
33
            process_flow(flow, packet);
34
        }
35

36
        // 报告静默状态：声明"我已离开临界区"
37
        rte_rcu_qsbr_quiescent(qsv, lcore_id);
38
    }
39

40
    rte_rcu_qsbr_thread_offline(qsv, lcore_id);
41
    rte_rcu_qsbr_thread_unregister(qsv, lcore_id);
42
    return 0;
43
}
44

45
// 写者线程：更新流表
46
void update_flow_entry(struct flow_entry *old_flow,
47
                        struct flow_entry *new_flow)
48
{
49
    // 1. 创建新版本（copy + modify）
50
    // new_flow 已经在外部创建并初始化
51

52
    // 2. 原子替换指针
53
    __atomic_store_n(&flow_table[old_flow->hash], new_flow,
54
                      __ATOMIC_RELEASE);
55

56
    // 3. 等待宽限期：所有读者都已离开旧数据的临界区
57
    rte_rcu_qsbr_synchronize(qsv, RTE_QSBR_THRID_INVALID, 0);
58

59
    // 4. 安全回收旧数据
60
    rte_free(old_flow);
61
}

1
// 异步 RCU 回调：宽限期结束后自动回收旧数据
2
struct rte_rcu_qsbr_dq *dq;
3

4
// 创建延迟回收队列
5
struct rte_rcu_qsbr_dq_parameters params = {
6
    .name = "flow_rcu_dq",
7
    .size = 1024,
8
    .esize = sizeof(struct flow_entry *),
9
    .qsv = qsv,
10
    .trigger_reclaim_limit = 128,  // 队列中超过 128 个待回收项时触发
11
    .reclaim_max = 64,             // 每次最多回收 64 个
12
};
13
dq = rte_rcu_qsbr_dq_create(&params);
14

15
// 写者：将旧数据放入延迟回收队列（不阻塞）
16
void update_flow_async(struct flow_entry *old_flow,
17
                        struct flow_entry *new_flow)
18
{
19
    // 原子替换指针
20
    __atomic_store_n(&flow_table[old_flow->hash], new_flow,
21
                      __ATOMIC_RELEASE);
22

23
    // 将旧数据入队，由 RCU 在宽限期后自动回收
24
    rte_rcu_qsbr_dq_enqueue(dq, &old_flow, 1);
25
}
26

27
// 定期调用（可在任意 lcore 的主循环中）
28
void reclaim_rcu_resources(void)
29
{
30
    rte_rcu_qsbr_dq_reclaim(dq, UINT_MAX, NULL, NULL, NULL);
31
}

1
#include <rte_eventdev.h>
2

3
#define NUM_EVENTS 32
4

5
// Eventdev 初始化
6
int setup_eventdev(uint8_t dev_id, uint16_t nb_ports, uint16_t nb_queues)
7
{
8
    struct rte_event_dev_config dev_conf = {
9
        .nb_events_limit = 4096,
10
        .nb_event_queues = nb_queues,
11
        .nb_event_ports = nb_ports,
12
        .nb_event_queue_flows = 1024,
13
        .nb_event_port_dequeue_depth = 64,
14
        .nb_event_port_enqueue_depth = 64,
15
    };
16

17
    int ret = rte_event_dev_configure(dev_id, &dev_conf);
18
    if (ret < 0)
19
        return ret;
20

21
    // 配置事件队列
22
    struct rte_event_queue_conf queue_conf = {
23
        .schedule_type = RTE_SCHED_TYPE_ATOMIC,
24
        .priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
25
        .nb_atomic_flows = 1024,
26
        .nb_atomic_order_sequences = 1024,
27
    };
28

29
    for (uint16_t i = 0; i < nb_queues; i++) {
30
        rte_event_queue_setup(dev_id, i, &queue_conf);
31
    }
32

33
    // 配置事件端口
34
    struct rte_event_port_conf port_conf = {
35
        .new_event_threshold = 1024,
36
        .dequeue_depth = 64,
37
        .enqueue_depth = 64,
38
    };
39

40
    for (uint16_t i = 0; i < nb_ports; i++) {
41
        rte_event_port_setup(dev_id, i, &port_conf);
42
        // 将所有队列链接到该端口
43
        uint8_t queue_ids[nb_queues];
44
        for (uint16_t q = 0; q < nb_queues; q++)
45
            queue_ids[q] = q;
46
        rte_event_port_link(dev_id, i, queue_ids, NULL, nb_queues);
47
    }
48

49
    return rte_event_dev_start(dev_id);
50
}
51

52
// RX Adapter：将网卡收包转换为事件
53
int setup_rx_adapter(uint8_t eventdev_id, uint16_t eth_port_id)
54
{
55
    struct rte_event_eth_rx_adapter_conf conf = {
56
        .event_buf_size = 1024,
57
    };
58

59
    uint8_t rx_adapter_id;
60
    rte_event_eth_rx_adapter_create_ext(&rx_adapter_id, eventdev_id,
61
                                         rx_adapter_conf_cb, &conf);
62

63
    struct rte_event_eth_rx_adapter_queue_conf queue_conf = {
64
        .rx_queue_flags = 0,
65
        .ev = {
66
            .queue_id = 0,           // 事件进入 Queue 0
67
            .priority = RTE_EVENT_DEV_PRIORITY_NORMAL,
68
            .sched_type = RTE_SCHED_TYPE_ATOMIC,
69
        },
70
        .servicing_weight = 1,       // 轮询模式
71
    };
72

73
    rte_event_eth_rx_adapter_queue_add(rx_adapter_id, eth_port_id,
74
                                        -1, &queue_conf);
75
    rte_event_eth_rx_adapter_start(rx_adapter_id);
76
    return 0;
77
}
78

79
// Worker lcore：事件驱动处理
80
static int eventdev_worker(void *arg)
81
{
82
    uint8_t dev_id = (uint8_t)(uintptr_t)arg;
83
    uint8_t port_id = rte_lcore_id();  // 每个 lcore 使用自己的 port
84
    struct rte_event events[NUM_EVENTS];
85

86
    while (!force_quit) {
87
        // 从调度器获取事件
88
        uint16_t nb_deq = rte_event_dequeue_burst(dev_id, port_id,
89
                                                    events, NUM_EVENTS, 0);
90
        if (nb_deq == 0)
91
            continue;
92

93
        for (uint16_t i = 0; i < nb_deq; i++) {
94
            struct rte_event *ev = &events[i];
95
            struct rte_mbuf *mbuf = ev->mbuf;
96

97
            // 处理数据包
98
            process_packet(mbuf);
99

100
            // 如果需要继续下一阶段处理，重新入队
101
            if (needs_further_processing(mbuf)) {
102
                ev->queue_id = 1;  // 发送到下一个队列
103
                rte_event_enqueue_burst(dev_id, port_id, ev, 1);
104
            } else {
105
                // 处理完成，释放 mbuf
106
                rte_pktmbuf_free(mbuf);
107
            }
108
        }
109
    }
110
    return 0;
111
}

1
# 启动 testpmd，4 个 lcore，4 个队列
2
dpdk-testpmd -l 0-4 -n 4 -- -i --rxq=4 --txq=4 --rss-ip
3

4
# testpmd 中查看各队列收包统计
5
show port xstats all
6

7
# 使用 pktgen 发送不同源 IP 的流量，观察 RSS 分发
8
# 预期：4 个队列的收包数大致相等

1
// 创建跨核 ring
2
struct rte_ring *rx_ring = rte_ring_create("rx_ring", 4096,
3
    rte_socket_id(), RING_F_SP_ENQ | RING_F_SC_DEQ);
4
struct rte_ring *tx_ring = rte_ring_create("tx_ring", 4096,
5
    rte_socket_id(), RING_F_SP_ENQ | RING_F_SC_DEQ);
6

7
// 启动 3 个 lcore 分别执行收包、处理、发包
8
rte_eal_remote_launch(rx_stage, (void *)(uintptr_t)port_id, 1);
9
rte_eal_remote_launch(process_stage, NULL, 2);
10
rte_eal_remote_launch(tx_stage, (void *)(uintptr_t)port_id, 3);

1
// 流表结构（RCU 保护）
2
struct flow_table_entry {
3
    uint32_t src_ip;
4
    uint32_t dst_ip;
5
    uint16_t src_port;
6
    uint16_t dst_port;
7
    uint8_t  proto;
8
    uint16_t output_port;
9
} __rte_cache_aligned;
10

11
// 查表（读者，无锁）
12
static inline struct flow_table_entry *
13
flow_lookup(struct flow_table *tbl, struct rte_mbuf *pkt)
14
{
15
    // 计算哈希
16
    uint32_t hash = flow_hash(pkt);
17
    // 读取指针（原子 load，RCU 保护）
18
    struct flow_table_entry *entry =
19
        __atomic_load_n(&tbl->entries[hash], __ATOMIC_ACQUIRE);
20
    return entry;
21
}
22

23
// 更新流表（写者，RCU 保护）
24
void flow_table_update(struct flow_table *tbl, uint32_t hash,
25
                        struct flow_table_entry *new_entry)
26
{
27
    struct flow_table_entry *old_entry =
28
        __atomic_exchange_n(&tbl->entries[hash], new_entry,
29
                             __ATOMIC_ACQ_REL);
30

31
    if (old_entry) {
32
        // 等待宽限期后回收旧表项
33
        rte_rcu_qsbr_synchronize(qsv, RTE_QSBR_THRID_INVALID, 0);
34
        rte_free(old_entry);
35
    }
36
}

1
# 使用 event_dsw PMD
2
# 在应用启动时指定 --vdev=event_dsw0
3
dpdk-app --vdev=event_dsw0 -l 0-4 -n 4 -- -p 0x3