dpdk mempool之概念学习
dpdk mempool学习总结
概序
dpdk的mempool是预先分配预固定大小的内存池,它可以通过名字来表示不同的内存池,通过注册不同钩子函数来申请、释放内存对象。默认的handler是无锁ring库来实现的。同时,它也提供了一些其它特性:
-
每个core独立的对象缓存区local_cache,用来减少多核访问造成的冲突(无锁ring需要用到cas机制,所以频繁使用,对系统性能还是有比较大的开销);
-
内存对其填充,确保内存池的obj可以在所有DRAM或DDR3通道上均匀地分布
调试支持特性
-
当开启CONFIG_RTE_LIBRTE_MEMPOOL_DEBUG的时候,创建mempool的时候,系统会在每个内存obj开始和结尾多申请内存保护空间,防止缓冲区操作溢出。
-
统计,当开启 CONFIG_RTE_LIBRTE_MEMPOOL_DEBUG的时候,会在mempool结构体里面增加对内存obj put/get的统计,统计是在每一个core单独统计的,减少对统计变量累加的竞争
内存对其
根据硬件内存的配置,在obj之间增加pad填充可以大大的提高性能。目标就是确保每一个obj的地址是在内存的不同的ranks和channel开始,这样能确保所有channel均等被加载。
这里内存的ranks和channel需要根据系统硬件来确定,具体到dpdk启动参数里面可以配置internal_config.force_nchannel和internal_config.force_nrank,具体使用-n和-r.
其实这里我的理解说的通俗点,就是保证内存池里面的每一个obj根据DIMM的ranks和channel来对其。
具体在dpdk代码里面体现实现:
rte_mempool_create flags的MEMPOOL_F_NO_SPREAD来控制,设置这个flag,则关闭内存对其,否则开启
rte_mempool_create_empty
rte_mempool_calc_obj_size
uint32_t
rte_mempool_calc_obj_size(uint32_t elt_size, uint32_t flags,
struct rte_mempool_objsz *sz)
{
.
.
.
/*
* increase trailer to add padding between objects in order to
* spread them across memory channels/ranks
*/
if ((flags & MEMPOOL_F_NO_SPREAD) == 0) {
unsigned new_size;
new_size = optimize_object_size(sz->header_size + sz->elt_size +
sz->trailer_size);
sz->trailer_size = new_size - sz->header_size - sz->elt_size;
}
/* this is the size of an object, including header and trailer */
sz->total_size = sz->header_size + sz->elt_size + sz->trailer_size;
return sz->total_size;
}
Local Cache
为了减少多核访问造成的冲突,引入了local_cache对象缓冲区。该local_cache非硬件上的cache,而是为了减少多核访问ring造成的临界区访问,coreX app会优先访问该local_cache上的对象。
入队的时候优先入local_cache中,出队的时候优先从local_cache中出队.
创建管理api
struct rte_mempool_cache *rte_mempool_cache_create(uint32_t size, int socket_id)
void rte_mempool_cache_free(struct rte_mempool_cache *cache)
static __rte_always_inline void rte_mempool_cache_flush(struct rte_mempool_cache *cache,
从内存池中获取和加入obj
static __rte_always_inline void rte_mempool_generic_put(struct rte_mempool *mp, void * const *obj_table,
static __rte_always_inline int rte_mempool_generic_get(struct rte_mempool *mp, void **obj_table, unsigned int n, struct rte_mempool_cache *cache)
static __rte_always_inline struct rte_mempool_cache *rte_mempool_default_cache(struct rte_mempool *mp, unsigned lcore_id)
内存池钩子函数
- 注册
int rte_mempool_register_ops(const struct rte_mempool_ops *h)
把struct rte_mempool_ops填充好后,保存到全局变量rte_mempool_ops_table里面,下次获取的时候,根据name来查找即可。
#define MEMPOOL_REGISTER_OPS(ops) \
void mp_hdlr_init_##ops(void); \
void __attribute__((constructor, used)) mp_hdlr_init_##ops(void)\
{ \
rte_mempool_register_ops(&ops); \
}
这里通过constructor实现了 MEMPOOL_REGISTER_OPS,让gcc在main函数调用之前,提前调用完成这一段的注册。
dpdk提供了mp sp mc sc四个ops的函数钩子,具体如下
static const struct rte_mempool_ops ops_mp_mc = {
.name = "ring_mp_mc",
.alloc = common_ring_alloc,
.free = common_ring_free,
.enqueue = common_ring_mp_enqueue,
.dequeue = common_ring_mc_dequeue,
.get_count = common_ring_get_count,
};
static const struct rte_mempool_ops ops_sp_sc = {
.name = "ring_sp_sc",
.alloc = common_ring_alloc,
.free = common_ring_free,
.enqueue = common_ring_sp_enqueue,
.dequeue = common_ring_sc_dequeue,
.get_count = common_ring_get_count,
};
static const struct rte_mempool_ops ops_mp_sc = {
.name = "ring_mp_sc",
.alloc = common_ring_alloc,
.free = common_ring_free,
.enqueue = common_ring_mp_enqueue,
.dequeue = common_ring_sc_dequeue,
.get_count = common_ring_get_count,
};
static const struct rte_mempool_ops ops_sp_mc = {
.name = "ring_sp_mc",
.alloc = common_ring_alloc,
.free = common_ring_free,
.enqueue = common_ring_sp_enqueue,
.dequeue = common_ring_mc_dequeue,
.get_count = common_ring_get_count,
};
MEMPOOL_REGISTER_OPS(ops_mp_mc);
MEMPOOL_REGISTER_OPS(ops_sp_sc);
MEMPOOL_REGISTER_OPS(ops_mp_sc);
MEMPOOL_REGISTER_OPS(ops_sp_mc);
到这里位置dpdk就完成了对钩子函数的注册动作,后面具体使用哪个钩子根据名字来设置。如下所示:
- 通过name设置获取ops
上面在dpdk进入main之前完成了对ops的注册,那么我们要使用的时候,应该怎么操作呢,怎么获取到指定的hander?
这里dpdk提供rte_mempool_set_ops_byname来设置,根据名字来选择想要的hander,如下所示。
if ((flags & MEMPOOL_F_SP_PUT) && (flags & MEMPOOL_F_SC_GET))
ret = rte_mempool_set_ops_byname(mp, "ring_sp_sc", NULL);
else if (flags & MEMPOOL_F_SP_PUT)
ret = rte_mempool_set_ops_byname(mp, "ring_sp_mc", NULL);
else if (flags & MEMPOOL_F_SC_GET)
ret = rte_mempool_set_ops_byname(mp, "ring_mp_sc", NULL);
else
ret = rte_mempool_set_ops_byname(mp, "ring_mp_mc", NULL);
这里dpdk也提供另外一个api来提供一个默认的ops name给mbuf使用,
/* Return mbuf pool ops name */
const char *
rte_mbuf_best_mempool_ops(void)
{
/* User defined mempool ops takes the priority */
const char *best_ops = rte_mbuf_user_mempool_ops();
if (best_ops)
return best_ops;
/* Next choice is platform configured mempool ops */
best_ops = rte_mbuf_platform_mempool_ops();
if (best_ops)
return best_ops;
/* Last choice is to use the compile time config pool */
return RTE_MBUF_DEFAULT_MEMPOOL_OPS;
}
这里的配置主要使用dpdk的配置文件,在config/common_base配置修改即可,如下
CONFIG_RTE_LIBRTE_MBUF=y
CONFIG_RTE_LIBRTE_MBUF_DEBUG=n
CONFIG_RTE_MBUF_DEFAULT_MEMPOOL_OPS="ring_mp_mc"
CONFIG_RTE_MBUF_REFCNT_ATOMIC=y
CONFIG_RTE_PKTMBUF_HEADROOM=128