并發內存池的三個主要組成部分:
- 線程緩存(Thread Cache)
- 每個線程擁有獨立的線程緩存,用于處理小于
256KB
的內存分配。 - 由于每個線程都有自己的緩存,線程在從線程緩存中分配內存時無需加鎖,這有效避免了競爭,提升了并發性能。
- 每個線程擁有獨立的線程緩存,用于處理小于
- 中心緩存(Central Cache)
- 中心緩存是全局共享的,負責管理所有線程緩存的內存。
- 當線程緩存中的內存不足時,會從中心緩存獲取資源;同時,中心緩存也會定期回收線程緩存中的未使用對象,避免單個線程占用過多內存,導致資源不均衡。
- 因為所有線程都從中心緩存獲取內存對象,中心緩存的訪問會存在競爭,通常通過使用桶鎖來減少鎖競爭的影響。只有當線程緩存沒有足夠資源時,才會從中心緩存獲取,這樣能確保鎖競爭不會過于激烈。
- 頁緩存(Page Cache)
- 頁緩存是管理頁級內存分配的緩存,位于中心緩存的上一級。
- 當中心緩存中的內存不足時,頁緩存會分配一定數量的頁,并將其切割成固定大小的小塊,供中心緩存使用。
- 頁緩存通過回收和合并相鄰的空閑頁,緩解內存碎片(外碎片)問題。當多個頁的內存塊被回收后,它們會合并成更大的頁,避免內存的浪費。
分層管理:
- 線程緩存:高效避免加鎖,提升并發性能。
- 中心緩存:確保內存資源的均衡分配,減少鎖競爭。
- 頁緩存:減少內存碎片,優化內存回收和分配策略。
線程緩存(Thread Cache)
核心思想
為每個線程提供一個獨立的緩存,使得線程在申請和釋放內存時不需要加鎖,從而提高性能。通過內存對齊和哈希桶的方式,控制內存碎片,同時避免了大量小內存分配帶來的效率損失。inline
和static inline
則用來優化代碼的執行效率。
- 線程局部存儲(TLS):每個線程都會有一個自己的線程緩存(
ThreadCache
),它通過“哈希桶”結構來管理內存。每個哈希桶對應一個特定大小的內存塊。 - 哈希桶結構:線程緩存內部是由多個哈希桶組成的,每個哈希桶代表一個內存大小區間。如果某個內存塊請求的大小在某個區間內,線程就會根據大小映射到相應的哈希桶。每個哈希桶里維護著一個自由鏈表,用于管理空閑的內存對象。
內存申請
當一個線程申請內存時,如果申請的內存小于256KB
,首先會查找自己線程的緩存。如果哈希桶中有可用的內存塊,直接返回。如果沒有,就會從 central cache(中央緩存) 中批量獲取對象,插入到哈希桶的自由鏈表里。
內存釋放
- 當一個內存塊被釋放時,如果它小于
256KB
,線程會將其釋放到自己的線程緩存里。釋放時,通過計算哈希桶的位置,將對象加入到自由鏈表中。 - 如果鏈表中內存對象的數量過多,線程會將一部分內存回收到中央緩存中,以保持內存池的效率。
內存對齊與映射
- 內存對齊:為了減少內存碎片,內存分配時會進行對齊。根據內存請求的大小,內存會被分配到不同的對齊大小:
1
~128
字節的內存會對齊到8
字節。128
~1024
字節的內存對齊到16
字節。1024
~8KB
的內存對齊到128
字節,以此類推。
- 映射規則:這種內存對齊規則通過Roundup函數實現,它將內存請求的大小對齊到最接近的有效大小。例如,如果你請求
11
字節的內存,但由于對齊規則,最終會分配16
字節 - 哈希桶的映射:通過 Index函數 ,內存的對齊后大小會映射到一個特定的哈希桶。在計算時,會根據內存對齊的大小,確定內存塊所在的桶。
static inline
與 inline
函數
inline
函數:是編譯器在調用時嘗試將函數代碼插入調用點,從而避免函數調用的開銷。這通常用于提高性能。static inline
函數:結合了static
和inline
的優勢。static
確保該函數只在當前文件中可見,防止了多文件間重復定義,而inline
則減少了調用開銷,提高了效率。
自由鏈表的設計
設計自由鏈表,其實就是實現一個單鏈表,方便插入刪除,同時標識鏈表是否為空,自由鏈表在中心緩存中也有使用,所以放入common.h
中。
Common.h
#pragma once
#include <iostream>
#include <assert.h>const int MAX_BYTES = 1024 * 256; // 最大內存塊尺寸限制(256KB)// 自由鏈表類 - 用于管理空閑內存塊
class FreeList {
public:// 將內存塊插入鏈表頭部void Push(void* obj) {assert(obj);// 將obj頭插到鏈表,利用內存塊頭部空間存儲下一個節點指針*(void**)obj = _free_list; // 將obj的前4/8字節指向當前鏈表頭_free_list = obj; // 更新鏈表頭為當前對象}// 從鏈表頭部彈出一個內存塊void* Pop() {assert(_free_list); // 確保鏈表非空void* obj = _free_list; // 獲取當前鏈表頭_free_list = *(void**)obj; // 將鏈表頭更新為下一個節點return obj;}// 判斷鏈表是否為空bool Empty() {return _free_list == nullptr;}private:void* _free_list = nullptr; // 鏈表頭指針(存儲空閑內存塊地址)
};// 內存塊尺寸對齊與索引計算類
class SizeClass {
public:// 內存對齊輔助函數(將bytes向上對齊到alignNum的倍數)static inline size_t _Roundup(size_t bytes, size_t alignNum) {// 計算公式解釋:(bytes + alignNum - 1) 確保超過對齊基數時進位// & ~(alignNum - 1) 用于抹去低位實現對齊// 示例:alignNum=8時,~(0b111) => 0b...11111000,清除后三位return (bytes + alignNum - 1) & ~(alignNum - 1);}// 根據請求大小計算對齊后的實際分配大小static size_t Roundup(size_t size) {if (size <= 128) { // [1,128]字節按8字節對齊return _Roundup(size, 8);}else if (size <= 1024) { // (128,1024]按16字節對齊return _Roundup(size, 16);}else if (size <= 8 * 1024) { // (1KB,8KB]按128字節對齊return _Roundup(size, 128);}else if (size <= 64 * 1024) { // (8KB,64KB]按1KB對齊return _Roundup(size, 1024);}else if (size <= 128 * 1024) { // (64KB,128KB]按8KB對齊return _Roundup(size, 8 * 1024);}else {return 0; // 超過最大限制返回0(需配合斷言使用)}}// _Index計算的是當前size所在區域的第幾個下標,所以Index的返回值需要加上前面所有區域的哈希桶的個數static inline size_t _Index(size_t bytes, size_t align_shift) {// align_shift表示對齊數的二進制位移量(如8字節對齊對應shift=3)// 公式等效:(bytes + alignNum - 1) / alignNum - 1return ((bytes + (1 << align_shift) - 1) >> align_shift) - 1;}// 計算內存塊在對應規格數組中的索引位置static inline size_t Index(size_t bytes) {assert(bytes <= MAX_BYTES); // 確保請求大小在合法范圍內// 各規格組的鏈表數量(經驗值設定)// 對應不同區間:[8B對齊組][16B對齊組][128B對齊組][1KB對齊組][8KB對齊組]static int group_array[4] = { 16, 56, 56, 56 };if (bytes <= 128) { // 8字節對齊區間(16個規格:8,16,...,128)return _Index(bytes, 3); // 3=log2(8)}else if (bytes <= 1024) { // 16字節對齊區間(56個規格:144,160,...,1024)return _Index(bytes - 128, 4) + group_array[0]; // 4=log2(16)}else if (bytes <= 8 * 1024) { // 128字節對齊區間(56個規格:1152,1280,...,8K)return _Index(bytes - 1024, 7) + group_array[1] + group_array[0];}else if (bytes <= 64 * 1024) { // 1KB對齊區間(56個規格:9K,10K,...,64K)return _Index(bytes - 8 * 1024, 10) + group_array[2] + group_array[1] + group_array[0];}else if (bytes <= 256 * 1024) { // 8KB對齊區間(3個規格:72K,80K,...,256K)return _Index(bytes - 64 * 1024, 13) + group_array[3] + group_array[2] + group_array[1] + group_array[0];}else {assert(false); // 觸發斷言表示超出設計容量}return -1; // 無效返回值(實際會被斷言攔截)}
};
關鍵設計說明
- 自由鏈表管理
Push
/Pop
操作時間復雜度O(1)
- 利用內存塊頭部空間存儲鏈表指針(節省管理開銷)
- 分級內存對齊策略
- 索引計算優化
- 使用分組累計偏移量(group_array)快速定位規格位置
- 示例:1024字節請求計算過程:
Index = _Index(1024-128,16) + 16= ((896 + 15)/16 - 1) + 16= (911/16 - 1) + 16= (56 - 1) + 16= 71
- 性能優勢
- 對齊操作使用位運算替代除法,效率提高
- 分級策略減少內存碎片
- 索引計算時間復雜度O(1),適合高頻調用場景
線程緩存的設計
在有了上述的基礎之后,我們可以開始搭建線程緩存(thread cache
)的框架。實際上,這個框架就是一個哈希桶,每個桶里都維護著一個自由鏈表,用來存儲可用的內存對象。
為了讓每個線程都有自己的緩存,我們可以使用一個叫做thread_local
的關鍵字。它的作用是聲明一個線程局部存儲 TLS
變量。線程局部存儲 TLS
,是一種變量的存儲方法,這個變量在它所在的線程內是全局可訪問的,但是不能被其他線程訪問到,這樣就保持了數據的線程獨立性。而熟知的全局變量,是所有線程都可以訪問的,這樣就不可避免需要鎖來控制,增加了控制成本和代碼復雜度。
ThreadCache.h
#pragma once
#include <assert.h>
#include <thread>
#include "Common.h" // 包含公共定義(如FreeList、SizeClass等)const size_t NFREELISTS = 208; // 自由鏈表總數(對應SizeClass的分組計算)// 線程本地緩存類 - 每個線程獨立的內存緩存
class ThreadCache {
public:// TLS指針(延遲初始化)static thread_local ThreadCache* pTLSThreadCache;// 內存分配接口void* Allocate(size_t size) {assert(size <= MAX_BYTES); // 校驗請求大小合法性// 1. 計算對齊后的實際需求大小size_t alignSize = SizeClass::Roundup(size);// 2. 計算對應的自由鏈表索引size_t index = SizeClass::Index(alignSize);// 3. 優先從本地自由鏈表獲取內存if (!_freelists[index].Empty()) {return _freelists[index].Pop();}// 4. 自由鏈表為空時從中央緩存批量獲取return FetchFromCentralCache(index, alignSize);}// 內存釋放接口void Deallocate(void* ptr, size_t size) {assert(ptr && size <= MAX_BYTES); // 校驗參數有效性// 1. 計算原始分配大小size_t alignSize = SizeClass::Roundup(size);// 2. 計算對應的自由鏈表索引size_t index = SizeClass::Index(alignSize);// 3. 將內存塊插回自由鏈表_freelists[index].Push(ptr);}// 從中央緩存補充內存塊(具體實現依賴CentralCache類)void* FetchFromCentralCache(size_t index, size_t size) {return nullptr;}private:FreeList _freelists[NFREELISTS]; // 自由鏈表數組(每個元素管理特定尺寸內存塊)
};// 靜態成員變量初始化
thread_local ThreadCache* ThreadCache::pTLSThreadCache = nullptr;
關鍵設計說明
- 線程本地存儲(TLS)機制
// 使用示例(需在cpp文件中初始化): thread_local ThreadCache* ThreadCache::pTLSThreadCache = nullptr;
- 每個線程首次訪問時初始化
pTLSThreadCache
- 線程退出時自動銷毀,無需顯式資源釋放
- 每個線程首次訪問時初始化
- 內存分配流程
- 內存釋放流程
- 性能優化點
- 零鎖競爭:通過
thread_local
實現無鎖操作 - 緩存友好:高頻操作完全在本地鏈表完成
- 批量傳輸:
FetchFromCentralCache
批量獲取內存塊
- 零鎖競爭:通過
該實現需要配合CentralCache
中央緩存類和PageHeap
頁堆類共同工作,形成完整的三層內存池架構。
只有線程緩存的內存池設計及測試
ConcurrentAlloc.h
#pragma once#include "Common.h"
#include "ThreadCache.h"// 并發內存分配入口(仿tcmalloc實現)
// 參數:size - 請求分配的內存大小(字節)
static void* ConcurrentAlloc(size_t size ){// TLS延遲初始化(每個線程首次調用時創建專屬緩存)if (ThreadCache::pTLSThreadCache == nullptr) {// 保證線程安全:thread_local保證每個線程只初始化一次ThreadCache::pTLSThreadCache = new ThreadCache();}std::cout << std::this_thread::get_id() << ":" << ThreadCache::pTLSThreadCache<< std::endl;// 執行實際內存分配void* ptr = ThreadCache::pTLSThreadCache->Allocate(size);return ptr;
}// 線程調用這個函數回收空間
static void ConcurrentFree(void* ptr){// 后續完成
}
test.cpp
#include "ConcurrentAlloc.h"// 分配內存的函數1
void Alloc1() {// 循環進行5次內存分配,每次分配6字節的內存for (size_t i = 0; i < 5; i++) {void* ptr = ConcurrentAlloc(6); // 使用并發內存分配函數分配6字節內存// 這里可以對 ptr 進行操作或釋放內存}
}// 分配內存的函數2
void Alloc2() {// 循環進行5次內存分配,每次分配7字節的內存for (size_t i = 0; i < 5; i++) {void* ptr = ConcurrentAlloc(7); // 使用并發內存分配函數分配7字節內存// 這里可以對 ptr 進行操作或釋放內存}
}// 測試線程本地存儲(TLS)機制
void TestTLS() {std::thread t1(Alloc1);std::thread t2(Alloc2);t1.join();t2.join();
}int main() {// 調用TestTLS函數,啟動并執行分配內存的測試TestTLS();
}
推薦一下
https://github.com/0voice