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 512flushthresh = 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四个要点：cache 是栈（LIFO）：注释明说 "The cache is a stack"——刚放回的对象最热，最先被取走（cache 友好）。编译期常量优化：__rte_constant(n) 为真时循环展开——收包路径 nb_pkts=32 这种常量调用会展开成无循环代码。miss 时补货量是 cache->size + remaining：先把缓存补满再满足请求。把对共享 ring 的竞争摊销到每 cache_size 次访问一次。请求巨大（>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 */ };DPDK mempool内核 SLUB差异本质rte_mempool_cache（每 lcore）kmem_cache_cpu（per-CPU freelist）同一哲学：per-CPU 缓存吸收竞争miss → ring 批量补 size+remainingfreelist 空 → 从 per-node partial 摘 slab都是"一次慢路径换多次快路径"无锁 ring 共享池，无锁全程partial 链表 + local_lock_t lock内核慢路径仍要锁；DPDK 靠"绑核不迁移"免锁固定大小对象任意大小内核服务通用分配器，DPDK 只服务 mbufSLUB 的 tid 用于检测抢占迁移（线程被切走又回来时发现 freelist 被别的 CPU 动过）。DPDK 没有这个烦恼——lcore 绑核，线程根本不迁移，这是 EAL 绑核的连锁收益。面试 3 问cache miss 时为什么一次补 cache->size + remaining 个？ → 把共享 ring 的竞争摊销为每 cache_size 次访问一次；对应 rte_mempool.h:1570-1571。flushthresh 为什么是 1.5 倍？ → CALC_CACHE_FLUSHTHRESH = c*3/2（rte_mempool.c:56）；水位 [size, 1.5×size]。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：免锁队列如何可能