無論對於DPDK做怎樣的增量開發,了解DPDK的內存管理模式有利於在滿足自己產品功能同時最大化的提高性能;
一:Mempool的基本單元概念(https://doc.dpdk.org/guides/prog_guide/mempool_lib.html)
Mempool是固定大小的對象分配器。 在DPDK中,它由名稱唯一標識,並且使用mempool操作來存儲空閑對象。Mempool的組織是通過三個部分實現的:
- mempool對象節點:mempool的對象掛接在 static struct rte_tailq_elem rte_mempool_tailq 全局隊列中,可以通過名字進行唯一標識符;此隊列只是mempool的一個對象指示結構,並不是實際的內存區;
- mempool實際內存區: struct rte_memzone 是實際分配的連續內存空間,存儲所創建的mempool對象;
- ring無鎖隊列:作為一個無鎖環形隊列 struct rte_ring ,存儲着mempool對象的指針,提供了方便存取使用mempool的空間的辦法。
二:一般結構
如圖所示,mempool的對象通過與ring無鎖隊列建立關聯方便存取;同時,為了減少多核訪問造成的沖突,引入了local_cache對象緩沖區。該local_cache非硬件上的cache,而是為了減少多核訪問ring造成的臨界區訪問,
coreX app會優先訪問該local_cache上的對象。入隊的時候優先入local_cache中,出隊的時候優先從local_cache中出隊。
三:mempool的創建和使用
先注意一下 rte_mempool_create 的參數中的兩個 mp_init 和 obj_init ,前者負責初始化mempool中配置的私有參數,如在數據包中加入的我們自己的私有結構;后者負責初始化每個mempool對象。我們然后按照mempool的3個關鍵部分展開說明。
(1)mempool頭結構的創建
mempool頭結構包含3個部分: struct rte_mempool , struct rte_mempool_cache 和mempool private。創建是在 rte_mempool_create_empty() 中完成的,看這個函數,先進行了對齊的檢查
RTE_BUILD_BUG_ON((sizeof(struct rte_mempool) & RTE_CACHE_LINE_MASK) != 0); RTE_BUILD_BUG_ON((sizeof(struct rte_mempool_cache) & RTE_CACHE_LINE_MASK) != 0);
然后從mempool隊列中取出頭節點,我們創建的mempool結構填充好,就掛接在這個節點上。接下來做一些檢查工作和創建flag的設置。
rte_mempool_calc_obj_size() 計算了每個obj的大小,這個obj又是由三個部分組成的,header_size、elt_size、trailer_size,即頭,數據區,尾。在沒有開啟RTE_LIBRTE_MEMPOOL_DEBUG調試時,沒有尾部分;頭部分的結構為: struct rte_mempool_objhdr ,
通過這個頭部,mempool中的obj都是鏈接到隊列中的,所以,提供了遍歷obj的方式(盡管很少這么用)。函數返回最后計算對齊后的obj的大小,為后面分配空間提供依據。
然后分配了一個mempool隊列條目,為后面掛接在隊列做准備。
/* try to allocate tailq entry */ te = rte_zmalloc("MEMPOOL_TAILQ_ENTRY", sizeof(*te), 0); if (te == NULL) { RTE_LOG(ERR, MEMPOOL, "Cannot allocate tailq entry!\n"); goto exit_unlock; }
接下來,就是計算整個mempool頭結構多大。
mempool_size = MEMPOOL_HEADER_SIZE(mp, cache_size); mempool_size += private_data_size; mempool_size = RTE_ALIGN_CEIL(mempool_size, RTE_MEMPOOL_ALIGN);
這里指的是計算mempool的頭結構的大小。而不是內存池實際的大小。在這里可以清晰的看出這個mempool頭結構是由三部分組成的。cache計算的是所有核上的cache之和。
然后,使用 rte_memzone_reserve() 分配這個mempool頭結構大小的空間,填充mempool結構體,並把mempool頭結構中的cache地址分配給mempool。初始化這部分cache。
最后就是掛接mempool結構。 TAILQ_INSERT_TAIL(mempool_list, te, next); (這里上了鎖?)。
(2)mempool實際空間的創建
這部分的創建是在函數 rte_mempool_populate_default(struct rte_mempool *mp) 中完成的。
首先計算為這些元素需要分配多大的空間, rte_mempool_ops_calc_mem_size()
接着 rte_memzone_reserve_aligned() 分配空間。把元素添加到mempool,實際上就是把申請的空間分給每個元素。
(3)ring的創建
先看到的是這么一段代碼:
static int mempool_ops_alloc_once(struct rte_mempool *mp) { int ret; /* create the internal ring if not already done */ if ((mp->flags & MEMPOOL_F_POOL_CREATED) == 0) { ret = rte_mempool_ops_alloc(mp); if (ret != 0) return ret; mp->flags |= MEMPOOL_F_POOL_CREATED; } return 0; }
這就是創建ring的過程咯,其中的函數rte_mempool_ops_alloc()就是實現。那么,對應的ops->alloc()在哪注冊的呢?
/* * Since we have 4 combinations of the SP/SC/MP/MC examine the flags to * set the correct index into the table of ops structs. */ 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);
就是根據ring的類型,來注冊對應的操作函數,如默認的就是ring_mp_mc,多生產者多消費者模型,其操作函數不難找到:
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, };
接下來,又分配了一個 struct rte_mempool_memhdr *memhdr; 結構的變量,就是這個變量管理着mempool的實際內存區,它記錄着mempool實際地址區的物理地址,虛擬地址,長度等信息。
再然后,就是把每個元素對應到mempool池中了: mempool_add_elem() 。在其中,把每個元素都掛在了elt_list中,可以遍歷每個元素。最后 rte_mempool_ops_enqueue_bulk(struct rte_mempool *mp, void * const *obj_table, ,最終,把元素對應的地址入隊,這樣,mempool中的每個元素都放入了ring中。
四:mempool的使用及實踐
mempool的常見使用是獲取元素空間和釋放空間。
待補充
部分內容轉載自:https://www.cnblogs.com/yhp-smarthome/p/6687175.html