DPDK内存管理 -----（二）rte_mempool

DPDK 以两种方式对外提供内存管理方法，一个是 rte_mempool，主要用于网卡数据包的收发；一个是 rte_malloc，主要为应用程序提供内存使用接口。

本文讨论 rte_mempool。rte_mempool 由函数 rte_mempool_create() 负责创建，从 rte_config.mem_config->free_memseg[] 中取出合适大小的内存，放到 rte_config.mem_config->memzone[] 中。

本文中，以 l2fwd 为例，说明 rte_mempool 的创建及使用。

一、rte_mempool的创建

l2fwd_pktmbuf_pool =

rte_mempool_create("mbuf_pool", NB_MBUF,

MBUF_SIZE, 32,

sizeof(structrte_pktmbuf_pool_private),

rte_pktmbuf_pool_init, NULL,

rte_pktmbuf_init, NULL,

rte_socket_id(), 0);

 

“mbuf_pool”：创建的rte_mempool的名称。

NB_MBUF：rte_mempool包含的rte_mbuf元素的个数。

MBUF_SIZE：每个rte_mbuf元素的大小。

 

#defineRTE_PKTMBUF_HEADROOM    128
#defineMBUF_SIZE (2048 + sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM)
#defineNB_MBUF   8192

struct rte_pktmbuf_pool_private {

  uint16_t mbuf_data_room_size; /**< Size of data space in each mbuf.*/

};

rte_mempool 由函数 rte_mempool_create() 负责创建。首先创建 rte_ring，再创建 rte_mempool，并建立两者之间的关联。

1．rte_ring_create()创建rte_ring无锁队列

 

r = rte_ring_create(rg_name, rte_align32pow2(n+1), socket_id, rg_flags);

具体步骤如下：

a、需要保证创建的队列数可以被2整除，即，count = rte_align32pow2(n + 1);

b、计算需要为count个队列分配的内存空间，即，ring_size = count * sizeof(void *) + sizeof(struct rte_ring);

struct rte_ring 的数据结构如下：

struct rte_ring  {

   TAILQ_ENTRY(rte_ring) next;     /**< Next in list. */

   char name[RTE_RING_NAMESIZE];   /**< Name of the ring. */

   int flags;                      /**< Flags supplied at creation. */

   /** Ring producer status. */

   struct prod {

       uint32_t watermark;      /**<Maximum items before EDQUOT. */

       uint32_t sp_enqueue;     /**<True, if single producer. */

       uint32_t size;           /**<Size of ring. */

       uint32_t mask;           /**<Mask (size-1) of ring. */

       volatile uint32_t head;  /**<Producer head. */

       volatile uint32_t tail;  /**<Producer tail. */

    } prod __rte_cache_aligned;

   /** Ring consumer status. */

   struct cons {

       uint32_t sc_dequeue;     /**<True, if single consumer. */

       uint32_t size;           /**<Size of the ring. */

        uint32_t mask;           /**< Mask (size-1) of ring. */

       volatile uint32_t head;  /**<Consumer head. */

       volatile uint32_t tail;  /**<Consumer tail. */

#ifdef RTE_RING_SPLIT_PROD_CONS

    } cons __rte_cache_aligned;

#else

    } cons;

#endif

#ifdef RTE_LIBRTE_RING_DEBUG

   struct rte_ring_debug_stats stats[RTE_MAX_LCORE];

#endif

 

/* < Memory space of ring startshere.

*  notvolatile so need to be careful

*  about compiler re-ordering 

*/

   void * ring[0] __rte_cache_aligned;

};

 

c、调用 rte_memzone_reserve()，在 rte_config.mem_config->free_memseg[] 中查找一个合适的 free_memseg（查找规则是 free_memseg 中剩余内存大于等于需要分配的内存，但是多余的部分是最小的），从该 free_memseg 中分配指定大小的内存，然后将分配的内存记录在 rte_config.mem_config->memzone[] 中。

d、初始化新分配的rte_ring。

r->flags = flags;

r->prod.watermark = count;

r->prod.sp_enqueue = !!(flags &RING_F_SP_ENQ);

r->cons.sc_dequeue = !!(flags &RING_F_SC_DEQ);

r->prod.size = r->cons.size = count;

r->prod.mask = r->cons.mask =count-1;

r->prod.head = r->cons.head = 0;

r->prod.tail = r->cons.tail = 0;

TAILQ_INSERT_TAIL(ring_list, r, next); // 挂到rte_config.mem_config->tailq_head[RTE_TAILQ_RING]队列中

2．创建并初始化rte_mempool

a、计算需要为rte_mempool申请的内存空间。包含：sizeof(struct rte_mempool)、private_data_size，以及n * objsz.total_size。

mempool_size = MEMPOOL_HEADER_SIZE(mp, pg_num) + private_data_size;

if (vaddr == NULL)

mempool_size += (size_t)objsz.total_size * n;

objsz.total_size = objsz.header_size + objsz.elt_size +objsz.trailer_size; 

其中

objsz.header_size = sizeof(struct rte_mempool *);

objsz.elt_size = MBUF_SIZE；

objsz.trailer_size = ????

b、调用rte_memzone_reserve()，在rte_config.mem_config->free_memseg[]中查找一个合适的free_memseg，在该free_memseg中分配mempool_size大小的内存，然后将新分配的内存记录到rte_config.mem_config->memzone[]中。

c、初始化新创建的rte_mempool，并调用rte_pktmbuf_pool_init()初始化rte_mempool的私有数据结构。

/* init the mempool structure */

mp = mz->addr;

memset(mp, 0, sizeof(*mp));

snprintf(mp->name, sizeof(mp->name),"%s", name);

mp->phys_addr = mz->phys_addr;

mp->ring = r;

mp->size = n;

mp->flags = flags;

mp->elt_size = objsz.elt_size;

mp->header_size = objsz.header_size;

mp->trailer_size = objsz.trailer_size;

mp->cache_size = cache_size;

mp->cache_flushthresh = (uint32_t)

(cache_size * CACHE_FLUSHTHRESH_MULTIPLIER);

mp->private_data_size =private_data_size;

/* calculate address of the first elementfor continuous mempool. */

obj = (char *)mp + MEMPOOL_HEADER_SIZE(mp,pg_num) +

private_data_size;

/* populate address translation fields. */

mp->pg_num = pg_num;

mp->pg_shift = pg_shift;

mp->pg_mask =RTE_LEN2MASK(mp->pg_shift, typeof(mp->pg_mask));

/* mempool elements allocated together withmempool */

mp->elt_va_start = (uintptr_t)obj;

mp->elt_pa[0] = mp->phys_addr +

 (mp->elt_va_start - (uintptr_t)mp);

mp->elt_va_end = mp->elt_va_start;

RTE_EAL_TAILQ_INSERT_TAIL(RTE_TAILQ_MEMPOOL,rte_mempool_list, mp); //挂到rte_config.mem_config->tailq_head[RTE_TAILQ_MEMPOOL]队列中

d、调用mempool_populate()，以及rte_pktmbuf_init()初始化rte_mempool的每个rte_mbuf元素。

 

3．总结

相关数据结构的关联关系如下图：



文章来源 
http://www.cnblogs.com/MerlinJ/p/4081986.html