P1-02 · mempool：预分配 + 每核缓存，消灭逐包分配

对应路线图：Phase 1 (Day 22-49) ｜ DPDK 24.11.6 LTS ｜ 内核对照 v6.6
代码出处：source/dpdk-stable-24.11.6/lib/mempool/，引用均标 文件:行号
上一篇：P1-01 EAL ｜ 下一篇：P1-03 ring

一句话定位

内核每收一个包要 kmalloc 一个 skb（快路径进 slab per-CPU 缓存，慢路径仍有锁和页分配）。DPDK 的回答：启动时一次性造好 N 个 mbuf 放池里，每核再缓存一批，热路径拿/还只碰自己核的缓存，零竞争。

架构图

真实数据结构（lib/mempool/rte_mempool.h）

/**
 * A structure that stores a per-core object cache.
 */
struct __rte_cache_aligned rte_mempool_cache {   // rte_mempool.h:90
    uint32_t size;          /**< Size of the cache */
    uint32_t flushthresh; /**< Threshold before we flush excess elements */
    uint32_t len;          /**< Current cache count */
    // ... 统计字段（编译开关控制）
    /**
     * Cache objects
     *
     * Cache is allocated to this size to allow it to overflow in certain
     * cases to avoid needless emptying of cache.
     */
    alignas(RTE_CACHE_LINE_SIZE) void *objs[RTE_MEMPOOL_CACHE_MAX_SIZE * 2];
};                                                      // rte_mempool.h:114

注意 objs 数组是 MAX_SIZE * 2——故意留一倍溢出空间，"overflow to avoid needless emptying"：put 时即使超过 flushthresh 也能先落进数组，下次再整批回刷，而不是立即碰后端。

池本体（rte_mempool.h:230-272 节选）：

struct __rte_cache_aligned rte_mempool {
    char name[RTE_MEMPOOL_NAMESIZE];
    union {
        void *pool_data;         /**< Ring or pool to store objects. */
        uint64_t pool_id;        /**< External mempool identifier. */
    };
    unsigned int flags;              /**< Flags of the mempool. */
    int socket_id;                   /**< Socket id passed at create. */
    uint32_t size;                   /**< Max size of the mempool. */
    uint32_t cache_size;             /**< Size of per-lcore default local cache. */
    uint32_t elt_size;               /**< Size of an element. */
    int32_t ops_index;               /**< ops 表索引（ring/stack/lf-stack） */
    struct rte_mempool_cache *local_cache; /**< Per-lcore local cache */
    // ...
};

两个必须记住的常量

// lib/mempool/rte_mempool.c:56
#define CALC_CACHE_FLUSHTHRESH(c) (((c) * 3) / 2)

// config/rte_config.h:61
#define RTE_MEMPOOL_CACHE_MAX_SIZE 512

• flushthresh = cache_size × 1.5：缓存水位保持在 [size, 1.5×size]——既能吸收突发，又不囤积（囤积会让别的核饿死）。初始化在 rte_mempool.c:746-757（mempool_cache_init）。
• 创建时强校验（rte_mempool.c:837-842）：

    /* asked cache too big */
    if (cache_size > RTE_MEMPOOL_CACHE_MAX_SIZE ||
        CALC_CACHE_FLUSHTHRESH(cache_size) > n) {
        rte_errno = EINVAL;
        return NULL;
    }

头文件文档再补一刀（rte_mempool.h:1033-1038）：

 *   RTE_MEMPOOL_CACHE_MAX_SIZE and n / 1.5. It is advised to choose
 *   cache_size to have "n modulo cache_size == 0": if this is
 *   not the case, some elements will always stay in the pool and will
 *   never be used.

——cache_size 必须 ≤ n/1.5（否则回刷阈值超过池容量），且建议整除 n。

put 热路径：逐行读 rte_mempool_do_generic_put

（rte_mempool.h:1377-1426，节选核心逻辑）

static __rte_always_inline void
rte_mempool_do_generic_put(struct rte_mempool *mp, void * const *obj_table,
               unsigned int n, struct rte_mempool_cache *cache)
{
    void **cache_objs;

    /* No cache provided */
    if (unlikely(cache == NULL))
        goto driver_enqueue;
    // ...

    /* The request itself is too big for the cache */
    if (unlikely(n > cache->flushthresh))
        goto driver_enqueue_stats_incremented;   // rte_mempool.h:1392-1393

    /*
     * The cache follows the following algorithm:
     *   1. If the objects cannot be added to the cache without crossing
     *      the flush threshold, flush the cache to the backend.
     *   2. Add the objects to the cache.
     */

    if (cache->len + n <= cache->flushthresh) {
        cache_objs = &cache->objs[cache->len];
        cache->len += n;
    } else {
        cache_objs = &cache->objs[0];
        rte_mempool_ops_enqueue_bulk(mp, cache_objs, cache->len);  // 整批回刷
        cache->len = n;
    }

    /* Add the objects to the cache. */
    rte_memcpy(cache_objs, obj_table, sizeof(void *) * n);
    // ...
driver_enqueue:
    /* push objects to the backend */
    rte_mempool_ops_enqueue_bulk(mp, obj_table, n);
}

三条规则：① 请求太大直接走后端；② 缓存放得下就只改 len（一次 memcpy）；③ 放不下先把旧缓存整批刷回后端，再收下新的。全程无锁——cache 是每核私有的。

get 热路径：逐行读 rte_mempool_do_generic_get

（rte_mempool.h:1506-1590，节选核心逻辑）

    /* The cache is a stack, so copy will be in reverse order. */
    cache_objs = &cache->objs[cache->len];          // :1521-1522

    if (__rte_constant(n) && n <= cache->len) {
        /*
         * The request size is known at build time, and
         * the entire request can be satisfied from the cache,
         * so let the compiler unroll the fixed length copy loop.
         */
        cache->len -= n;
        for (index = 0; index < n; index++)
            *obj_table++ = *--cache_objs;           // :1524-1538
        return 0;
    }

    /* Use the cache as much as we have to return hot objects first. */
    len = __rte_constant(n) ? cache->len : RTE_MIN(n, cache->len);
    cache->len -= len;
    remaining = n - len;
    for (index = 0; index < len; index++)
        *obj_table++ = *--cache_objs;               // :1545-1549
    // ...

    /* if dequeue below would overflow mem allocated for cache */
    if (unlikely(remaining > RTE_MEMPOOL_CACHE_MAX_SIZE))
        goto driver_dequeue;                        // :1566-1567

    /* Fill the cache from the backend; fetch size + remaining objects. */
    ret = rte_mempool_ops_dequeue_bulk(mp, cache->objs,
            cache->size + remaining);               // :1570-1571
    // ...

    /* Satisfy the remaining part of the request from the filled cache. */
    cache_objs = &cache->objs[cache->size + remaining];
    for (index = 0; index < remaining; index++)
        *obj_table++ = *--cache_objs;

    cache->len = cache->size;                       // :1582-1587

四个要点：

1. cache 是栈（LIFO）：注释明说 "The cache is a stack"——刚放回的对象最热，最先被取走（cache 友好）。
2. 编译期常量优化：__rte_constant(n) 为真时循环展开——收包路径 nb_pkts=32 这种常量调用会展开成无循环代码。
3. miss 时补货量是 cache->size + remaining：先把缓存补满再满足请求。把对共享 ring 的竞争摊销到每 cache_size 次访问一次。
4. 请求巨大（>512）直接走后端，绕过缓存——避免缓存被一次性掏空。

🐧 内核对照：SLUB 的 per-CPU 缓存

真实结构（include/linux/slub_def.h:50-66）：

struct kmem_cache_cpu {
    union {
        struct {
            void **freelist;    /* Pointer to next available object */
            unsigned long tid;    /* Globally unique transaction id */
        };
        freelist_aba_t freelist_tid;
    };
    struct slab *slab;    /* The slab from which we are allocating */
#ifdef CONFIG_SLUB_CPU_PARTIAL
    struct slab *partial;    /* Partially allocated frozen slabs */
#endif
    local_lock_t lock;    /* Protects the fields above */
};

SLUB 的 tid 用于检测抢占迁移（线程被切走又回来时发现 freelist 被别的 CPU 动过）。DPDK 没有这个烦恼——lcore 绑核，线程根本不迁移，这是 EAL 绑核的连锁收益。

面试 3 问

1. cache miss 时为什么一次补 cache->size + remaining 个？ → 把共享 ring 的竞争摊销为每 cache_size 次访问一次；对应 rte_mempool.h:1570-1571。
2. flushthresh 为什么是 1.5 倍？ → CALC_CACHE_FLUSHTHRESH = c*3/2（rte_mempool.c:56）；水位 [size, 1.5×size]。
3. objs 数组为什么开 MAX×2？ → 允许溢出暂存，"avoid needless emptying of cache"（rte_mempool.h:110-113 注释）。

🔧 动手实验

写 code/P1/mempool-bench/：8 核并发 alloc/free，cache_size = 0 / 128 / 512 对比吞吐与延迟，出火焰图。cache=0 时 CAS 竞争会给出这道题的标准答案。 数据落点 perf-data/P1/。

下一篇：P1-03 · ring：免锁队列如何可能