4. DPDK环形缓冲区操作

任务目标

本案例使用DPDK的rte_ring无锁环形缓冲区，实现多生产者（MP）与多消费者（MC）之间的数据传递。核心是验证：多个线程同时向ring写入数据，多个线程同时从ring读取数据，且所有数据均被正确接收、无丢失、无重复。

环境与依赖

• DPDK版本：20.11+（需编译含librte_ring库）
• 编译依赖：pkg-config --libs libdpdk 或手动链接 -lrte_ring -lrte_eal
• 运行环境：需root权限（HugePages与EAL初始化）

核心实现

代码创建两个生产者线程和两个消费者线程。每个生产者发送PRODUCER_COUNT个递增编号的数据项（共BULK_SIZE个），消费者接收并校验编号连续性，最终统计接收总数。使用rte_ring_mp_enqueue_burst与rte_ring_mc_dequeue_burst保证多线程安全。

#include <rte_ring.h>
#include <rte_eal.h>
#include <rte_lcore.h>
#include <stdio.h>
#include <stdint.h>

#define RING_SIZE 1024
#define PRODUCER_COUNT 1000
#define BULK_SIZE 32
#define NUM_PRODUCERS 2
#define NUM_CONSUMERS 2

static struct rte_ring *ring;
static volatile int producer_done[NUM_PRODUCERS];
static volatile int consumer_done[NUM_CONSUMERS];

struct prod_ctx {
    int id;
    uint64_t start_val;
};

struct cons_ctx {
    int id;
    uint64_t received;
    int error;
};

static int producer_thread(void *arg) {
    struct prod_ctx *ctx = (struct prod_ctx *)arg;
    uint64_t val = ctx->start_val;
    unsigned sent_total = 0;

    while (sent_total < PRODUCER_COUNT) {
        unsigned batch = (PRODUCER_COUNT - sent_total) > BULK_SIZE ?
                         BULK_SIZE : (PRODUCER_COUNT - sent_total);
        unsigned i;
        void *objs[BULK_SIZE];

        for (i = 0; i < batch; i++)
            objs[i] = (void *)(uintptr_t)val++;

        unsigned sent = rte_ring_mp_enqueue_burst(ring, objs, batch, NULL);
        sent_total += sent;
        /* 若ring满，重试剩余部分 */
        val -= (batch - sent);
    }
    producer_done[ctx->id] = 1;
    return 0;
}

static int consumer_thread(void *arg) {
    struct cons_ctx *ctx = (struct cons_ctx *)arg;
    void *objs[BULK_SIZE];
    uint64_t expected = ctx->id; /* 每个消费者校验自己的起始编号 */

    while (1) {
        unsigned received = rte_ring_mc_dequeue_burst(ring, objs, BULK_SIZE, NULL);
        if (received == 0) {
            /* 所有生产者结束且ring为空则退出 */
            int all_done = 1;
            for (int i = 0; i < NUM_PRODUCERS; i++) {
                if (!producer_done[i]) {
                    all_done = 0;
                    break;
                }
            }
            if (all_done && rte_ring_count(ring) == 0)
                break;
            continue;
        }
        for (unsigned i = 0; i < received; i++) {
            uint64_t val = (uint64_t)(uintptr_t)objs[i];
            if (val != expected) {
                ctx->error = 1;
                printf("Consumer %d: expected %lu, got %lu\n",
                       ctx->id, expected, val);
            }
            expected++;
            ctx->received++;
        }
    }
    consumer_done[ctx->id] = 1;
    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) {
        printf("Ring creation failed\n");
        return -1;
    }

    /* 注意：为演示MP/MC，需使用非SP/SC标志 */
    rte_ring_free(ring);
    ring = rte_ring_create("test_ring", RING_SIZE, SOCKET_ID_ANY, 0);
    if (!ring) {
        printf("Ring creation failed (MP/MC)\n");
        return -1;
    }

    struct prod_ctx pctx[NUM_PRODUCERS];
    struct cons_ctx cctx[NUM_CONSUMERS];
    unsigned lcore_id;
    unsigned used_lcores = 0;

    /* 启动生产者 */
    for (int i = 0; i < NUM_PRODUCERS; i++) {
        pctx[i].id = i;
        pctx[i].start_val = i * PRODUCER_COUNT;
        producer_done[i] = 0;
        lcore_id = rte_get_next_lcore(lcore_id, 1, 0);
        if (lcore_id == RTE_MAX_LCORE) {
            printf("Not enough lcores\n");
            return -1;
        }
        rte_eal_remote_launch(producer_thread, &pctx[i], lcore_id);
        used_lcores++;
    }

    /* 启动消费者 */
    for (int i = 0; i < NUM_CONSUMERS; i++) {
        cctx[i].id = i;
        cctx[i].received = 0;
        cctx[i].error = 0;
        consumer_done[i] = 0;
        lcore_id = rte_get_next_lcore(lcore_id, 1, 0);
        if (lcore_id == RTE_MAX_LCORE) {
            printf("Not enough lcores\n");
            return -1;
        }
        rte_eal_remote_launch(consumer_thread, &cctx[i], lcore_id);
        used_lcores++;
    }

    rte_eal_mp_wait_lcore();

    uint64_t total_received = 0;
    int has_error = 0;
    for (int i = 0; i < NUM_CONSUMERS; i++) {
        total_received += cctx[i].received;
        if (cctx[i].error)
            has_error = 1;
        printf("Consumer %d received: %lu\n", i, cctx[i].received);
    }

    uint64_t total_sent = NUM_PRODUCERS * PRODUCER_COUNT;
    printf("Total sent: %lu, Total received: %lu\n",
           total_sent, total_received);

    if (total_sent == total_received && !has_error)
        printf("TEST PASSED\n");
    else
        printf("TEST FAILED\n");

    rte_ring_free(ring);
    rte_eal_cleanup();
    return 0;
}

编译与运行

# 编译（假设DPDK已安装并通过pkg-config暴露）
gcc -o ring_test ring_test.c $(pkg-config --cflags --libs libdpdk) -lpthread

# 运行（需4个可用lcore，建议预留主核）
sudo ./ring_test -l 0-4

验证方法

• 数据完整性：每个消费者校验接收到的编号是否连续递增（起始编号为消费者ID），若出现不连续则打印错误并置error标志。
• 总数一致性：程序结束时比较total_sent（2×1000=2000）与total_received，相等且无错误时输出TEST PASSED。
• 并发正确性：由于使用rte_ring_mp_enqueue_burst和rte_ring_mc_dequeue_burst，多线程同时操作不会产生数据竞争。若改用SP/SC标志（代码中已注释），则多线程场景下会出错，可对比验证MP/MC的必要性。

预期输出

Consumer 0 received: 1000
Consumer 1 received: 1000
Total sent: 2000, Total received: 2000
TEST PASSED

注：每个消费者接收的具体数量可能不同（取决于调度），但总和必须为2000且无编号错误。