4. 环形缓冲区操作

任务目标

使用DPDK的rte_ring无锁环形缓冲区，实现多生产者多消费者（MPMC）模式下数据包的入队与出队操作，验证其无锁并发性能。本案例聚焦于rte_ring的核心API调用与MPMC模式下的正确性验证，不涉及具体数据包处理逻辑。

环境依赖

• DPDK版本：21.11或更高（已编译并设置RTE_SDK与RTE_TARGET环境变量）
• 编译依赖：libdpdk-dev或DPDK源码编译产物
• 硬件要求：x86_64架构，多核CPU（至少2个物理核心用于并发测试）

核心实现

以下代码创建了一个容量为1024的MPMC环形缓冲区，启动4个生产者线程和4个消费者线程。每个生产者连续入队10000个递增编号的数据包指针，消费者统计出队总数并校验数据完整性。使用rte_ring_mp_enqueue_bulk和rte_ring_mc_dequeue_bulk批量操作以减少竞争开销。

#include <rte_ring.h>
#include <rte_lcore.h>
#include <rte_memory.h>
#include <rte_launch.h>
#include <stdio.h>
#include <stdint.h>

#define RING_SIZE 1024
#define PRODUCER_NUM 4
#define CONSUMER_NUM 4
#define PKT_PER_PRODUCER 10000

struct rte_ring *ring;
static uint64_t producer_done[PRODUCER_NUM];
static uint64_t consumer_count[CONSUMER_NUM];
static int consumer_error[CONSUMER_NUM];

static int producer_worker(void *arg) {
    uintptr_t id = (uintptr_t)arg;
    void *bulk[32];
    uint64_t base = id * PKT_PER_PRODUCER;
    
    for (int i = 0; i < PKT_PER_PRODUCER; i += 32) {
        int batch = (PKT_PER_PRODUCER - i) > 32 ? 32 : (PKT_PER_PRODUCER - i);
        for (int j = 0; j < batch; j++) {
            bulk[j] = (void *)(base + i + j + 1);  // 数据包编号从1开始
        }
        unsigned int enq = rte_ring_mp_enqueue_bulk(ring, bulk, batch, NULL);
        while (enq < (unsigned int)batch) {  // 队列满时重试剩余部分
            enq += rte_ring_mp_enqueue_bulk(ring, &bulk[enq], batch - enq, NULL);
        }
    }
    producer_done[id] = 1;
    return 0;
}

static int consumer_worker(void *arg) {
    uintptr_t id = (uintptr_t)arg;
    void *bulk[32];
    uint64_t total = 0;
    int error = 0;
    
    while (1) {
        unsigned int deq = rte_ring_mc_dequeue_bulk(ring, bulk, 32, NULL);
        for (unsigned int i = 0; i < deq; i++) {
            uint64_t pkt_id = (uint64_t)bulk[i];
            if (pkt_id == 0 || pkt_id > PRODUCER_NUM * PKT_PER_PRODUCER) {
                error = 1;
            }
            total++;
        }
        // 所有生产者完成后，队列清空则退出
        int all_done = 1;
        for (int p = 0; p < PRODUCER_NUM; p++) {
            if (!producer_done[p]) {
                all_done = 0;
                break;
            }
        }
        if (all_done && rte_ring_count(ring) == 0) {
            break;
        }
    }
    consumer_count[id] = total;
    consumer_error[id] = error;
    return 0;
}

int main(int argc, char **argv) {
    int ret = rte_eal_init(argc, argv);
    if (ret < 0) {
        return -1;
    }
    
    ring = rte_ring_create("test_ring", RING_SIZE, SOCKET_ID_ANY, 
                           RING_F_SP_ENQ | RING_F_SC_DEQ);
    if (ring == NULL) {
        /* 若创建失败，回退到MPMC模式 */
        ring = rte_ring_create("test_ring", RING_SIZE, SOCKET_ID_ANY, 0);
        if (ring == NULL) {
            rte_eal_cleanup();
            return -1;
        }
    }
    
    /* 启动消费者线程 */
    for (uintptr_t i = 0; i < CONSUMER_NUM; i++) {
        rte_eal_remote_launch(consumer_worker, (void *)i, rte_get_next_lcore(-1, 1, 0));
    }
    /* 启动生产者线程 */
    for (uintptr_t i = 0; i < PRODUCER_NUM; i++) {
        rte_eal_remote_launch(producer_worker, (void *)i, rte_get_next_lcore(-1, 1, 0));
    }
    
    rte_eal_mp_wait_lcore();
    
    /* 验证结果 */
    uint64_t total_enq = PRODUCER_NUM * PKT_PER_PRODUCER;
    uint64_t total_deq = 0;
    int any_error = 0;
    for (int i = 0; i < CONSUMER_NUM; i++) {
        total_deq += consumer_count[i];
        any_error |= consumer_error[i];
    }
    
    printf("Enqueued: %lu\n", total_enq);
    printf("Dequeued: %lu\n", total_deq);
    printf("Errors: %s\n", any_error ? "YES" : "NO");
    
    if (total_enq == total_deq && !any_error) {
        printf("RESULT: PASS\n");
    } else {
        printf("RESULT: FAIL\n");
    }
    
    rte_ring_free(ring);
    rte_eal_cleanup();
    return 0;
}

编译与运行

# 编译（假设DPDK环境已就绪）
gcc -o ring_test ring_test.c $(pkg-config --cflags --libs libdpdk) -lpthread

# 以4个lcore运行（2个生产者+2个消费者，需根据CPU拓扑调整）
sudo ./ring_test -l 0-3 -n 4

验证方法

程序输出中RESULT: PASS表示验证通过。判定标准：

• 出队总数等于入队总数（40000）
• 所有数据包编号均在有效范围内（1~40000）且无重复
• 程序无死锁或崩溃，正常退出

若需验证无锁性能，可对比相同逻辑下使用互斥锁保护普通队列的实现，观察本程序在多核下的吞吐优势。性能测试需使用rte_ring的批量操作并确保生产者/消费者分布在不同物理核心上。

常见问题

• 队列满导致死循环：生产者在队列满时使用while循环重试剩余批次，确保所有数据最终入队。
• 消费者提前退出：消费者通过检查producer_done标志和队列空状态判断是否结束，避免漏读数据。
• lcore不足：运行参数-l指定的核心数需大于等于8（4生产者+4消费者+主核心），否则线程会排队等待。