2_高并發內存池_各層級的框架設計及ThreadCache（線程緩存）申請內存設計

一、高并發內存池框架設計

高并發池框架設計，特別是針對內存池的設計，需要充分考慮多線程環境下：

性能問題
鎖競爭問題
內存碎片問題

高并發內存池的整體框架設計旨在提高內存的申請和釋放效率，減少鎖競爭和內存碎片。

高并發內存池設計的基本框架包括**ThreadCache（線程緩存）、CentralCache（中心緩存）和PageCache（頁緩存）**三個主要部分。主要組成部分包括：

接下來，我們先粗略介紹一下它們各自的結構和其之間的聯系，有利于后續創建每一層時聯系上下層編寫所需的代碼。

（一）ThreadCache的框架設計

功能：每個線程獨有的內存緩存，用于小于一定大小（如256KB）的內存的分配。線程從這里申請內存不需要加鎖，每個線程獨享一個Cache，大大提高了并發性能。

結構：通常設計為哈希桶結構，每個桶是一個按桶位置映射大小的內存塊對象的自由鏈表。這樣，線程在申請或釋放內存時，可以直接在自己的ThreadCache中進行，無需與其他線程競爭，無須加鎖。

內存管理：支持不同大小的內存塊申請，通過哈希映射和對齊規則，將不同大小的內存塊映射到不同的哈希桶中。

1.ThreadCache申請內存

線程A向ThreadCache申請內存，申請的內存大小≤256KB；
線程A向ThreadCache申請內存就是從ThreadCache中獲取空閑的內存塊，從而尋找ThreadCache中的空閑內存塊使用；
通過映射關系找到對應的哈希桶，查看對應的哈希桶中是否擁有空閑內存塊。如果有則直接使用（Pop一個空閑內存塊返回），如果沒有則向上一層CentralCache對應的哈希桶中申請空閑內存塊插入到自由鏈表中，再Pop給線程A；
由于每個線程都有屬于自己的Thread，因此我們引入一個知識點TLS無鎖訪問使得每個線程都可以有屬于自己的Thread。

2.ThreadCache釋放內存

線程A釋放內存，ThreadCache回收內存塊，內存大小依舊局限于256KB；
正常回收到對應的自由鏈表中；
當某條自由鏈表過長時，將其釋放給CentralCache。

哈希桶的結構：

由_freeList自由鏈表掛在對應的哈希桶位置組成；
每個哈希桶與其懸掛自由鏈表中的內存塊存在映射規則（如圖）。

3.如何劃分哈希桶布局-完成ThreadCache中的小組件

比如在向 ThreadCache 申請內存時，我們申請size個字節的內存，需要去有該size字節的對應的哈希桶中查找是否有空閑的內存塊，如果有則使用（如果沒有，則要去CentralCache中申請內存塊）。

因此，我們要考慮的變量一是對應的哈希桶，二是向CentralCache申請內存時，大小應該為多少。

在ThreadCache中如何通過申請size個字節，找到對應的哈希桶。
如何通過size個字節，找到向CentralCache申請的內存塊大小（找size對齊后的alignSize）。

256KB劃分情況如下（整體控制在最多10%左右的內存碎片浪費）：

字節數	對齊數	哈希桶下標
[ 1 , 128 ]	8	[ 0 , 16 )
[ 128+1 , 1024 ]	16	[ 16 , 72 )
[ 1024+1 , 8 * 1024 ]	128	[ 72 , 128 )
[ 8 * 1024+1 , 64 * 1024 ]	1024	[ 128 , 184 )
[ 64 * 1024+1 , 256 * 1024 ]	8 * 1024	[ 184 , 208 )

對齊數就是我們在分配內存時最后得到的內存大小一定是對齊數的倍數，比如我們申請129個字節，由表得知，它的對齊數是16字節，因此對齊后的大小應該是129÷16=8……1，還需要15個字節才能對齊，對齊后所需要的內存大小為129+15 = 144，也就是說浪費了15個字節。在前篇文章中，我們也已經介紹過內碎片的存在，這份浪費的15字節就是內碎片。

我們可以通過公式：

浪費率 = 浪費的字節數/對齊后的字節數；

得到 15÷144 ≈ 10%，依次類推，整體控制在10%左右的內存碎片浪費。對于1~128這個區間不做討論。

要想得到某個區間最大浪費率，我們可以通過讓分子變大，分母變小的方式來得到該區域的最大浪費率。

因此可以得到后幾組對齊數的最大浪費率：

對齊數為128，對齊后字節數為1152：127÷1152 ≈ 11.02%；

對齊數為1024，對齊后字節數為9216：1023÷9216 ≈ 11.10%；

對齊數為8*1024=8192，對齊后字節數為1152：8192÷73728 ≈ 11.11%

4.對象的映射規則

通過對齊數得到對齊后的字節數

如申請內存為14字節，在0~128的區間中對齊數為8，得到對齊后的字節數16

通過對齊數得到對應的哈希桶

如申請的內存為15字節，在0~128的區間中對齊數為8，得到（15+（8-1））/8 -1 = 1，對應的哈希桶為一號桶

將兩種映射規則放在類SizeClass中。

// 申請size字節,通過對齊數獲取對齊后的字節數alignNum/*static inline size_t _RoundUp(size_t size, size_t alignNum)      // RoundUp 的子函數
{// 找size對齊后的對齊后的字節數，也就是申請的內存大小alignNumif (size % alignNum == 0)return size;elsereturn (bytes / alignNum + 1)*alignNum;
}*/static inline size_t _RoundUp(size_t size, size_t alignNum)
{return (size + (alignNum - 1)) & ~(alignNum - 1);
}// size  的大小 size>=0  size<=256KB
static inline size_t RoundUp(size_t size)
{assert(size <= MAX_BYTES);         // MAX_BYTES = 256 * 1024；if (size <= 128){return _RoundUp(size, 8);}else if (size <= 1024){return _RoundUp(size, 16);}else if (size <= 8 * 1024){return _RoundUp(size, 128);}else if (size <= 64 * 1024){return _RoundUp(size, 1024);}else if (size <= 256 * 1024){return _RoundUp(size, 8 * 1024);}else{assert(false);// 出錯了return -1;}
}

// 對象申請的內存大小與對應的哈希桶之間的映射//方法一
// 申請內存與對應的哈希桶之間的映射
static inline size_t _Index(size_t size, size_t align_shift)
{ return ((size + (1 << align_shift) - 1) >> align_shift) - 1; 
}static inline size_t Index(size_t size)
{assert(size <= MAX_BYTES);static int group_array[4] = { 16, 56, 56, 56 };if (size <= 128){return _Index(size, 3);}else if (size <= 1024){return _Index(size - 128, 4) + group_array[0];}else if (size <= 8 * 1024){return _Index(size - 1024, 7) + group_array[0] + group_array[1];}else if (size <= 64 * 1024){return _Index(size - 8 * 1024, 10) + group_array[0] + group_array[1] + group_array[2];}else if (size <= 256 * 1024){return _Index(size - 64 * 1024, 13) + group_array[0] + group_array[1] + group_array[2] + group_array[3];}else{assert(false);return -1;}
}//方法二
/*
static inline size_t _Index(size_t bytes, size_t align)
{if (bytes % align == 0)return bytes / align - 1;elsereturn bytes / align;
}static inline size_t Index(size_t Bytes)
{assert(Bytes <= MAX_BYTES);// 每個桶有多少個節點static int group_array[4] = { 16, 56, 56, 56 };if (Bytes <= 128) {return _Index(Bytes, 8);}else if (Bytes <= 1024) {return _Index(Bytes - 128, 16) + group_array[0];}else if (Bytes <= 8 * 1024) {return _Index(Bytes - 1024, 128) + group_array[1] + group_array[0];}else if (Bytes <= 64 * 1024) {return _Index(Bytes - 8 * 1024, 1024) + group_array[2] + group_array[1] + group_array[0];}else if (Bytes <= 256 * 1024) {return _Index(Bytes - 64 * 1024, 8192) + group_array[3] + group_array[2] + group_array[1] + group_array[0];}else {assert(false);}return -1;
}
*/

需要注意的是，SizeClass類當中的成員函數最好設置為靜態成員函數，否則我們在調用這些函數時就需要通過對象去調用，并且對于這些可能會頻繁調用的函數，可以考慮將其設置為內聯函數。

（二）CentralCache的框架設計

CentralCache的結構與ThreadCache相似，映射關系也相似，但是由圖得知仍然存在不同。

功能：CentralCache作為所有線程共享的內存池，確保了內存資源在多個線程之間的公平分配和高效利用。也就是整個進程中只有一個CentralCache，可以通過設計一個單例模式來實現，由于每次訪問的都是共同的CentralCache，在進行讀寫的情況下，都要對其加鎖處理（因為線程1在讀的時候，可能線程2正在申請釋放內存塊，導致線程1獲取到的內存塊并不是可利用的空閑內存）

結構：采用哈希桶結構，但映射規則與ThreadCache相同。每個哈希桶下懸掛的是名為Span的變量，而每個span中都會指向一個自由鏈表，自由鏈表下懸掛的內存塊大小與它的桶號一一對應。

內存管理：CentralCache負責為各個線程的ThreadCache分配內存。當ThreadCache中的內存不足時，會向CentralCache申請新的內存塊。CentralCache會在適當的時候回收ThreadCache中不再使用的內存，以避免內存浪費和碎片化。

CentralCache與PageCache的框架結構中具有相似的Span結構，這也意味著又有一個共同點設計在Common.h文件中，在這里我們就不像ThreadCache中的自由鏈表與哈希桶的映射關系一樣單獨拿出來解釋，因為我們對Span的設計，會隨著各層之間的聯系發生變化，比如增加多個變量，如果在這里就進行介紹，在后續的講解中可能會遺忘。

1.CentralCache申請內存

ThreadCache向CentralCache申請內存，首先還是先找到對應哈希桶中的空閑內存塊，也就是說要分析每個Span，找到非空Span后，將一塊（或一批量）內存塊掛在ThreadCache對應自由鏈表下；
當對應哈希桶下，沒有非空的Span時，意味著ThreadCache無法獲取到空閑內存塊，因此CentralCache需要向上一層的PageCache中申請空閑的Span利用，申請到這份NewSpan后，還需要對這份NewSpan做處理，將其劃分為合適大小的內存塊，懸掛在Span中的自由鏈表下進行管理，然后就可以將這份新的非空Span中的內存塊劃分給ThreadCache；
需要注意的是我們在申請內存訪問CentralCache時，需要加上對應的桶鎖，避免其他同時間對CentralCache的操作影響我們的結果。

2.CentralCache釋放內存

CentralCache釋放內存有兩個大方向，一個是從ThreadCache中回收內存塊，一個是回收后的內存塊使得部分Span處于完全空閑狀態，可以釋放給PageCache，增加空間利用；
回收從ThreadCache中釋放的內存：當ThreadCache中某一自由鏈表下懸掛的空閑內存塊過多時（什么樣的判斷標準），會由CentralCache進行回收，這些內存塊都是有對應哈希桶中的Span下的內存塊分配出去的，但是屬于哪個Span是不確定的，因此在我們創建Span時就可以將內存塊的地址與Span做映射關系，確保回收回的內存塊可以找到對應的Span，至于怎么設計這份映射關系，后面再討論；
將完全空閑的Span釋放給PageCache：如何判斷這個Span是完全空閑的？也就是它管理的自由鏈表中沒有被使用的內存塊，設計一個_usecount變量進行管理，當分配出去對象時，_usecount就++，當_usecount為0時就表示所有對象回到了Span中，則將Span釋放回PageCache，需要相鄰的空閑Span合并，掛在PageCache對應的哈希桶下；
但是當我們從PageCache中申請的Span時這份Span也是空閑的，如果與要釋放的Span合并起來該怎么辦？在物理空間中，無論我們怎么切分內存空間，它們在物理位置上并沒有發生變化，因此在我們查找相鄰空閑Span的地址時，可能會導致合并后的Span一部分在CentralCache中，一部分在PageCache，一部分難以訪問到的情況，因為如果與CentralCache中相鄰Span合并，這部分Span仍然處于CentralCache中并沒有被釋放到PageCache中（如下圖所示）；
也就是說想要合并Span有兩個條件：空閑的Span和在PageCache中。思考：如何判斷Span時空閑Span，如何判斷Span是在PageCache中的？
將某個Span合并到PageCache中后，別忘了在CentralCache中對應的SpanList中刪除哦，不然會出錯的。

（三）PageCache的框架設計

功能：作為內存池的最上層緩存，以頁為單位存儲和分配內存。當CentralCache內存不足時，PageCache會向系統申請新的內存頁，并切割成小塊內存分配給CentralCache。當一個Span的幾個跨度頁的對象都回收以后，PageCache會回收CentralCache滿足條件的Span對象，并且合并相鄰的頁組成更大的頁，緩解內存碎片的問題。

結構：通常也采用哈希桶結構，但映射規則與ThreadCache和CentralCache不同，主要按頁號進行映射。

內存管理：PageCache負責大塊內存的分配和回收，以及與系統的內存交互。同時，也會合并相鄰的空閑頁，以減少內存碎片。

1.PageCach申請內存

CentralCache從PageCache中申請內存，對應哈希桶恰好有Span，切割好內存后交給CentralCache；
向頁數更大的哈希桶中查找Span，找到并切割合適的Span給CentralCache，要記得對切割前的Span和切割后的Span要清理和掛在對應的位置。比如從100頁切割了42頁大小的Span給CentralCache，那么原先的100頁Span要從100page的哈希桶中刪除，并將58頁大小Span重新掛在58page的哈希桶中，記得更新Span的自身信息；
當PageCache中無合適的Span時，需要從系統中申請內存，通過使用brk、mmp或者是VirtualAlloc等方式從操作系統中申請128頁Span，掛在對應哈希桶中。
注意PageCache與前兩者的映射關系不同，PageCache中的第i號桶中掛的Span都是i頁內存。

2.PageCach釋放內存

PageCache回收CentralCache釋放的內存，無可合并的Span，將釋放回的Span掛在對應哈希桶中；
PageCache中有可合并的相鄰的Span，合并直到無相鄰Span或最大為128頁后不再合并，減少內存碎片問題
PageCache釋放內存給操作系統。

二、ThreadCache（線程緩存）_內存設計

（一）處理哈希桶相關問題

首先創建一個ThreadCache哈希桶，我們需要先創建它其中的小組件。

從上列文字和圖形中，我們了解到ThreadCache是由不同內存塊組成的自由鏈表掛在對應哈希桶上的結構。

創建哈希桶中每個桶所屬的自由鏈表，由于CentralCache也是哈希桶結構，我們的自由鏈表創建可以放在一個公共的頭文件中，方便使用。將自由鏈表創建為類，管理切分好的小對象的自由鏈表。

// 這里是一個共同訪問的Common.h頭文件，會被TCMalloc不同層使用到的相同變量放在此處
#include <iostream>
#include <vector>
#include<assert.h>
using std::cout;
using std::endl;void*& NextObj(void* obj) {return *(void**)obj;
}// 創建ThreadCache和CentralCache都會使用到的_freelist
// 線程會從自由鏈表從申請到內存或釋放內存塊回到自由鏈表
// 
class FreeList
{
public:// 線程申請內存就是從對應的自由鏈表中取出內存塊（頭刪）void Push(void* obj){NextObj(obj) = _freeList;_freeList = obj;}// 線程釋放內存就是將內存塊釋放到對應的自由鏈表中（頭插）void* Pop(){assert(_freeList);	// 自由鏈表為空時，無法取出空閑內存塊void* obj = _freeList;_freeList = NextObj(_freeList);return obj;}// 判斷自由鏈表中是否存在內存bool Empty(){if (_freeList == nullptr)return true;return false;}private:void* _freeList = nullptr;};

我們現在可以通過FreeLsit創建一個哈希桶了。

接下來是對哈希桶進行管理，對于哈希桶的映射關系，如果還是不了解，去查看ThreadCache的框架設計中的講解

（二）ThreadCache–申請內存

// 放在公用頭文件中// thread cache和central cache自由鏈表哈希桶的表大小
static const int NFREELISTS = 208;// 小于 256 * 1024 向 ThreadCache 申請
static const int MAX_BYTES = 256 * 1024;
// 大于則向 PageCache 或者 堆上申請，這里我們暫時不多思考，一步一步來

當內存申請size<=256KB，先獲取到線程本地存儲的ThreadCache對象，計算size映射的對齊后的對象大小和哈希桶自由鏈表下標 Index 。
申請內存時，如果ThreadCache對應哈希桶中的自由鏈表**（_freeLists[index]）**有空閑內存塊時，則直接Pop()一個內存對象返回。
申請內存時，如果ThreadCache對應哈希桶中的自由鏈表**（_freeLists[index]）**沒有空閑內存塊時，則批量從CentralCache中獲取對齊后大小的數量的對象，插入到自由鏈表并返回一個對象。

// ThreadCache.h#pragma once#include "CommonPool.h"class ThreadCache
{
private:FreeList _freeLists[NFREELISTS]; // 哈希桶public:// 申請和釋放對象內存void* ThreadAlloc(size_t size);void* ThreadFree(void* obj,size_t size);// 從CentalCache中申請size內存void* FetchFromCentralCache(size_t index,size_t size);
};

定義ThreadCache 中的成員函數，并且完善其他類。

/* 申請內存 */
/* ThreadCache.cpp */
void* ThreadCache::ThreadAlloc(size_t size)
{// size 一定是小于等于256KB，且申請獲得的空間大小一定是向上對齊的assert(size <= MAX_BYTES);// 根據對象申請的大小，在對應的哈希桶中找到對應的內存塊// 如果有就直接使用，如果沒有就需要先從CentralCache中申請對應內存塊給ThreadCachesize_t alignBytes = SizeClass::RoundUp(size);size_t index = SizeClass::Index(size);// 接下來申請內存，注意我們要在對應的哈希桶中找到空閑的內存塊使用if (!_freeLists[index].Empty()){return _freeLists[index].Pop();// 從該哈希桶中的存放空閑內存塊的自由鏈表中取出空閑內存塊給對象}else{// 如果對應的哈希桶中沒有空閑內存塊，需要向centralcache中申請內存塊分配給threadcache// 申請到的內存塊要分配給以alignBytes為對齊數，放在在下標位index的哈希桶中return FetchFromCentralCache(index, alignBytes);}
}

關于 FetchFromCentralCache(index, alignSize); ，需要聯系到 CentralCache ，后續再講解。

（三）線程局部存儲–無鎖訪問ThreadCache

Thread Local Storage（線程局部存儲）TLS

線程本地存儲（Thread Local Storage） - 坦坦蕩蕩 - 博客園

如果一個變量是全局的，那么所有線程訪問的是同一份，某一個線程對其修改會影響其他所有線程。如果希望每個線程訪問到這個變量，并且不會影響其他線程中的這個變量，那么我們該怎么創建？

我們的高并發內存池就是每個線程獨有一個ThreadCache線程緩存，互不影響，也無需加鎖，那么如何創建每個線程各自的ThreadCache？

因此我們引入一個概念：

如果我們需要一個變量在每個線程中都能訪問，并且這個變量在每個線程中互不影響，這就是TLS。

線程局部存儲TLS(Thread Local Storage)是一種變量的存儲方法，這個變量在它所在的線程內是全局可訪問的，但是不能被其他線程訪問到，這樣就保持了數據的線程獨立性。

而熟知的全局變量，是所有線程都可以訪問的，這樣就不可避免需要鎖來控制，增加了控制成本和代碼復雜度。

TLS依情況而定有Windows下的，也有Linux下的。既有動態也有靜態，這里我們選擇靜態TLS：

_declspec(thread) DWORD data=0;

聲明了_declspec(thread)的變量，會為每一個線程創建一個單獨的拷貝。

// TLS thread local storage（線程局部存儲、線程本地存儲）
// 通過TLS 每個線程無鎖的獲取自己的專屬的ThreadCache對象
static _declspec(thread) ThreadCache* pTLSThreadCache = nullptr;// thread 用于聲明一個線程本地變量， 
// _declspec(thread)的前綴是Microsoft添加給Visual C++編譯器的一個修改符。

上列代碼在ThreadCache.h頭文件中聲明了為每個線程創建一份ThreadCache。但也僅僅是聲明，并不是為每個線程創建了屬于自己的ThreadCache。

下列代碼是對每個線程創建屬于自己的ThreadCache，當該線程調用相關申請內存的接口時才會創建自己的ThreadCache。

//通過TLS，每個線程無鎖的獲取自己專屬的ThreadCache對象
if (pTLSThreadCache == nullptr)
{pTLSThreadCache = new ThreadCache;
}

（四）ThreadCache–封裝使用及測試

1.對ThreadCache對象的使用進行封裝

在使用過TLS后，對每一個線程創建一份ThreadCache時，將申請（釋放）的功能進行再一次的封裝，有利于后續對象申請（釋放）空間直接使用。

// 我們創建一個新的頭文件 Concurrent.h
// 在這里將threadcache封裝起來以便對象申請釋放內存時更方便使用
#include "Common.h"
#include "ThreadCache.h"static void* concurrentAlloc(size_t size)
{// 為每一個線程申請屬于他自己的threadcacheif (PTLSThreadCache == nullptr){PTLSThreadCache = new ThreadCache;// 這里可以連同后面的測試 編寫一段代碼// cout<< "Thread x creates a threadcache object."<< endl	;}// get_id()它用于獲取當前線程的唯一標識符。// 每個線程在程序執行期間都有一個唯一的標識符，這個函數返回一個表示當前線程ID的對象。//cout << std::this_thread::get_id() << ":" << pTLSThreadCache << endl;// 調用threadcache申請內存return PTLSThreadCache->Allocate(size);
}

2.測試每個線程有屬于自己的ThreadCache對象

在Test.cpp文件中，我們對上列代碼進行測試，觀察線程之間使用同一變量是否會互相影響，測試每一個線程是否獲得了屬于自己的ThreadCache對象。

// Test.cpp#include <thread>
void Alloc1()
{for (size_t i = 0; i < 5; ++i){//申請6字節大小的內存對象void* ptr = concurrentAlloc(6);}
}void Alloc2()
{for (size_t i = 0; i < 5; ++i){//申請9字節大小的內存對象void* ptr = concurrentAlloc(9);}
}void TextpTLSThreadCache()
{std::thread t1(Alloc1);std::thread t2(Alloc2);t1.join();t2.join();
}int main()
{//測試每個線程是否有屬于自己的ThreadCacheTextpTLSThreadCache();return 0;
}