DPDK非lcore线程调用rte_pktmbuf_free_bulk无法释放rte_mbuf问题

核心问题 #

使用DPDK高吞吐量接收UDP数据包时，在非DPDK lcore线程调用rte_pktmbuf_free_bulk()无法实际释放rte_mbuf，导致接收数据包无法超过4096个。询问是否必须将所有逻辑迁移至lcore线程，或有无其他解决方法。

DPDK配置说明 #

刚接触DPDK，接手的代码初始化部分过长无法全部提供，若有配置相关具体问题可补充细节。以下为主接收循环代码及说明。

代码说明 #

• 主接收循环_receive_packets()仅通过rte_eal_remote_launch()调用，参数以void指针传递
• thread_arg为自定义结构体，用于传递调用所需参数
• 因特定要求，_receive_packets()必须为静态类方法，通过input_arg->ds_instance指向调用该方法的Data_Streamer实例
• 应用关闭时，input_arg->ds_instance->m_active会被设为false
• BURST_SIZE值为512
• IO_Job为自定义结构体，用于通过rte_ring将数据包发送至非lcore线程处理并写入文件；出队作业代码会调用rte_eal_remote_launch()处理数据包并删除_receive_packets()中创建的对象，无内存泄漏
• input_arg->ds_instance->enqueue_io_job()为辅助函数，仅用于将作业入队至上述rte_ring

代码 #

void Data_Streamer::_receive_packets(void* arg) 
{
    thread_arg* input_arg = (thread_arg*) arg;

    uint16_t port_id = input_arg->port_id;
    uint16_t queue_id = input_arg->queue_id;
    long long bytes_recorded = 0;

    
    while (input_arg->ds_instance->m_active)
    {
        // If BURST_SIZE packets not ready, then do nothing this cycle
        if (rte_eth_rx_queue_count(port_id, queue_id) < BURST_SIZE) continue;
        
        // Once data is available, get packets as burst
        rte_mbuf** packets = new rte_mbuf*[BURST_SIZE];
        const uint16_t got_num_packets = rte_eth_rx_burst(port_id, queue_id, packets, BURST_SIZE);

        // Package packets into job and enqueue the job
        if (got_num_packets)
        {
            IO_Job* job = new IO_Job;
            job->byte_offset = bytes_recorded;
            job->packets = packets;
            job->num_packets = got_num_packets;

            input_arg->ds_instance->enqueue_io_job(job);
        }
    }
}

排查笔记 #

• 若在_receive_packets()入队作业后立即调用rte_pktmbuf_free_bulk()，数据包可正常释放，但会出现竞态条件：非lcore线程处理前，数据包数据可能被新数据包覆盖
• 尝试将数据包缓存在_receive_packets()中，待非lcore线程发送处理完成通知后再释放，结果与在非lcore线程释放类似——数据包未被释放（仿佛仅在非lcore线程读取数据就会导致即使在lcore线程释放也无法正常完成）
• 进一步修改为使用rte_pktmbuf_mtod()提取缓冲区地址，不向非lcore线程发送数据包对象指针，rte_mbuf仍无法正常释放

解决方法 #

1. 初始化非lcore线程的EAL环境 #

DPDK内存池（rte_mempool）操作依赖线程本地存储（TLS），非DPDK lcore线程（未通过rte_eal_remote_launch()启动的线程）必须先调用rte_eal_init_thread()完成线程初始化，否则无法正确访问内存池本地缓存，导致rte_pktmbuf_free_bulk()无法将mbuf放回内存池。

2. 配置线程安全的内存池 #

初始化rte_mempool时，根据业务场景设置对应的线程安全标志：

• 若非lcore线程仅作为mbuf的释放方，设置MEMPOOL_F_SP_PUT（单生产者）+MEMPOOL_F_SC_GET（单消费者）标志
• 若存在多线程生产/消费场景，使用MEMPOOL_F_MP_MC（多生产者、多消费者）标志，确保内存池支持跨线程安全操作

3. 拷贝数据包数据而非传递mbuf指针 #

在lcore线程中提取数据包内容并拷贝到独立缓冲区，随后立即释放mbuf，再将拷贝后的数据传递给非lcore线程处理，彻底避免竞态和mbuf跨线程释放问题。示例代码如下：

// 在lcore线程中处理
if (got_num_packets)
{
    IO_Job* job = new IO_Job;
    job->byte_offset = bytes_recorded;
    job->data_len = 0;

    // 计算总数据长度并分配缓冲区
    for (int i = 0; i < got_num_packets; i++) {
        job->data_len += rte_pktmbuf_pkt_len(packets[i]);
    }
    job->data = new char[job->data_len];

    // 拷贝所有数据包内容
    char* dst_ptr = job->data;
    for (int i = 0; i < got_num_packets; i++) {
        char* pkt_data = rte_pktmbuf_mtod(packets[i], char*);
        uint16_t pkt_len = rte_pktmbuf_pkt_len(packets[i]);
        memcpy(dst_ptr, pkt_data, pkt_len);
        dst_ptr += pkt_len;
        // 立即释放当前mbuf
        rte_pktmbuf_free(packets[i]);
    }

    delete[] packets;
    input_arg->ds_instance->enqueue_io_job(job);
}

4. 用rte_ring传递拷贝后的数据 #

利用rte_ring的线程安全特性传递拷贝后的数据包数据，而非直接传递mbuf指针，保证lcore线程可以及时回收mbuf，避免内存池耗尽。

内容的提问来源于stack exchange，提问作者Smitch