Netty内存池之PoolChunk

Posted 2021-04-30 Linux内核之旅

netty4之后，netty中加入了内存池管理，看了netty后突然对这块很感兴趣，怎么办？最简单的方式无非看源码呗。我们顺着思路强行分析一波。分析下简单需求，为了能够简单的操作内存，必须保证每次分配到的内存时连续的。我们来看下netty的poolChunk。

PoolChunk

该类主要负责内存块的分配与回收，我们可以从源码中给的注释差不多看清PoolChunk的数据结构与算法。先介绍两个术语：

page-是可以分配的内存块的最小单位
chunk-是一堆page的集合
下面是chunk的构造方法

1. PoolChunk(PoolArena<T> 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 < 30, but is: " + maxOrder;  

22.        maxSubpageAllocs = 1 << maxOrder;  

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 << d;  

30.            for (int p = 0; p < depth; ++ 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 << d;  

6.             for (int p = 0; p < depth; ++ 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 << d); // 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 < d || (id & initial) == 0) { // id & initial == 1 << d for all ids at depth d, for < d it is 0  

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 << d : String.format("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 < val2 ? val1 : val2;  

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<T> 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<T>[] subpages = this.subpages;  

13.             final int pageSize = this.pageSize;  

14.   

15.             freeBytes -= pageSize;  

16.   

17.             int subpageIdx = subpageIdx(id);  

18.             PoolSubpage<T> subpage = subpages[subpageIdx];  

19.             if (subpage == null) {  

20.                 subpage = new PoolSubpage<T>(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<T> 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<T> 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 < val2 ? val1 : val2;  

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<T> 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<T> 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等等。