Netty記憶體池之PoolChunk

netty4之後，netty中加入了記憶體池管理，看了netty後突然對這塊很感興趣，怎麼辦？最簡單的方式無非看原始碼唄。我們順著思路強行分析一波。分析下簡單需求，為了能夠簡單的操作記憶體，必須保證每次分配到的記憶體時連續的。我們來看下netty的poolChunk。

PoolChunk

該類主要負責記憶體塊的分配與回收，我們可以從原始碼中給的註釋差不多看清PoolChunk的資料結構與演演算法。先介紹兩個術語：

page-是可以分配的記憶體塊的最小單位
chunk-是一堆page的集合
下麵是chunk的構造方法

1. PoolChunk(PoolArena arena, T memory, int pageSize, int maxOrder, int pageShifts, int chunkSize, int offset) {  

2.        unpooled = false;  

3.        this.arena = arena;  

4.        // memory是一個容量為chunkSize的byte[](heap方式)或ByteBuffer(direct方式)  

5.        this.memory = memory;  

6.        // 每個page的大小，預設為8192,8k  

7.        this.pageSize = pageSize;  

8.        // 13  

9.        this.pageShifts = pageShifts;  

10.        // 11層  

11.        this.maxOrder = maxOrder;  

12.        // 8k * 2048  

13.        this.chunkSize = chunkSize;  

14.        this.offset = offset;  

15.        unusable = (byte) (maxOrder + 1);  

16.        log2ChunkSize = log2(chunkSize);  

17.        // -8192  

18.        subpageOverflowMask = ~(pageSize – 1);  

19.        freeBytes = chunkSize;  

20.   

21.        assert maxOrder 30 : “maxOrder should be  + maxOrder;  

22.        maxSubpageAllocs = 1 <

23.   

24.        // Generate the memory map.  

25.        memoryMap = new byte[maxSubpageAllocs <1];  

26.        depthMap = new byte[memoryMap.length];  

27.        int memoryMapIndex = 1;  

28.        for (int d = 0; d <= maxOrder; ++ d) { // move down the tree one level at a time  

29.            int depth = 1 <

30.            for (int p = 0; p 

31.                // in each level traverse left to right and set value to the depth of subtree  

32.                memoryMap[memoryMapIndex] = (byte) d;  

33.                depthMap[memoryMapIndex] = (byte) d;  

34.                memoryMapIndex ++;  

35.            }  

36.        }  

37.   

38.        subpages = newSubpageArray(maxSubpageAllocs);  

39.    }  

可能光看構造我們會摸不清頭腦，我去網上挖了張圖，方便大家理解。

Netty中PoolChunk內部維護一棵滿平衡二叉樹memoryMap，所有子節點管理的記憶體也屬於其父節點。這麼理解，chunk是真正分配記憶體的空間，它裡面的page就是最小可分配的單元，我們發現平衡二叉樹的葉子節點(11層)與page一一對應，再往上走，比如1024節點，它指向2048跟2049，那無非1024節點指向了page0跟page1兩個儲存單元。我們繼續看下圖

這圖跟上圖的區別無非把節點的id換成了層數。其實完全二叉樹天生就可以用陣列來表示，於是id跟層數對應起來可以用陣列memoryMap表示。陣列下標為id，值為層數。depthMap也是這樣存的，但它初始化後資料不變了，僅僅查表作用。poolChunk維護memoryMap主要用來標記使用情況。

1. 1) memoryMap[id] = depth_of_id  => it is free / unallocated  

2.  * 2) memoryMap[id] > depth_of_id  => at least one of its child nodes is allocated, so we cannot allocate it, but  

3.  *                                    some of its children can still be allocated based on their availability  

4.  * 3) memoryMap[id] = maxOrder + 1 => the node is fully allocated & thus none of its children can be allocated, it  

5.  *                                    is thus marked as unusable  

順便舉個例子吧，我們看下id為512的節點，它層數為9，此時memoryMap[512]=9，說明512節點下麵的page一個都沒分配，如果此時我們分配了一塊page1(隨便抽的)，那麼按照演演算法，memoryMap[2049]=12，memoryMap[1024]=11，memoryMap[512]=10。一直更新上去。後面會具體從程式碼層面分析。

道理講了這麼多，看程式碼吧。建構式之前都是簡單賦值，大小我也註釋都標上了，我們看下下麵這塊程式碼，其實邏輯也很簡單，無非初始化memoryMap跟depthMap，此時可以註意到memoryMap的第一個節點即memoryMap[0]=0，這個元素是空出來的，下標為1的元素才是代表第一個節點，這樣也有好處，下標為i的父元素的左孩子就是2*i(i<<1)，右孩子呢? 2*i+1 ((i<<1)^1)。位運算效率總是討人喜歡的。

1. memoryMap = new byte[maxSubpageAllocs <1];  

2.         depthMap = new byte[memoryMap.length];  

3.         int memoryMapIndex = 1;  

4.         for (int d = 0; d <= maxOrder; ++ d) { // move down the tree one level at a time  

5.             int depth = 1 <

6.             for (int p = 0; p 

7.                 // in each level traverse left to right and set value to the depth of subtree  

8.                 memoryMap[memoryMapIndex] = (byte) d;  

9.                 depthMap[memoryMapIndex] = (byte) d;  

10.                 memoryMapIndex ++;  

11.             }  

12.         }  

13.   

14.         subpages = newSubpageArray(maxSubpageAllocs);  

最後無非賦初值subpages。

poolChunk的資料結構差不多說清楚了，具體來看下如何向PoolChunk申請空間的

allocate()函式

1. long allocate(int normCapacity) {  

2.     if ((normCapacity & subpageOverflowMask) != 0) {  

3.         return allocateRun(normCapacity);  

4.     } else {  

5.         return allocateSubpage(normCapacity);  

6.     }  

7. }  

一開始的位運算就用的很討人喜歡，(normCapacity & subpageOverflowMask) != 0，normCapacity無非是申請空間的位元組數大小，我們可以從建構式中看到 subpageOverflowMask = ~(pageSize – 1); 那麼subpageOverflowMask大小為-8192(光看數值看不出所以然，它在記憶體中是最高位跟低13位全為0，剩下全為1)那麼很容易想到，這句話意思無非等價於normCapacity >= pageSize;

然後分兩種情況，當申請大小大於pageSize，那麼一塊page肯定不夠，於是需要多塊，傳回的是可分配normCapacity大小的節點的id；第二中情況，一個page已經夠分了，但是實際應用中會存在很多小塊記憶體的分配，如果小塊記憶體也佔用一個page明顯浪費，於是page也可以繼續拆分了，這是後話。我們先來看下allocateRun()

1. private long allocateRun(int normCapacity) {  

2.     int d = maxOrder – (log2(normCapacity) – pageShifts);  

3.     int id = allocateNode(d);  

4.     if (id 0) {  

5.         return id;  

6.     }  

7.     freeBytes -= runLength(id);  

8.     return id;  

9. }  

d表示申請normCapacity大小對應的節點的深度，透過很簡單的運算即(normCapacity/8k)再取對數，拿maxOrder(11)減一下。

1. private long allocateRun(int normCapacity) {  

2.     int d = maxOrder – (log2(normCapacity) – pageShifts);  

3.     int id = allocateNode(d);  

4.     if (id 0) {  

5.         return id;  

6.     }  

7.     freeBytes -= runLength(id);  

8.     return id;  

9. }  

d表示申請normCapacity大小對應的節點的深度，透過很簡單的運算即(normCapacity/8k)再取對數，拿maxOrder(11)減一下。

1. private int allocateNode(int d) {  

2.         int id = 1;  

3.         int initial = – (1 <// has last d bits = 0 and rest all = 1  

4.         byte val = value(id);  

5.         if (val > d) { // unusable  

6.             return –1;  

7.         }  

8.         while (val 0) { // id & initial == 1 <  

9.             id <<= 1;  

10.             val = value(id);  

11.             if (val > d) {  

12.                 id ^= 1;  

13.                 val = value(id);  

14.             }  

15.         }  

16.         byte value = value(id);  

17.         assert value == d && (id & initial) == 1 <“val = %d, id & initial = %d, d = %d”,  

18.                 value, id & initial, d);  

19.         setValue(id, unusable); // mark as unusable  

20.         updateParentsAlloc(id);  

21.         return id;  

22.     }  

id賦初值為1，從根節點開始找，如果(val > d)第一層的節點的值大於d,表示d層及以上都不可分配，說明當前Chunk取不出那麼大的空間，此次分配失敗。一直在樹上找直到找到與d恰好匹配的節點，即高度不小於d並且最小可分配，id<<=1往下一層走；當前節點對應大小不夠分配，則id^=1切換到父節點的另一子節點中(若父節點夠分配，則此節點肯定能分配)；找到對應節點id後，把當前節點的值設為不可用，並且遞迴迴圈調整父節點的MemoryMap的值

1. private void updateParentsAlloc(int id) {  

2.     while (id > 1) {  

3.         int parentId = id >>> 1;  

4.         byte val1 = value(id);  

5.         byte val2 = value(id ^ 1);  

6.         byte val = val1 

7.         setValue(parentId, val);  

8.         id = parentId;  

9.     }  

10. }  

程式碼邏輯很簡單，父節點值設為子節點中的最小值。直到根節點，調整結束。

我們再看當申請大小小於pageSize的第二種情況：

1. private long allocateSubpage(int normCapacity) {  

2.         // Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.  

3.         // This is need as we may add it back and so alter the linked-list structure.  

4.         PoolSubpage head = arena.findSubpagePoolHead(normCapacity);  

5.         synchronized (head) {  

6.             int d = maxOrder; // subpages are only be allocated from pages i.e., leaves  

7.             int id = allocateNode(d);  

8.             if (id 0) {  

9.                 return id;  

10.             }  

11.   

12.             final PoolSubpage[] subpages = this.subpages;  

13.             final int pageSize = this.pageSize;  

14.   

15.             freeBytes -= pageSize;  

16.   

17.             int subpageIdx = subpageIdx(id);  

18.             PoolSubpage subpage = subpages[subpageIdx];  

19.             if (subpage == null) {  

20.                 subpage = new PoolSubpage(head, this, id, runOffset(id), pageSize, normCapacity);  

21.                 subpages[subpageIdx] = subpage;  

22.             } else {  

23.                 subpage.init(head, normCapacity);  

24.             }  

25.             return subpage.allocate();  

26.         }  

27.     }  

針對這種情況，可以將8K的page拆成更小的塊，這已經超出chunk的管理範圍了，這個時候就出現了PoolSubpage, 透過normCapacity大小找到對應的PoolSubpage頭結點，然後加鎖(分配過程會改變連結串列結構)，d=maxOrder，因此此時只需一塊page足夠，所以d直接設定為maxOrder即葉子節點就ok，透過allcNode()分配到節點的id，如果id小於0表示分配失敗(葉子節點分配完了)傳回。之後找到最高的合適節點後，開始新建(並加入到poolChunk的subpages陣列中)或初始化subpage，然後呼叫subpage的allocate()申請空間，傳回了個handle;下一篇博文會將subpage的具體邏輯。

我們再來看下釋放空間，free()函式。

1. void free(long handle) {  

2.         int memoryMapIdx = memoryMapIdx(handle);  

3.         int bitmapIdx = bitmapIdx(handle);  

4.   

5.         if (bitmapIdx != 0) { // free a subpage  

6.             PoolSubpage subpage = subpages[subpageIdx(memoryMapIdx)];  

7.             assert subpage != null && subpage.doNotDestroy;  

8.   

9.             // Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.  

10.             // This is need as we may add it back and so alter the linked-list structure.  

11.             PoolSubpage head = arena.findSubpagePoolHead(subpage.elemSize);  

12.             synchronized (head) {  

13.                 if (subpage.free(head, bitmapIdx & 0x3FFFFFFF)) {  

14.                     return;  

15.                 }  

16.             }  

17.         }  

18.         freeBytes += runLength(memoryMapIdx);  

19.         setValue(memoryMapIdx, depth(memoryMapIdx));  

20.         updateParentsFree(memoryMapIdx);  

21.     }  

其中傳入的handle無非是allocate得到的handle，透過handle得到page的id，如果透過handle計算得到的bitmapIdx為0，則表示是之前申請空間的第一種條件，大小大於pageSize，於是直接釋放整個節點對應的page空間，則將其memoryMap[id]=其對應的層數（depth[id]）,於是調整父親節點，此時是申請時的逆過程。

1. private void updateParentsFree(int id) {  

2.         int logChild = depth(id) + 1;  

3.         while (id > 1) {  

4.             int parentId = id >>> 1;  

5.             byte val1 = value(id);  

6.             byte val2 = value(id ^ 1);  

7.             logChild -= 1; // in first iteration equals log, subsequently reduce 1 from logChild as we traverse up  

8.   

9.             if (val1 == logChild && val2 == logChild) {  

10.                 setValue(parentId, (byte) (logChild – 1));  

11.             } else {  

12.                 byte val = val1 

13.                 setValue(parentId, val);  

14.             }  

15.   

16.             id = parentId;  

17.         }  

18.     }  

如果左右孩子value值一樣，則父節點value設為孩子的value-1，否則設為孩子節點的最小值。

第二種情況，則需釋放subpage再繼續上面介紹的釋放page。

1. if (bitmapIdx != 0) { // free a subpage  

2.             PoolSubpage subpage = subpages[subpageIdx(memoryMapIdx)];  

3.             assert subpage != null && subpage.doNotDestroy;  

4.   

5.             // Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.  

6.             // This is need as we may add it back and so alter the linked-list structure.  

7.             PoolSubpage head = arena.findSubpagePoolHead(subpage.elemSize);  

8.             synchronized (head) {  

9.                 if (subpage.free(head, bitmapIdx & 0x3FFFFFFF)) {  

10.                     return;  

11.                 }  

12.             }  

13.         }  

這兒有涉及到subpage部分不是講得很清，沒關係，以poolChunk為netty記憶體池的切入點。接下來會，向下細挖subpage，向上探索poolArea等等。