前言：DPDK的內存管理工作主要分布在幾個大的部分：大頁初始化與管理，內存管理。使用大頁可以減少頁表開銷，是為了盡量減少TBL miss導致的性能損失。基於大頁，DPDK又進一步細化管理這部分內存，使得分配，回收更加方便。
一.內存管理的對象說明
1.1. 從大頁(hugepage)說起
linux內存是按照頁來划分的，默認的每頁為4K大小，對應的就存在頁表（TBL）來記錄每個頁的地址等該單元的信息。這樣就存在一個問題，當訪問的內容不在本頁時，就會觸發 tbl miss,導致頁換出換入，很影響性能。而一個解決辦法就是使用hugepage，大頁的每頁大小可以設置，常用設置如2M，1G等，比如1G大小的內存，使用4k的頁面，需要256K個，而使用1G的大頁，只需要一個。這樣子就能大大減少tbl miss的概率。 更加詳細的大頁的相關內容，請參考下面的鏈接：
http://www.tuicool.com/articles/vYZJ3i3
二. DPDK內存的初始化
內存的初始化在rte_eal_init()中完成，由於DPDK的進程分為primary和secondary，內存的初始化工作只能在primory進程中完成。主要的步驟如下：
eal_hugepage_info_init()；獲取大頁的信息，並初始化內部的結構。rte_config_init();創建配置文件,並做內存映射。rte_eal_memory_init()；大頁的內存初始化，並連接成連續的內存區。rte_eal_memzone_init()；初始化memzone子系統。
2.1 eal_hugepage_info_init()
這一步是獲取系統中已配置的大頁的信息，以及大頁的掛載點（在DPDK的參數中可以指定大頁的掛載點，默認應該是/mnt/huge）。
dir = opendir(sys_dir_path);先打開"/sys/kernel/mm/hugepages"目錄，讀取系統中的大頁目錄，存儲在internal_config.hugepage_info[]結構中，頁面的大小在目錄名中。然后獲取大頁的大小和掛載點：
hpi = &internal_config.hugepage_info[num_sizes];
hpi->hugepage_sz =
rte_str_to_size(&dirent->d_name[dirent_start_len]);
hpi->hugedir = get_hugepage_dir(hpi->hugepage_sz);
最后獲取空閑頁面數量,並且都先放在第一個核上：
hpi->num_pages[0] = get_num_hugepages(dirent->d_name);
可以通過設置MAX_HUGEPAGE_SIZES宏的值來調整DPDK允許配置的大頁的頁面值個數。默認是3個。
之后，把這些大頁按照大小順序排一下序，大的頁面在前面。
qsort(&internal_config.hugepage_info[0], num_sizes,
sizeof(internal_config.hugepage_info[0]), compare_hpi);
最后做一下檢查，這樣，對於大頁的信息的獲取就做完了。
2.2 rte_config_init()
因為DPDK支持primary進程和secondary進程，他們都需要內存的配置信息，進程間通信使用了共享內存的方法，把struct rte_mem_config *mem_config結構做內存映射。
switch (rte_config.process_type){
case RTE_PROC_PRIMARY:
rte_eal_config_create();
break;
case RTE_PROC_SECONDARY:
rte_eal_config_attach();
rte_eal_mcfg_wait_complete(rte_config.mem_config);
rte_eal_config_reattach();
break;
開始就根據進程的類型決定啟動順序問題，如果是primary進程，下面看看他的處理過程：
if (internal_config.base_virtaddr != 0)
rte_mem_cfg_addr = (void *)
RTE_ALIGN_FLOOR(internal_config.base_virtaddr -
sizeof(struct rte_mem_config), sysconf(_SC_PAGE_SIZE));
else
rte_mem_cfg_addr = NULL;
if (mem_cfg_fd < 0){
mem_cfg_fd = open(pathname, O_RDWR | O_CREAT, 0660);
if (mem_cfg_fd < 0)
rte_panic("Cannot open '%s' for rte_mem_config\n", pathname);
}
retval = ftruncate(mem_cfg_fd, sizeof(*rte_config.mem_config));
if (retval < 0){
close(mem_cfg_fd);
rte_panic("Cannot resize '%s' for rte_mem_config\n", pathname);
}
先根據啟動的參數選擇內存配置文件共享內存開始的地址，如果配置了base_viraddr，這個地址應該是可以指定大頁開始的地址。在大頁開始地址的前面映射內存配置文件。然后打開內存配置文件，裁剪大小。
然后選擇地址，映射sizeof(*rte_config.mem_config)大小的內存到內存配置文件。
rte_config.mem_config = (struct rte_mem_config *) rte_mem_cfg_addr;
/* store address of the config in the config itself so that secondary
* processes could later map the config into this exact location */
rte_config.mem_config->mem_cfg_addr = (uintptr_t) rte_mem_cfg_addr;
填充映射后的地址，這里最后一句比較有意思，把primary進程中映射的地址保存下來，后面我們就會看到，是為了讓secondary進程也映射同樣的邏輯地址。
接下來就看看secondary進程的地址映射情況：
首先，做了一個attach操作，就是先對共享文件做了映射，記錄了映射后的地址。
rte_eal_config_attach()
之后，就等待primary進程完整eal層的初始化完成。等初始化完成后，魔數就會填充，rte_eal_mcfg_complete()。secondary進程會再次進行內存映射，這次映射的目的就是使得secondary進程中對內存配置文件映射后的邏輯地址和primary進程一樣，這樣做有什么好處我們后面再仔細說。
rte_mem_cfg_addr = (void *) (uintptr_t) rte_config.mem_config->mem_cfg_addr;
munmap(rte_config.mem_config, sizeof(struct rte_mem_config));
mem_config = (struct rte_mem_config *) mmap(rte_mem_cfg_addr,
sizeof(*mem_config), PROT_READ | PROT_WRITE, MAP_SHARED,
mem_cfg_fd, 0);
最后需要說明的一點是：在DPDK中，創建的mempool，ring等可以在多個進程間訪問，也是因為在rte_config.mem_config中有個成員是struct rte_tailq_head tailq_head[RTE_MAX_TAILQ]，創建的ring等隊列頭都是掛在其中，是通過構造函數在main函數之前就掛接上的。
2.3 rte_eal_memory_init()
這個函數是初始化內存子系統，任務很多，對於primary進程，則映射大頁內存，而對於secondary進程，則把大頁attach到primary進程。
2.3.1 rte_eal_hugepage_init()
這就是在primary進程中進行大頁的映射。非常有趣，來看看他的主要工作吧！下面直接引用函數原型中的說明:
/*
* Prepare physical memory mapping: fill configuration structure with
* these infos, return 0 on success.
* 1. map N huge pages in separate files in hugetlbfs
* 2. find associated physical addr
* 3. find associated NUMA socket ID
* 4. sort all huge pages by physical address
* 5. remap these N huge pages in the correct order
* 6. unmap the first mapping
* 7. fill memsegs in configuration with contiguous zones
*/
首先，獲取全局的配置信息：
mcfg = rte_eal_get_configuration()->mem_config;
這里比較有意思的地方是，primary進程和secondary進程中配置信息映射的邏輯地址是一樣的。
然后獲取當前使用的大頁的大小和頁數。
for (i = 0; i < (int) internal_config.num_hugepage_sizes; i++) {
/* meanwhile, also initialize used_hp hugepage sizes in used_hp */
used_hp[i].hugepage_sz = internal_config.hugepage_info[i].hugepage_sz;
nr_hugepages += internal_config.hugepage_info[i].num_pages[0];
}
分配大頁頁表，
tmp_hp = malloc(nr_hugepages * sizeof(struct hugepage_file));
if (tmp_hp == NULL)
goto fail;
memset(tmp_hp, 0, nr_hugepages * sizeof(struct hugepage_file));
然后就到了非常重要的一步：內存映射大頁。主要分為三步
- 第一次映射大頁。
- 按大頁的物理地址重新排序。
- 第二次映射大頁。
先看第一次映射大頁：map_all_hugepages(&tmp_hp[hp_offset], hpi, 1)，最后一個參數就是指明是第一次映射。由於是第一次映射，所以，先填充大頁的文件信息
if (orig) {
hugepg_tbl[i].file_id = i;
hugepg_tbl[i].size = hugepage_sz;
eal_get_hugefile_path(hugepg_tbl[i].filepath,
sizeof(hugepg_tbl[i].filepath), hpi->hugedir,
hugepg_tbl[i].file_id);
hugepg_tbl[i].filepath[sizeof(hugepg_tbl[i].filepath) - 1] = '\0';
}
之后，就在/mnt/huge目錄下創建每個大頁文件，並映射每個大頁到內存中。為什么是/mnt/huge目錄？因為這是掛載大頁文件系統的位置，掛載大頁文件系統，可以通過 mount -t hugetlbfs nodev /mnt/huge來掛載。
fd = open(hugepg_tbl[i].filepath, O_CREAT | O_RDWR, 0600);
if (fd < 0) {
RTE_LOG(DEBUG, EAL, "%s(): open failed: %s\n", __func__,
strerror(errno));
return i;
}
/* map the segment, and populate page tables,
* the kernel fills this segment with zeros */
virtaddr = mmap(vma_addr, hugepage_sz, PROT_READ | PROT_WRITE,
MAP_SHARED | MAP_POPULATE, fd, 0);
在這里，新創建的大頁文件並沒有大小，但是在映射后，文件大小就變成了映射的大小，貌似只能映射頁大小的整數倍。
第一次映射，填充orig_va地址：
hugepg_tbl[i].orig_va = virtaddr;
然后計算下一個頁面映射的地址：
vma_addr = (char *)vma_addr + hugepage_sz;
等把所有的頁面都映射完了后，這部分對應的物理內存就不會被換出到磁盤。此時，我們映射的這部分內存，邏輯地址是連續的，但是物理地址不一定是連續的。
接下來查找已經映射的每個大頁的物理地址，並填充其結構。
find_physaddrs()
具體的虛擬地址到物理地址的查找關系
rte_mem_virt2phy()
然后找到映射的大頁內存被放在哪個NUMA node上。
if (find_numasocket(&tmp_hp[hp_offset], hpi) < 0){
RTE_LOG(DEBUG, EAL, "Failed to find NUMA socket for %u MB pages\n",
(unsigned)(hpi->hugepage_sz / 0x100000));
goto fail;
}
把映射的大頁的物理地址按照從小到大的順序進行排序。
qsort(&tmp_hp[hp_offset], hpi->num_pages[0],
sizeof(struct hugepage_file), cmp_physaddr);
接下來就是第二次對大頁進行映射：
if (map_all_hugepages(&tmp_hp[hp_offset], hpi, 0) !=
hpi->num_pages[0])
這里我們看到最后一個參數就已經是0了。
這樣進來函數之后，第一個循環時，vma_len就是0，然后就去查找物理地址連續的頁：
for (j = i+1; j < hpi->num_pages[0] ; j++) {
#ifdef RTE_ARCH_PPC_64
/* The physical addresses are sorted in
* descending order on PPC64 */
if (hugepg_tbl[j].physaddr !=
hugepg_tbl[j-1].physaddr - hugepage_sz)
break;
#else
if (hugepg_tbl[j].physaddr !=
hugepg_tbl[j-1].physaddr + hugepage_sz)
break;
#endif
}
num_pages = j - i;
vma_len = num_pages * hugepage_sz;
這樣，就能確定連續的物理頁有幾個，然后，去嘗試分配和連續物理頁一樣大的虛擬地址空間，如果不能，就減小一個頁再嘗試，直到成功（返回地址）或者失敗（返回NULL）。如果能拿到地址，那么就以這個地址開始，依次映射物理地址連續的幾個頁。如果不能拿到這么大且連續的虛擬地址，那么，就讓內核自己去分配地址，然后映射這一頁。
第二次映射后，就填充final_va地址了：hugepg_tbl[i].final_va = virtaddr;。
既然已經重新映射了大頁的虛擬地址，那么就應該撤銷原來的映射。
if (unmap_all_hugepages_orig(&tmp_hp[hp_offset], hpi) < 0)
goto fail;
這樣過后，對於大頁內存的映射工作就完成了。
接下來就是分配映射的大頁內存咯。
首先，清空配置信息中的每個socket中大頁的數量，等待重新分配。
for (i = 0; i < (int)internal_config.num_hugepage_sizes; i++)
for (j = 0; j < RTE_MAX_NUMA_NODES; j++)
internal_config.hugepage_info[i].num_pages[j] = 0;
然后獲取每個socket上的大頁數量，
for (i = 0; i < nr_hugefiles; i++) {
int socket = tmp_hp[i].socket_id;
/* find a hugepage info with right size and increment num_pages */
const int nb_hpsizes = RTE_MIN(MAX_HUGEPAGE_SIZES,
(int)internal_config.num_hugepage_sizes);
for (j = 0; j < nb_hpsizes; j++) {
if (tmp_hp[i].size ==
internal_config.hugepage_info[j].hugepage_sz) {
internal_config.hugepage_info[j].num_pages[socket]++;
}
}
}
重新計算調整每個socket上的大頁的分布，最后返回大頁個數。
nr_hugepages = calc_num_pages_per_socket(memory,
internal_config.hugepage_info, used_hp,
internal_config.num_hugepage_sizes);
默認每個socket上的大頁數量是按核心數量的比例分配的。
然后為大頁映射信息文件創建共享內存，用於secondary進程來映射地址。
先撤銷不用的大頁映射，然后把臨時大頁信息文件拷貝到創建的共享內存中。
if (unmap_unneeded_hugepages(tmp_hp, used_hp,
internal_config.num_hugepage_sizes) < 0) {
RTE_LOG(ERR, EAL, "Unmapping and locking hugepages failed!\n");
goto fail;
}
if (copy_hugepages_to_shared_mem(hugepage, nr_hugefiles,
tmp_hp, nr_hugefiles) < 0) {
RTE_LOG(ERR, EAL, "Copying tables to shared memory failed!\n");
goto fail;
}
最后把大頁內存切成段保存在內存管理結構中。大頁內存切段的條件是：
- 不在同一個socket上。
- 頁的大小不相同
- 物理地址不連續
- 虛擬地址不連續
然后把切好的內存段放入mcfg配置表中：
mcfg->memseg[j].phys_addr = hugepage[i].physaddr;
mcfg->memseg[j].addr = hugepage[i].final_va;
mcfg->memseg[j].len = hugepage[i].size;
mcfg->memseg[j].socket_id = hugepage[i].socket_id;
mcfg->memseg[j].hugepage_sz = hugepage[i].size;
這樣，大頁的初始化就完成了！
2.3.2 rte_eal_hugepage_attach()
對於secondary進程而言，它並不能創建大頁的共享內存，而只能attach上去。
開始大頁內存attach的前提是先attach內存配置文件，我們再來看一下attach配置的過程：
rte_eal_config_attach();
rte_eal_mcfg_wait_complete(rte_config.mem_config);
rte_eal_config_reattach();
第一個函數中，先映射一下/var/run/.rte_config文件，拿到內存配置的結構信息，就是為了第二個函數的等待判斷用的。第三個函數中，既然主進程已經初始化完成，那么，就先解除第一個函數的映射，以primary進程中映射的內存配置文件地址作為新的映射地址，重新映射，映射完成后，primary進程和secondary進程中，對於/var/run/.rte_config映射的虛擬地址是一樣的。（雖然，對於配置文件映射地址一樣，感覺並沒什么卵用~，但后面的大頁映射也是這么做的，映射地址的一致，就有用啦）。
接下來就來看大頁內存的attach，首先打開/dev/zero文件，按照primary的段的虛擬地址依次映射所有的內存段，這一步相當於先測試一下是否能分配這樣的連續地址空間。
base_addr = mmap(mcfg->memseg[s].addr, mcfg->memseg[s].len,
PROT_READ, MAP_PRIVATE, fd_zero, 0);
然后映射大頁信息共享文件/var/run/.rte_hugepage_info，並計算頁個數等。
size = getFileSize(fd_hugepage);
hp = mmap(NULL, size, PROT_READ, MAP_PRIVATE, fd_hugepage, 0);
if (hp == MAP_FAILED) {
RTE_LOG(ERR, EAL, "Could not mmap %s\n", eal_hugepage_info_path());
goto error;
}
num_hp = size / sizeof(struct hugepage_file);
最后解除映射到/dev/zero，重新映射到各個大頁文件中，
for (i = 0; i < num_hp && offset < mcfg->memseg[s].len; i++){
if (hp[i].memseg_id == (int)s){
fd = open(hp[i].filepath, O_RDWR);
if (fd < 0) {
RTE_LOG(ERR, EAL, "Could not open %s\n",
hp[i].filepath);
goto error;
}
mapping_size = hp[i].size;
addr = mmap(RTE_PTR_ADD(base_addr, offset),
mapping_size, PROT_READ | PROT_WRITE,
MAP_SHARED, fd, 0);
close(fd); /* close file both on success and on failure */
if (addr == MAP_FAILED ||
addr != RTE_PTR_ADD(base_addr, offset)) {
RTE_LOG(ERR, EAL, "Could not mmap %s\n",
hp[i].filepath);
goto error;
}
offset+=mapping_size;
}
}
到這里我們仔細看一下，進程中是以primary中的虛擬地址作為映射地址來映射的，也就是說在映射完成后，primary進程和secondary進程中映射的大頁地址是一樣的。這很關鍵，這正是實現零拷貝的原理。虛擬地址一樣，那么從大頁內存中拿到的數據包，就可以不經過拷貝，直接把地址傳到secondary進程中。
這些都映射完了后，就完成了attach工作。
2.4 rte_eal_memzone_init()
memzone是內存分配器，上一步中，我們已經把大頁內存分段放好了，但是在使用的時候，怎么來分配呢？自然需要內存分配器，就是memzone。而memzone_init主要就是把內存放到空閑鏈表中，等需要的時候，能夠分配出去。
在看初始化前，先看一個結構，struct malloc_elem，這個結構表示一個內存對象，
struct malloc_elem {
struct malloc_heap *heap;
struct malloc_elem *volatile prev; /* points to prev elem in memseg */
LIST_ENTRY(malloc_elem) free_list; /* list of free elements in heap */
const struct rte_memseg *ms;
volatile enum elem_state state;
uint32_t pad;
size_t size;
#ifdef RTE_LIBRTE_MALLOC_DEBUG
uint64_t header_cookie; /* Cookie marking start of data */
/* trailer cookie at start + size */
#endif
} __rte_cache_aligned;
然后看初始化
rte_eal_malloc_heap_init()
依次把每一段都添加到heap中，段屬於哪個socket，就添加到哪個socket的heap中，分配就從這里拿。
for (ms = &mcfg->memseg[0], ms_cnt = 0;
(ms_cnt < RTE_MAX_MEMSEG) && (ms->len > 0);
ms_cnt++, ms++) {
malloc_heap_add_memseg(&mcfg->malloc_heaps[ms->socket_id], ms);
}
把每一段做初始化，並掛在空閑鏈表中:
malloc_elem_init(start_elem, heap, ms, elem_size);
malloc_elem_mkend(end_elem, start_elem);
malloc_elem_free_list_insert(start_elem);
heap->total_size += elem_size;
然后就初始化完了！
三. DPDK內存的分配
內存分配有一系列的接口：大多定義在rte_malloc.c文件中。我們重點挑兩個來看一下。
rte_malloc_socket()
這個是一個基礎函數，可以在這個函數的基礎上進行封裝，主要參數是類型，大小，對齊，以及從哪個socket上分。一般來說，分配內存從當前線程運行的socket上分配，可以避免內存跨socket訪問，提高性能。
ret = malloc_heap_alloc(&mcfg->malloc_heaps[socket], type,
size, 0, align == 0 ? 1 : align, 0);
if (ret != NULL || socket_arg != SOCKET_ID_ANY)
return ret;
先在指定的socket上分配，如果不能成功，再去嘗試其他的socket。我們接着看函數malloc_heap_alloc():
void *
malloc_heap_alloc(struct malloc_heap *heap,
const char *type __attribute__((unused)), size_t size, unsigned flags,
size_t align, size_t bound)
{
struct malloc_elem *elem;
size = RTE_CACHE_LINE_ROUNDUP(size);
align = RTE_CACHE_LINE_ROUNDUP(align);
rte_spinlock_lock(&heap->lock);
elem = find_suitable_element(heap, size, flags, align, bound);
if (elem != NULL) {
elem = malloc_elem_alloc(elem, size, align, bound);
/* increase heap's count of allocated elements */
heap->alloc_count++;
}
rte_spinlock_unlock(&heap->lock);
return elem == NULL ? NULL : (void *)(&elem[1]);
先去空閑鏈表中找是否有滿足需求的內存塊，如果找到，就進行分配，否則返回失敗。進一步的，在函數malloc_elem_alloc()分配的的時候，如果存在的內存大於需要的內存時，會對內存進行切割，然后把用不完的重新掛在空閑鏈表上。就不細致的代碼分析了。
rte_memzone_reserve_aligned()
這個函數的返回值類型是struct rte_memzone型的，這是和上一個分配接口的不同之處，同時注意分配時的flag的不同。分配出來的memzone可以直接使用名字索引到。這個函數最終也是會調用到malloc_heap_alloc(),就不多說了，可以看看代碼。
除此以外，需要額外提到的內存分配的地方是創建內存池。在創建內存池時，會創建一個ring來存儲分配的對象，同時，為了減少多核之間對同一個ring的訪問，每一個核都維護着一個cache，優先從cache中取。
四. DPDK內存的回收
說完了DPDK的內存分配，最后來說一下內存回收。跟分配的接口對應，也有多個回收函數。
-
rte_free()
同樣這個函數，在上層封裝了多種接口。如rte_memzone_free()等。主要的過程也是重新把elem放進free list上，如果有能夠合並的，還會對其進行合並。 -
rte_memzone_free()
上面都說過了，這個里面也是對rte_free()的封裝，不多說了，just see the code!
同樣，關於回收也有點注意的，對於內存池中的元素的回收，不是釋放回空閑鏈表，而是重新放到ring或者cache中，就這么多了。