netty5学习笔记-内存池1-PoolChunk

曾经终败给现在 2022-05-28 01:30 154阅读 0赞

转载自：https://blog.csdn.net/youaremoon/article/details/47910971

从netty 4开始，netty加入了内存池管理，采用内存池管理比普通的new ByteBuf性能提高了数(2)十倍。相信有些朋友会和我一样，对他的实现方式很感兴趣。这里把学习netty内存池的过程记录下来，与大家一起分享。

首先给大家介绍的是PoolChunk, 该类主要负责内存块的分配与回收，首先来看看两个重要的术语：

page: 可以分配的最小的内存块单位。

chunk: 一堆page的集合。

下面我们用一张图直观的看下PoolChunk是如何管理内存的：

![Center][]

上图中是一个默认大小的chunk, 由2048个page组成了一个chunk,一个page的大小为8192, chunk之上有11层节点，最后一层节点数与page数量相等。每次内存分配需要保证内存的连续性，这样才能简单的操作分配到的内存，因此这里构造了一颗完整的平衡二叉树，所有子节点的管理的内存也属于其父节点。如果我们想获取一个8K的内存，则只需在第11层找一个可用节点即可，而如果我们需要16K的数据，则需要在第10层找一个可用节点。如果一个节点存在一个已经被分配的子节点，则该节点不能被分配，例如我们需要16K内存，这个时候id为2048的节点已经被分配，id为2049的节点未分配，就不能直接分配1024这个节点，因为这个节点下的内存只有8K了。

通过上面这个树结构，我们可以看到每次内存分配都是8K\*(2^n), 比如需要24K内存时，实际上会申请到一块32K的内存。为了分配一个大小为chunkSize/(2^k)的内存段，需要在深度为k的层从左开始查找可用节点。如想分配16K的内存，chunkSize = 16M, 则k=10, 需要从第10层找一个空闲的节点分配内存。

如何高效的从这么多page中分配到指定的内存呢。来看看下面这个图：

![Center 1][]

这个图与上图结构一致，不同的是上方的二叉树的值为当前的层数，两张图和起来用一个数组memoryMap表示，上面的图中的数字表示数组的index,下面的图中的数字表示当前节点及其子节点可以分配的层的高度。如对于id=512的节点，其深度为9，则：

1）memoryMap\[512\] = 9，则表示其本身到下面所有的子节点都可以被分配；

2）memoryMap\[512\] = val (从10到11), 则表示512节点下有子节点已经分配过，则该节点不能直接被分配，而其子节点中的第val和val以下层还存在未分配的节点;

3）memoryMap\[512\] = 12 (即总层数 + 1）, 可分配的深度已经大于总层数,  则该节点下的所有子节点都已经被分配。

下面我们在从源码分析下PoolChunk是如何实现的，首先看看它的构造方法：

**\[java\]**  [view plain][]  [copy][view plain]

1.  PoolChunk(PoolArena<T> arena, T memory, int pageSize, int maxOrder, int pageShifts, int chunkSize) \{  
2.          unpooled = false;  
3.          this.arena = arena;   
4.          // memory是一个容量为chunkSize的byte\[\](heap方式)或ByteBuffer(direct方式)  
5.          this.memory = memory;  
6.          // 每个page的大小，默认为8192  
7.          this.pageSize = pageSize;  
8.          // 13,   2 ^ 13 = 8192  
9.          this.pageShifts = pageShifts;  
10.         // 默认11  
11.         this.maxOrder = maxOrder;  
12.         // 默认 8192 << 11 = 16MB  
13.         this.chunkSize = chunkSize;  
14.         // 12, 当memoryMap\[id\] = unusable时，则表示id节点已被分配  
15.         unusable = (byte) (maxOrder + 1);  
16.         // 24, 2 ^ 24 = 16M  
17.         log2ChunkSize = log2(chunkSize);  
18.         // -8192  
19.         subpageOverflowMask = ~(pageSize - 1);  
20.         freeBytes = chunkSize;  
21.   
22.         assert maxOrder < 30 : "maxOrder should be < 30, but is: " + maxOrder;  
23.         // 2048, 最多能被分配的Subpage个数  
24.         maxSubpageAllocs = 1 << maxOrder;  
25.   
26.         // Generate the memory map.  
27.         memoryMap = new byte\[maxSubpageAllocs << 1\];  
28.         depthMap = new byte\[memoryMap.length\];  
29.         int memoryMapIndex = 1;  
30.         // 分配完成后，memoryMap->\[0, 0, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3…\]  
31.         // depthMap->\[0, 0, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3…\]  
32.         for (int d = 0; d <= maxOrder; ++ d) \{ // move down the tree one level at a time  
33.             int depth = 1 << d;  
34.             for (int p = 0; p < depth; ++ p) \{  
35.                 // in each level traverse left to right and set value to the depth of subtree  
36.                 memoryMap\[memoryMapIndex\] = (byte) d;  
37.                 depthMap\[memoryMapIndex\] = (byte) d;  
38.                 memoryMapIndex ++;  
39.             \}  
40.         \}  
41.   
42.         // subpages包含了maxSubpageAllocs(2048)个PoolSubpage, 每个subpage会从chunk中分配到自己的内存段，两个subpage不会操作相同的段，此处只是初始化一个数组，还没有实际的实例化各个元素  
43.         subpages = newSubpageArray(maxSubpageAllocs);  
44.     \}

需要注意的是由于memoryMap表示的树中一个包含2 ^ maxOrde - 1个节点，因此memoryMap中的其中一个节点是无用的，为了方便后续的计算，这里将第一个节点作为无用的节点，这样从父节点计算左子结点只需要简单的\*2即可，即left node id = parent id << 1。随着内存不断的分配和回收，memoryMap中的值也不停的更新，而depthMap中保存的值表示各个id对应的深度，是个固定值，初始化后不再变更。

下面看看如何向PoolChunk申请一段内存：

**\[java\]**  [view plain][]  [copy][view plain]

1.  // normCapacity已经处理过  
2.  long allocate(int normCapacity) \{  
3.          if ((normCapacity & subpageOverflowMask) != 0) \{ // >= pageSize  
4.              // 大于等于pageSize时返回的是可分配normCapacity的节点的id  
5.              return allocateRun(normCapacity);  
6.          \} else \{  
7.              // 小于pageSize时返回的是一个特殊的long型handle  
8.              // handle = 0x4000000000000000L | (long) bitmapIdx << 32 | memoryMapIdx;  
9.              // 与上面直接返回节点id的逻辑对比可以知道，当handle<Integer.MAX\_VALUE时，它表示chunk的节点id；  
10.             // 当handle>Integer.MAX\_VALUE，他分配的是一个Subpage,节点id=memoryMapIdx, 且能得到Subpage的bitmapIdx，bitmapIdx后面会讲其用处  
11.             return allocateSubpage(normCapacity);  
12.         \}  
13.     \}

allocateRun与allocateSubpage都需要查询到一个可用的id, allocateSubpage相对allocateRun多出分配PoolSubpage的步骤，下面先看看allocateRun的实现：

**\[java\]**  [view plain][]  [copy][view plain]

1.  private long allocateRun(int normCapacity) \{  
2.      // log2(val) -> Integer.SIZE - 1 - Integer.numberOfLeadingZeros(val)  
3.      // 如normCapacity=8192,log2(8192)=13,d=11  
4.          int d = maxOrder - (log2(normCapacity) - pageShifts);  
5.          // 通过上面一行算出对应大小需要从那一层分配，得到d后，调用allocateNode来获取对应id  
6.          int id = allocateNode(d);  
7.          if (id < 0) \{  
8.              return id;  
9.          \}  
10.         // runLenth=id所在深度占用的字节数  
11.         freeBytes -= runLength(id);  
12.         return id;  
13.     \}

allocateNode(int d)传入的参数为depth, 通过depth来搜索对应层级第一个可用的node id。下面看allocateNode的实现：

**\[java\]**  [view plain][]  [copy][view plain]

1.  private int allocateNode(int d) \{  
2.          int id = 1;  
3.          // 如d=11,则initial=-2048  
4.          int initial = - (1 << d); // has last d bits = 0 and rest all = 1  
5.          // value(id)=memoryMap\[id\]  
6.          byte val = value(id);  
7.          // 第一层的节点的值大于d,表示d层及以上都不可分配，此次分配失败  
8.          if (val > d) \{ // unusable  
9.              return -1;  
10.         \}  
11.         // 这里从第二层开始从上往下查找, 一直找到指定层  
12.         while (val < d || (id & initial) == 0) \{ // id & initial == 1 << d for all ids at depth d, for < d it is 0  
13.             // 往下一层  
14.             id <<= 1;  
15.             val = value(id);  
16.             // 上面对一层节点的值判断已经表示有可用内存可分配，因此发现当前节点不可分配时，  
17.             // 直接切换到父节点的另一子节点，即id ^= 1  
18.             if (val > d) \{  
19.                 id ^= 1;  
20.                 val = value(id);  
21.             \}  
22.         \}  
23.         byte value = value(id);  
24.         assert value == d && (id & initial) == 1 << d : String.format("val = %d, id & initial = %d, d = %d",  
25.                 value, id & initial, d);  
26.         // 该节点本次分配成功，将其标为不可用  
27.         setValue(id, unusable); // mark as unusable  
28.         // 更新父节点的值  
29.         updateParentsAlloc(id);  
30.         return id;  
31.     \}

上面有几个需要注意的点：1、分配的节点在d层；2、每一层都可能遇到节点被分配的情况，此时需要切换到父节点的另一个子节点继续往下查找；3、分配的节点需要标记为不可用，防止后面再被分配；4、分配一个节点后，其父节点的值会发生变更，并可能引起更上层的父节点的变更。下面我们来举个例子看看第4点（depth=10节点的子节点从2个可用变为1个可用的情况）：

![Center 2][]

1） parent的depth=10,表示该层及所有子层可直接分配；

2）left节点被分配，其depth=12, 不能再被分配；

3）由于left节点已经被分配，此时parent节点已经不能直接分配，但还存在一个可分配节点right，此时parent的depth被设置为right的值11；

4）parent不可分配，则其以上的所有父节点的状态可能也会发生变化，此时需要递归的修改更上层parent的值，及指针上移后重复上面的操作。

**\[java\]**  [view plain][]  [copy][view plain]

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.              // 得到左节点和右节点较小的值赋给父节点，即两个节点只要有一个可分配，则父节点的值设为可分配的这个节点的值  
7.              byte val = val1 < val2 ? val1 : val2;  
8.              setValue(parentId, val);  
9.              id = parentId;  
10.         \}  
11.     \}

到这里一个节点的分配就完成了，但是还有另一种情况，即分配的ByteBuf大小小于pageSize, 这种情况会调用allocateSubpage方法：

**\[java\]**  [view plain][]  [copy][view plain]

1.  private long allocateSubpage(int normCapacity) \{  
2.          int d = maxOrder; // subpages are only be allocated from pages i.e., leaves  
3.          int id = allocateNode(d);  
4.          if (id < 0) \{  
5.              return id;  
6.          \}  
7.    
8.          final PoolSubpage<T>\[\] subpages = this.subpages;  
9.          final int pageSize = this.pageSize;  
10.   
11.         freeBytes -= pageSize;  
12.     // 包含数据的节点在最后一层，而最后一层的左边第一个节点index=2048,因此若id=2048则subpageIdx=0，id=2049，subpageIdx=1  
13.     // 根据这个规律找到对应位置的PoolSubpage  
14.         int subpageIdx = subpageIdx(id);  
15.         PoolSubpage<T> subpage = subpages\[subpageIdx\];  
16.         if (subpage == null) \{  
17.             // 如果PoolSubpage未创建则创建一个，创建时传入当前id对应的offset,pageSize,本次分配的大小normCapacity  
18.             subpage = new PoolSubpage<T>(this, id, runOffset(id), pageSize, normCapacity);  
19.             subpages\[subpageIdx\] = subpage;  
20.         \} else \{  
21.             // 已经创建则初始化数据  
22.             subpage.<span style="color:\#009900;">init</span>(normCapacity);  
23.         \}  
24.         // 调用此方法得到一个可以操作的handle  
25.         return subpage.<span style="color:\#009900;">allocate</span>();  
26.     \}

这里可以看到，在小size的对象分配时会先分配一个 PoolSubpage，最终返回一个包含该PoolSubpage信息的handle，后面的操作也是通过此handle进行操作。在看如何操作内存之前，我们先来看看分配一个节点后如何释放（比较简单，所以先讲这里），其对应的方法：

**\[java\]**  [view plain][]  [copy][view plain]

1.  void free(long handle) \{  
2.      // 传入的handle即allocate得到的handle  
3.          int memoryMapIdx = (int) handle;  
4.          int bitmapIdx = (int) (handle >>> Integer.SIZE);  
5.    
6.          if (bitmapIdx != 0) \{ // free a subpage  
7.              // !=0表示分配的是一个subpage  
8.              PoolSubpage<T> subpage = subpages\[subpageIdx(memoryMapIdx)\];  
9.              assert subpage != null && subpage.doNotDestroy;  
10.             if (subpage.free(bitmapIdx & 0x3FFFFFFF)) \{ //subpage释放方法后面再讲  
11.                 return;  
12.             \}  
13.         \}  
14.         freeBytes += runLength(memoryMapIdx);  
15.         // 将节点的值改为可用，及其depth  
16.         setValue(memoryMapIdx, depth(memoryMapIdx));  
17.         // 修改其父节点的值  
18.         updateParentsFree(memoryMapIdx);  
19.     \}

这里会将节点的值改为可用，完成 后该节点就可以再次被分配了，此时父节点的值还未更新，可能会导致分配节点时无法访问到此节点 ，因此还需要同时改变其父节点的值，

**\[java\]**  [view plain][]  [copy][view plain]

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.             //当两个子节点都可分配时，该节点变回自己所在层的depth，表示该节点也可被分配  
11.                setValue(parentId, (byte) (logChild - 1));  
12.            \} else \{  
13.             // 否则与上面的updateParentsAlloc逻辑相同  
14.                byte val = val1 < val2 ? val1 : val2;  
15.                setValue(parentId, val);  
16.            \}  
17.   
18.            id = parentId;  
19.        \}  
20.    \}

PoolChunk本身主要是负责节点的分配与释放，因此节点的分配与释放都了解了对这个类的了解就已经差不多了。但这里还有几个方法没讲到，这几个方法是干什么用的呢，下面我们来看看：

**\[java\]**  [view plain][]  [copy][view plain]

1.      void initBuf(PooledByteBuf<T> buf, long handle, int reqCapacity) \{  
2.          int memoryMapIdx = (int) handle;  
3.          int bitmapIdx = (int) (handle >>> Integer.SIZE);  
4.          if (bitmapIdx == 0) \{  
5.              // 到这里表示分配的是>=pageSize的数据  
6.              byte val = value(memoryMapIdx);  
7.              assert val == unusable : String.valueOf(val);  
8.              buf.init(this, handle, runOffset(memoryMapIdx), reqCapacity, runLength(memoryMapIdx));  
9.          \} else \{  
10.             // 到这里表示分配的是<pageSize的数据  
11.             initBufWithSubpage(buf, handle, bitmapIdx, reqCapacity);  
12.         \}  
13.     \}  
14.   
15.     void initBufWithSubpage(PooledByteBuf<T> buf, long handle, int reqCapacity) \{  
16.         initBufWithSubpage(buf, handle, (int) (handle >>> Integer.SIZE), reqCapacity);  
17.     \}  
18.   
19.     private void initBufWithSubpage(PooledByteBuf<T> buf, long handle, int bitmapIdx, int reqCapacity) \{  
20.         assert bitmapIdx != 0;  
21.   
22.         int memoryMapIdx = (int) handle;  
23.   
24.         PoolSubpage<T> subpage = subpages\[subpageIdx(memoryMapIdx)\];  
25.         assert subpage.doNotDestroy;  
26.         assert reqCapacity <= subpage.elemSize;  
27.   
28.         buf.init(  
29.             this, handle,  
30.             runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) \* subpage.elemSize, reqCapacity, subpage.elemSize);  
31.     \}

通过allocate拿到了handle，相当于得到了一把钥匙，现在就可以拿这把钥匙开启后面的操作了。首先我们拿handle+buf+reqCapacity几个参数来进行实际的分配，前面讲过handle有两种含义，1、handle<Integer.MAX\_VALUE, 表示一个node id; 2、handle>Integer.MAX\_VALUE, 则里面包含node id + 对应的subpage的bitmapIdx。最终传递给PooledByteBuf的信息包括，PoolChunk中的memory: 完整的byte数组或ByteBuf；handle：表示node id的数值，释放内存段的时候使用这个参数； offset：该PooledByteBuf可操作memory的第一个位置； length：该PooledByteBuf本次申请的长度；maxLength：该PooledByteBuf最大可用的长度。PooledByteBuf通过这些参数完成自身的初始化后，就可以开始实际的读写了，它的可读写区域就是memory数组的offset 到 offset + length,  如果写入的数据超过length, 可以扩容至maxLength, 即可读写的区域变为offset 到 offset + maxLength， 如果超过maxLength, 则需要重新申请数据块了，这个后面再说。

到这里PoolChunk的功能基本上已经讲完，下面来回顾下PoolChunk的简单使用过程（真实的使用比较复杂，后面再讲）：

**\[html\]**  [view plain][]  [copy][view plain]

1.  1) 初始化PoolChunk，初始时传入分配好的内存块memory(heap或direct)，通过maxOrder构造完整的平衡二叉树，并且将对应值存入memoryMap;  
2.  2) 应用获取一个大小为size0的内存段，通过调用PoolChunk.allocate方法获取到对应的二叉树节点id并转换为handle，  
3.     如果size0 >= 8192, handle=id,如果size0 < 8192, handle = 0x4000000000000000L | (long) bitmapIdx << 32 | memoryMapIdx；  
4.     获取完成后更新memoryMap, 保证该节点不会重复分配，同时其父节点的可分配状况也级联的改变。  
5.  3) 通过上一步获取到的handle来初始化应用的ByteBuf,调用的方法为PoolChunk.initBuf，ByteBuf会接收memory, handle, offset, length, maxLengh;  
6.  4) 应用对ByteBuf进行写入读取等操作（可能会触发扩容等操作，后面的文章会深入讲；  
7.  5) 应用使用完ByteBuf后释放它，ByteBuf释放时调用PoolChunk.free,传入的参数为handle，PoolChunk将handle对应的节点置为可分配，同时改变其父节点的状态。  
8.     该节点可以再次分配了（其父节点也有可能被分配了）。  
9.  6）下次应用获取内存段时再次进入第二步。

需要注意的是PoolChunk提供了内存池的功能，然后它的所有方法都是线程不安全的，因此需要应用保证其可以被安全的访问，否则会导致节点被重复分配等问题；ByteBuf的设计使内存不使用后不需要手动的将数据置0，直接重置节点状态即可；申请小内存后会通过PoolSubpage进行管理，为什么这么做，下一篇再慢慢讲

[Center]: /images/20220528/f82262320fd94166adce1ac4180d89bf.png
[Center 1]: /images/20220528/06cb6d6c69b44c29bf973a6900ad16b8.png
[view plain]: https://blog.csdn.net/youaremoon/article/details/47910971#
[Center 2]: /images/20220528/a10f82b1742b4f149215cd43eb83f752.png