死磕Netty原始碼之記憶體分配詳解(四)PoolArena全域性記憶體分配
阿新 • • 發佈：2018-12-25
記憶體分配
全域性分配
記憶體池的初始階段執行緒是沒有記憶體快取的，所以最開始的記憶體分配都需要在全域性分配區進行分配
全域性分配區的記憶體構造和執行緒私有分配區的類似(包含Tiny、Small、Normal幾種規模 計算索引的方式也都是一模一樣的)，無論是TinySubpagePools還是SmallSubpagePools成員在記憶體池初始化時是不會預置記憶體的，所以最開始記憶體分配階段都會進入PoolArena的allocateNormal方法，程式碼如下
private void allocateNormal(PooledByteBuf<T> buf, int 分配記憶體時為什麼選擇從q050開始
1.qinit的chunk利用率低，但不會被回收
2.q075和q100由於記憶體利用率太高，導致記憶體分配的成功率大大降低，因此放到最後
3.q050儲存的是記憶體利用率50%~100%的Chunk，這應該是個折中的選擇。這樣能保證Chunk的利用率都會保持在一個較高水平提高整個應用的記憶體利用率，並且記憶體利用率在50%~100%的Chunk記憶體分配的成功率有保障
4.當應用在實際執行過程中碰到訪問高峰，這時需要分配的記憶體是平時的好幾倍需要建立好幾倍的Chunk，如果先從q0000開始，這些在高峰期建立的chunk被回收的概率會大大降低，延緩了記憶體的回收進度，造成記憶體使用的浪費1.嘗試從現有的Chunk進行分配
boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
if (head == null || normCapacity > maxCapacity) {
return false;
}
// 從head節點開始遍歷
for (PoolChunk<T> cur = head;;) {
// 嘗試使用已有PoolChunk進行分配
long handle = cur.allocate(normCapacity);
if (handle < 0) {
cur = cur.next;
// 如果到了尾節點還沒分配成功
// 說明當前PoolChunkList無法分配記憶體
if (cur == null) {
return false;
}
} else {
// 如果分配記憶體成功 初始化ByteBuf
cur.initBuf(buf, handle, reqCapacity);
// 判斷PoolChunkList是否需要重新調整
if (cur.usage() >= maxUsage) {
remove(cur);
nextList.add(cur);
}
return true;
}
}
}2.嘗試建立一個Chuank進行記憶體分配
使用newChunk(pageSize, maxOrder, pageShifts, chunkSize)對PoolChunk進行初始化，後再呼叫PoolChunk.allocate方法進行真正的記憶體分配動作，在分析這個分配動作之前先來了解一下PoolChunk。下圖是PoolChunk的資料結構
PoolChunk預設由2048個Page組成(Page預設大小為8k)，圖中節點的值為在陣列MemoryMap的下標
1、如果需要分配大小8k的記憶體則只需要在第11層找到第一個可用節點即可
2、如果需要分配大小16k的記憶體則只需要在第10層找到第一個可用節點即可
3、如果節點1024存在一個已經被分配的子節點2048則該節點不能被分配，如需要分配大小16k的記憶體，這個時候節點2048已被分配節點2049未被分配，就不能直接分配節點1024，因為該節點目前只剩下8k記憶體PoolChunk內部會保證每次分配記憶體大小為8K*(2n)，為了分配一個大小為ChunkSize/(2k)的節點，需要在深度為K的層從左開始匹配節點，那麼如何快速的分配到指定記憶體？memoryMap初始化
memoryMap = new byte[maxSubpageAllocs << 1];
depthMap = new byte[memoryMap.length];
int memoryMapIndex = 1;
for (int d = 0; d <= maxOrder; ++ d) {
int depth = 1 << d;
for (int p = 0; p < depth; ++ p) {
memoryMap[memoryMapIndex] = (byte) d;
depthMap[memoryMapIndex] = (byte) d;
memoryMapIndex ++;
}
}
分配完成後
depthMap->[0, 0, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3…]
memoryMap->[0, 0, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3…]MemoryMap陣列中每個位置儲存的是該節點所在的層數，有什麼作用？
對於節點512其層數是9則：
1、如果memoryMap[512]=9，則表示其本身到下面所有的子節點都可以被分配
2、如果memoryMap[512]=10，則表示節點512下有子節點已經分配過，而其子節點中的第10層還存在未分配的節點
3、如果memoryMap[512]=12(即總層數+1)可分配的深度已經大於總層數, 則表示該節點下的所有子節點都已經被分配下面看看如何向PoolChunk申請一段記憶體
long allocate(int normCapacity) {
// 當需要分配的記憶體大於pageSize時
if ((normCapacity & subpageOverflowMask) != 0) {
return allocateRun(normCapacity);
} else {
// 否則使用方法allocateSubpage分配記憶體
return allocateSubpage(normCapacity);
}
}PoolChunk分配
Page級別記憶體分配
當需要分配的記憶體大於PageSize的時候，使用allocateRun()進行分配
private long allocateRun(int normCapacity) {
// 計算出當前要分配的節點在Page樹結構中的層級
int d = maxOrder - (log2(normCapacity) - pageShifts);
// 根據層級在Page樹找出可分配記憶體節點進行記憶體分配並返回節點編號
// 並將當前節點狀態置為被使用狀態 且父結構的節點設定為被使用狀態
int id = allocateNode(d);
if (id < 0) {
return id;
}
freeBytes -= runLength(id);
return id;
}Page級別記憶體分配 PooledByteBuf的初始化
void initBuf(PooledByteBuf<T> buf, long handle, int reqCapacity) {
int memoryMapIdx = memoryMapIdx(handle);
int bitmapIdx = bitmapIdx(handle);
if (bitmapIdx == 0) {
byte val = value(memoryMapIdx);
// 斷言該節點未被使用
assert val == unusable : String.valueOf(val);
// Page級別PooledByteBuf初始化
// runOffset(memoryMapIdx)計算偏移量 runLength(memoryMapIdx)獲取當前節點長度
buf.init(this, handle, runOffset(memoryMapIdx) + offset, reqCapacity, runLength(memoryMapIdx), arena.parent.threadCache());
} else {
// SubPage級別PooledByteBuf初始化
initBufWithSubpage(buf, handle, bitmapIdx, reqCapacity);
}
}
void init(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
init0(chunk, handle, offset, length, maxLength, cache);
}
private void init0(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
assert handle >= 0;
assert chunk != null;
this.chunk = chunk;
memory = chunk.memory;
allocator = chunk.arena.parent;
this.cache = cache;
this.handle = handle;
this.offset = offset;
this.length = length;
this.maxLength = maxLength;
tmpNioBuf = null;
}SubPage級別記憶體分配
當需要分配的記憶體小於PageSize的時候，使用allocateSubpage()進行分配
private long allocateSubpage(int normCapacity) {
// 獲取規格對應的PoolSubpage
PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity);
synchronized (head) {
int d = maxOrder;
// 找到一個可分配的Page的ID
int id = allocateNode(d);
if (id < 0) {
return id;
}
final PoolSubpage<T>[] subpages = this.subpages;
final int pageSize = this.pageSize;
// 修改該chunk的空閒記憶體大小
freeBytes -= pageSize;
// 獲取page在subPages中的索引
int subpageIdx = subpageIdx(id);
PoolSubpage<T> subpage = subpages[subpageIdx];
if (subpage == null) {
// 新建PoolSubpage 並新增到tinySubpagePools或smallSubpagePools中的一個快取中
subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity);
subpages[subpageIdx] = subpage;
} else {
subpage.init(head, normCapacity);
}
// 使用subpage分配記憶體(從點陣圖中找到一個未被使用的子Page)
return subpage.allocate();
}
}SubPage級別記憶體分配 PooledByteBuf的初始化
void initBuf(PooledByteBuf<T> buf, long handle, int reqCapacity) {
int memoryMapIdx = memoryMapIdx(handle);
int bitmapIdx = bitmapIdx(handle);
if (bitmapIdx == 0) {
byte val = value(memoryMapIdx);
// 斷言該節點未被使用
assert val == unusable : String.valueOf(val);
// Page級別PooledByteBuf初始化
// runOffset(memoryMapIdx)計算偏移量 runLength(memoryMapIdx)獲取當前節點長度
buf.init(this, handle, runOffset(memoryMapIdx) + offset, reqCapacity, runLength(memoryMapIdx), arena.parent.threadCache());
} else {
// SubPage級別PooledByteBuf初始化
initBufWithSubpage(buf, handle, bitmapIdx, reqCapacity);
}
}
private void initBufWithSubpage(PooledByteBuf<T> buf, long handle, int bitmapIdx, int reqCapacity) {
assert bitmapIdx != 0;
int memoryMapIdx = memoryMapIdx(handle);
PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
assert subpage.doNotDestroy;
assert reqCapacity <= subpage.elemSize;
// runOffset(memoryMapIdx)計算偏移量 runLength(memoryMapIdx)獲取當前節點長度
buf.init(this, handle, runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize + offset, reqCapacity, subpage.elemSize, arena.parent.threadCache());
}
private void init0(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
assert handle >= 0;
assert chunk != null;
this.chunk = chunk;
memory = chunk.memory;
allocator = chunk.arena.parent;
this.cache = cache;
this.handle = handle;
this.offset = offset;
this.length = length;
this.maxLength = maxLength;
tmpNioBuf = null;
}最終ByteBuf可根據chunk，memory ，handle ，offset 等找到對應的記憶體資訊
將PoolChunk新增到PoolChunkList中
初始狀態下所有的PoolChunkList都是空的，所以在此先建立chunk塊要新增到PoolChunkList中，需要注意的是雖然都是通過qInit.add新增chunk，這並不代表chunk都會被新增到qInit這個PoolChunkList，看一下PoolChunkList的add方法就可以知道
void add(PoolChunk<T> chunk) {
if (chunk.usage() >= maxUsage) {
nextList.add(chunk);
return;
}
add0(chunk);
}
void add0(PoolChunk<T> chunk) {
chunk.parent = this;
if (head == null) {
head = chunk;
chunk.prev = null;
chunk.next = null;
} else {
chunk.prev = null;
chunk.next = head;
head.prev = chunk;
head = chunk;
}
}PoolChunkList有兩個重要的引數MinUsage和MaxUsage(最小/大使用率)，當Chunk中記憶體可用率在[MinUsage,MaxUsage]區間時這個Chunk才會落到該PoolChunkList中，否則把Chunk傳到下一個PoolChunkList進行檢查。從這裡可以看出Chunk只會被新增到記憶體匹配的PoolChunkList中，為了更有說服力，再看一下free方法的程式碼
void free(PoolChunk<T> chunk, long handle) {
chunk.free(handle);
if (chunk.usage() < minUsage) {
remove(chunk);
if (prevList == null) {
// 如果前置節點為空
// 並且使用率為0 則釋放記憶體
assert chunk.usage() == 0;
arena.destroyChunk(chunk);
} else {
prevList.add(chunk);
}
}
} 在釋放記憶體時也會檢查MinUsage如果不匹配傳到上一個PoolChunkList進行檢查，最終歸還到大小跟它匹配的PoolChunkList中
PoolSubpage
關於PoolSubpage應該是在之前的部落格[02.死磕Netty原始碼之記憶體分配詳解(二)PoolArena記憶體分配結構分析]就已經提及過的，這裡我們將對PoolSubpage進行詳細解析。Netty提供了PoolSubpage把PoolChunk的一個Page節點8k記憶體劃分成更小的記憶體段，通過對每個記憶體段的標記與清理標記進行記憶體的分配與釋放，它的資料結構如下
成員變數
// 當前page所屬的chunk
final PoolChunk<T> chunk;
// 當前page在chunk中的id
private final int memoryMapIdx;
// 當前page在chunk.memory的偏移量
private final int runOffset;
// page大小
private final int pageSize;
// 每個元素是一個長整形數數記錄記憶體頁的分配資訊，每個二進位制位都代表頁內的一個記憶體單元
// 當二進位制位為1表示對應的記憶體塊被分配過，第一個元素對應0-63號記憶體單元 第二個元素對應64-127號記憶體單元，第三個元素對應128-191號記憶體單元等等
// bitmap[0]=0b0000000...0001111表示0,1,2,3這四個記憶體單元都已經被分配給對應的請求了。這個bitmap用來輔助計算下一個分配塊的索引也即上面的nextAvail引數
private final long[] bitmap;
// 維護連結串列結構
PoolSubpage<T> prev;
PoolSubpage<T> next;
// 當前PoolSubpage不能銷燬
boolean doNotDestroy;
// 該page切分後每一段的大小
int elemSize;
// 該page包含的段數量 (maxNumElems=pageSize/elemSize)
private int maxNumElems;
private int bitmapLength;
// 下一個可用的位置
private int nextAvail;
// 記憶體頁還能分配多少次，它的初始值等同於maxNumElems 分配一次值遞減
private int numAvail;構造方法
PoolSubpage(PoolSubpage<T> head, PoolChunk<T> chunk, int memoryMapIdx, int runOffset, int pageSize, int elemSize) {
this.chunk = chunk;
this.memoryMapIdx = memoryMapIdx;
this.runOffset = runOffset;
this.pageSize = pageSize;
// pageSize >>> 10 => pageSize/16/64
bitmap = new long[pageSize >>> 10];
init(head, elemSize);
}預設初始化bitmap長度為8，這裡解釋一下為什麼只需要8個元素
其實分配記憶體大小都是處理過的最小為16，說明一個Page可以分成8192/16=512個記憶體段，一個long有64位可以描述64個記憶體段，這樣只需要512/64=8個long就可以描述全部記憶體段了init根據當前需要分配的記憶體大小，確定需要多少個bitmap元素，實現如下
void init(PoolSubpage<T> head, int elemSize) {
doNotDestroy = true;
this.elemSize = elemSize;
if (elemSize != 0) {
maxNumElems = numAvail = pageSize / elemSize;
nextAvail = 0;
bitmapLength = maxNumElems >>> 6;
if ((maxNumElems & 63) != 0) {
bitmapLength ++;
}
for (int i = 0; i < bitmapLength; i ++) {
bitmap[i] = 0;
}
}
addToPool(head);
}下面通過分佈申請4096和32大小的記憶體，說明如何確定bitmapLength的值
比如當前申請大小4096的記憶體maxNumElems和numAvail為2，說明一個page被拆分成2個記憶體段，2>>>6=0且2&63=0，所以bitmapLength為1，說明只需要一個long就可以描述2個記憶體段狀態。如果當前申請大小32的記憶體maxNumElems和numAvail為256，說明一個page被拆分成256個記憶體段，256>>>6=4，說明需要4個long描述256個記憶體段狀態記憶體分配
PoolSubpage的記憶體分配由allocate()完成
long allocate() {
if (elemSize == 0) {
return toHandle(0);
}
if (numAvail == 0 || !doNotDestroy) {
return -1;
}
// 找到當前page中可分配記憶體段的bitmapIdx
final int bitmapIdx = getNextAvail();
// 確定bitmap陣列下標為q的long數 用來描述bitmapIdx記憶體段的狀態
int q = bitmapIdx >>> 6;
// 將超出64的那一部分二進位制數抹掉得到一個小於64的數
int r = bitmapIdx & 63;
// 斷言該記憶體段未被使用
assert (bitmap[q] >>> r & 1) == 0;
// 將對應位置設定為1
bitmap[q] |= 1L << r;
if (-- numAvail == 0) {
removeFromPool();
}
return toHandle(bitmapIdx);
}
GetNextAvail如何實現找到下一個可分配的記憶體段？
private int getNextAvail() {
int nextAvail = this.nextAvail;
// 如果nextAvail大於等於0
// 說明nextAvail指向了下一個可分配的記憶體段，直接返回nextAvail值
if (nextAvail >= 0) {
this.nextAvail = -1;
return nextAvail;
}
// 每次分配完成nextAvail被置為-1，這時只能通過方法findNextAvail重新計算出下一個可分配的記憶體段位置
return findNextAvail();
}FindNextAvail查詢下一個可分配記憶體
private int findNextAvail() {
final long[] bitmap = this.bitmap;
final int bitmapLength = this.bitmapLength;
for (int i = 0; i < bitmapLength; i ++) {
long bits = bitmap[i];
// ~bits != 0說明這個long所描述的64個記憶體段還有未分配的
if (~bits != 0) {
return findNextAvail0(i, bits);
}
}
return -1;
}
private int findNextAvail0(int i, long bits) {
final int maxNumElems = this.maxNumElems;
final int baseVal = i << 6;
for (int j = 0; j < 64; j ++) {
// 來判斷該位置是否未分配
// 否則bits右移一位，從左到右遍歷值為0的位置
if ((bits & 1) == 0) {
int val = baseVal | j;
if (val < maxNumElems) {
return val;
} else {
break;
}
}
bits >>>= 1;
}
return -1;
}至此關於Netty的記憶體分配告一段落……