本系列：

上一節中分析瞭如何在poolChunk中分配一塊大於pageSize的記憶體，但在實際應用中，存在很多分配小記憶體的情況，如果也佔用一個page，明顯很浪費。針對這種情況，Netty提供了PoolSubpage把poolChunk的一個page節點8k記憶體劃分成更小的記憶體段，通過對每個記憶體段的標記與清理標記進行記憶體的分配與釋放。

PoolSubpage

final class PoolSubpage<T> { // 當前page在chunk中的id private final int memoryMapIdx; // 當前page在chunk.memory的偏移量 private final int runOffset; // page大小 private final int pageSize; //通過對每一個二進位制位的標記來修改一段記憶體的佔用狀態 private final long[] bitmap; PoolSubpage<T> prev; PoolSubpage<T> next; boolean doNotDestroy; // 該page切分後每一段的大小 int elemSize; // 該page包含的段數量 private int maxNumElems; private int bitmapLength; // 下一個可用的位置 private int nextAvail; // 可用的段數量 private int numAvail; ... }

12345678910111213141516171819202122232425

finalclassPoolSubpage<T>{// 當前page在chunk中的idprivatefinalintmemoryMapIdx;// 當前page在chunk.memory的偏移量privatefinalintrunOffset;// page大小privatefinalintpageSize;//通過對每一個二進位制位的標記來修改一段記憶體的佔用狀態privatefinallong[]bitmap;PoolSubpage <T>prev;PoolSubpage<T>next;booleandoNotDestroy;// 該page切分後每一段的大小intelemSize;// 該page包含的段數量privateintmaxNumElems;privateintbitmapLength;// 下一個可用的位置privateintnextAvail;// 可用的段數量privateintnumAvail;...}

假設目前需要申請大小為4096的記憶體：

long allocate(int normCapacity) { if ((normCapacity & subpageOverflowMask) != 0) { // >= pageSize return allocateRun(normCapacity); } else { return allocateSubpage(normCapacity); } }

1234567

longallocate(intnormCapacity){if((normCapacity&subpageOverflowMask)!=0){// >= pageSizereturnallocateRun(normCapacity);}else{returnallocateSubpage(normCapacity);}}

因為 4096 < pageSize(8192)，所以採用 allocateSubpage 進行記憶體分配，具體實現如下：

private long allocateSubpage(int normCapacity) { // Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it. // This is need as we may add it back and so alter the linked-list structure. PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity); synchronized (head) { int d = maxOrder; // subpages are only be allocated from pages i.e., leaves int id = allocateNode(d); if (id < 0) { return id; } final PoolSubpage<T>[] subpages = this.subpages; final int pageSize = this.pageSize; freeBytes -= pageSize; int subpageIdx = subpageIdx(id); PoolSubpage<T> subpage = subpages[subpageIdx]; if (subpage == null) { subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity); subpages[subpageIdx] = subpage; } else { subpage.init(head, normCapacity); } return subpage.allocate(); } }

123456789101112131415161718192021222324252627

privatelongallocateSubpage(intnormCapacity){// Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.// This is need as we may add it back and so alter the linked-list structure.PoolSubpage<T>head=arena.findSubpagePoolHead(normCapacity);synchronized(head){intd=maxOrder;// subpages are only be allocated from pages i.e., leavesintid=allocateNode(d);if(id<0){returnid;}finalPoolSubpage<T>[]subpages=this.subpages;finalintpageSize=this.pageSize;freeBytes-=pageSize;intsubpageIdx=subpageIdx(id);PoolSubpage<T>subpage=subpages[subpageIdx];if(subpage==null){subpage=newPoolSubpage<T>(head,this,id,runOffset(id),pageSize,normCapacity);subpages[subpageIdx]=subpage;}else{subpage.init(head,normCapacity);}returnsubpage.allocate();}}

1、Arena負責管理PoolChunk和PoolSubpage；
2、allocateNode負責在二叉樹中找到匹配的節點，和poolChunk不同的是，只匹配葉子節點；
3、poolChunk中維護了一個大小為2048的poolSubpage陣列，分別對應二叉樹中2048個葉子節點，假設本次分配到節點2048，則取出poolSubpage陣列第一個元素subpage；
4、如果subpage為空，則進行初始化，並加入到poolSubpage陣列；

subpage初始化實現如下：

PoolSubpage(PoolSubpage<T> head, PoolChunk<T> chunk, int memoryMapIdx, int runOffset, int pageSize, elemSize) { this.chunk = chunk; this.memoryMapIdx = memoryMapIdx; this.runOffset = runOffset; this.pageSize = pageSize; bitmap = new long[pageSize >>> 10]; // pageSize / 16 / 64 init(head, elemSize); }

123456789101112

PoolSubpage(PoolSubpage<T>head,PoolChunk<T>chunk,intmemoryMapIdx,intrunOffset,intpageSize,elemSize){this.chunk=chunk;this.memoryMapIdx=memoryMapIdx;this.runOffset=runOffset;this.pageSize=pageSize;bitmap=newlong[pageSize>>>10];// pageSize / 16 / 64init(head,elemSize);}

1、預設初始化bitmap長度為8，這裡解釋一下為什麼只需要8個元素：其中分配記憶體大小都是處理過的，最小為16，說明一個page可以分成8192/16 = 512個記憶體段，一個long有64位，可以描述64個記憶體段，這樣只需要512/64 = 8個long就可以描述全部記憶體段了。
2、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); }

1234567891011121314151617

voidinit(PoolSubpage<T>head,intelemSize){doNotDestroy=true;this.elemSize=elemSize;if(elemSize!=0){maxNumElems=numAvail=pageSize/elemSize;nextAvail=0;bitmapLength=maxNumElems>>>6;if((maxNumElems&63)!=0){bitmapLength++;}for(inti=0;i<bitmapLength;i++){bitmap[i]=0;}}addToPool(head);}

下面通過分佈申請4096和32大小的記憶體，說明如何確定bitmapLength的值：

1. 比如，當前申請大小4096的記憶體，maxNumElems 和 numAvail 為2，說明一個page被拆分成2個記憶體段，2 >>> 6 = 0，且2 & 63 ！= 0，所以bitmapLength為1，說明只需要一個long就可以描述2個記憶體段狀態。
2. 如果當前申請大小32的記憶體，maxNumElems 和 numAvail 為 256，說明一個page被拆分成256個記憶體段， 256>>> 6 = 4，說明需要4個long描述256個記憶體段狀態。

下面看看subpage是如何進行記憶體分配的：

long allocate() { if (elemSize == 0) { return toHandle(0); } if (numAvail == 0 || !doNotDestroy) { return -1; } final int bitmapIdx = getNextAvail(); int q = bitmapIdx >>> 6; int r = bitmapIdx & 63; assert (bitmap[q] >>> r & 1) == 0; bitmap[q] |= 1L << r; if (-- numAvail == 0) { removeFromPool(); } return toHandle(bitmapIdx); }

123456789101112131415161718192021

longallocate(){if(elemSize==0){returntoHandle(0);}if(numAvail==0||!doNotDestroy){return-1;}finalintbitmapIdx=getNextAvail();intq=bitmapIdx>>>6;intr=bitmapIdx&63;assert(bitmap[q]>>>r&1)==0;bitmap[q]|=1L<<r;if(--numAvail==0){removeFromPool();}returntoHandle(bitmapIdx);}

1、方法getNextAvail負責找到當前page中可分配記憶體段的bitmapIdx；
2、q = bitmapIdx >>> 6，確定bitmap陣列下標為q的long數，用來描述 bitmapIdx 記憶體段的狀態；
3、bitmapIdx & 63將超出64的那一部分二進位制數抹掉，得到一個小於64的數r；
4、bitmap[q] |= 1L << r將對應位置q設定為1；

如果以上描述不直觀的話，下面換一種方式解釋，假設需要分配大小為128的記憶體，這時page會拆分成64個記憶體段，需要1個long型別的數字描述這64個記憶體段的狀態，所以bitmap陣列只會用到第一個元素。

狀態轉換

getNextAvail如何實現找到下一個可分配的記憶體段？

private int getNextAvail() { int nextAvail = this.nextAvail; if (nextAvail >= 0) { this.nextAvail = -1; return nextAvail; } return findNextAvail(); }

12345678

privateintgetNextAvail(){intnextAvail=this.nextAvail;if(nextAvail>=0){this.nextAvail=-1;returnnextAvail;}returnfindNextAvail();}

1、如果nextAvail大於等於0，說明nextAvail指向了下一個可分配的記憶體段，直接返回nextAvail值；
2、每次分配完成，nextAvail被置為-1，這時只能通過方法findNextAvail重新計算出下一個可分配的記憶體段位置。

private int findNextAvail() { final long[] bitmap = this.bitmap; final int bitmapLength = this.bitmapLength; for (int i = 0; i >>= 1; } return -1; }

1234567

privateintfindNextAvail(){finallong[]bitmap=this.bitmap;finalintbitmapLength=this.bitmapLength;for(inti=0;i>>=1;}return-1;}

1、~bits != 0說明這個long所描述的64個記憶體段還有未分配的；
2、(bits & 1) == 0 用來判斷該位置是否未分配，否則bits又移一位，從左到右遍歷值為0的位置；

至此，subpage記憶體段已經分配完成。

打賞支援我寫出更多好文章，謝謝！

打賞作者

打賞支援我寫出更多好文章，謝謝！