PoolArena实现了用于高效分配和释放内存，并尽可能减少内存碎片的内存池，这个内存管理实现使用PageRun/PoolSubpage算法。分析代码之前，先熟悉一些重要的概念:
- page: 页，一个页是可分配的最小的内存块单元，页的大小:pageSize = 1 << n (n <= 12)。
- chunk: 块，块是多个页的集合。chunkSize是块中所有page的pageSize之和。
- Tiny: <512B的内存块。
- Small: >=512B, <pageSize的内存块。
- Normal: >=pageSize, <=chunkSize的内存块。
- Huge: >chunkSize的内存块。
PoolArena维护了一个PoolChunkList组成的双向链表，每个PoolChunkList内部维护了一个PoolChunk双向链表。分配内存时，PoolArena通过在PoolChunkList找到一个合适的PoolChunk，然后从PoolChunk中分配一块内存。
关键属性
pageSize: page的大小。必须满足 pageSize = 1 << n (n>=12)。
maxOrder: 完全平衡二叉树的高度。
chunkSize: chunk的大小。chunkSize = pageSize * (1 << maxOrder)。
memory: chunk的内存，大小必须>=chunkSize。
offset: chunk内存在memory中的起始位置。 memory大小必须>=offet+chunkSize。
page管理
chunk以完全平衡二叉树的数据结构管理page, 这颗树的节点以堆的方式保存在数组中, 如果这棵树的高度maxOrder=4, 它的结构如下图所示:
图-1
节点名字格式是d-i, d是节点在树中的深度，i是节点在数组中的索引。
它有如下一些性质:
- 任意一个节点i, i的取值范围是: [1, 1 << (maxOrder + 1) )。i == 1节点是根节点。
- 如果节点i在区间[1 << maxOrder，1 << (maxOrder +1) ), 那么这些节点都是叶节点。
- 除叶节点以外的节点i, i << 1是它的左子节点，(i << 1) + 1 是它的右子节点。除根节点以外的节点i, i >> 1是它的父节点, i ^ 1是它的另外一个兄弟节点。
- 对于一个节点i, 它树中的深度d = log2(i) (d是整数)。 d相同的节点位于树中的同一层上，他们包含相同的页节点数，有相同的最大可分配内存。
- 任意节点i, 深度为d, 如果把同一层的节点放在一个单独的数组中，那么节点i在这个数据组中的偏移量doffset=i ^ (1 << d)。
- 任意节点i, 深度为d, 它包含的页节点的数量是1 << (maxOrder - d), 内存大小是(1 << (maxOrder - d)) * pageSize。
- 已知深度d, [1 << d, 1 << (d + 1) )区间内的所有节点的深度都是d。
- 任意节点i, 深度d，在memory中的起始位置偏移量是offset + （1 ^ (1 << d) * (1 << (maxOrder - d)) * pageSize。
请记住这些性质。PoolChunk的代码很简洁，可是如果不熟悉这些性质，这些简洁的代码也会难以理解。
完全平衡二叉树的初始化
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 this.memory = memory; 5 this.pageSize = pageSize; 6 this.pageShifts = pageShifts; 7 this.maxOrder = maxOrder; 8 this.chunkSize = chunkSize; 9 this.offset = offset; 10 unusable = (byte) (maxOrder + 1); 11 log2ChunkSize = log2(chunkSize); 12 subpageOverflowMask = ~(pageSize - 1); 13 freeBytes = chunkSize; 14 15 assert maxOrder < 30 : "maxOrder should be < 30, but is: " + maxOrder; 16 maxSubpageAllocs = 1 << maxOrder; 17 18 // Generate the memory map. 19 memoryMap = new byte[maxSubpageAllocs << 1]; 20 depthMap = new byte[memoryMap.length]; 21 int memoryMapIndex = 1; 22 for (int d = 0; d <= maxOrder; ++ d) { // move down the tree one level at a time 23 int depth = 1 << d; 24 for (int p = 0; p < depth; ++ p) { 25 // in each level traverse left to right and set value to the depth of subtree 26 memoryMap[memoryMapIndex] = (byte) d; 27 depthMap[memoryMapIndex] = (byte) d; 28 memoryMapIndex ++; 29 } 30 } 31 32 subpages = newSubpageArray(maxSubpageAllocs); 33 }
在构造方法中，19-30行初始化了两棵完全一样的完全平衡二叉树(形如图-1): memoryMap, depthMap。这两个map都是以数组的方式保存二叉树，数组的长度都是maxSubpageAllocs << 1, 由于maxSubpageAllocs = 1 << maxOrder, 因此长度还可以表示为 1 << (maxOrder + 1)。 map数组的0项保留，[1, 1 << maxOrder)区间中的每个项是二叉树的一个节点，每个项的值是节点在树中的深度。
depthMap用来记录每个节点在树中的深度，初始化之后，值不会发生变化。已知一个节点在数组中的索引id, 可以使用这个id查找节点在树中的深度: depthMap[id]。
memoryMap用来记录树中节点被分配出去的情况，每个项的值会随着节点分配情况变化而变化。已知一个节点在数组中的索引id，memoryMap[id]的值会有三中情况:
- memoryMap[id] == depth[id]: 所有子节点都没被分配出去。
- memoryMap[id] > depth[id]: 至少有一个子节点被分配出去了, 还有可以分配的子节点。
- memoryMap[id] == maxOrder + 1: 这个节点以及完全被分配出去了，没有可分配的子节点了。
从二叉树中分配一个内存大小合适的节点
1 long allocate(int normCapacity) { 2 if ((normCapacity & subpageOverflowMask) != 0) { // >= pageSize 3 return allocateRun(normCapacity); 4 } else { 5 return allocateSubpage(normCapacity); 6 } 7 }
这个方法是分配内存节点的入口方法，参数normCapacity必须满足normCapacity = 1 << n。第2行判断normCapacity和pageSize的大小关系，在前面的构造方法中，subpageOverflowMask = ~(pageSize - 1), 如果pageSize=2048, subpageOverflowMask的0-11位是0， 12-31位是1，它的二进制值是: 1111111111111111111100000000000, (normCapacity & subpageOverflowMask) != 0表示，normCapacity的12-31位中至少有一位是1，此时它>=pageSize, 反之比pageSize小。
如果normCapacity >= pageSize, 调用allocateRun分配一个深度d < maxOrder的节点。
如果normaCapacity < pageSize, 调用allocateSubpage分配一个d == maxOrder的叶叶节点, 即一个page。
PoolChunk分配内存的最小单元是一个page，不能分配比一个page更小的内存了。
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 }
第2行，计算normCapacity大小的内存在二叉树的最大深度d, 只有深度<=d的节点才有可以分配到>=normCapacity的内存。normCapacity可以表示为normCapacity = 2k, log2(normCapacity)就是已知normCapacity求解k。pageShifts可表示为pageSize = 2pageShifts, pageShifts = log2(pageSize)。 normCapacity在二叉树上的反向深度 rd = log2(mormCapacity) - pageShifts, 这个表达式比较难以理解，这样会更加直观一些:
pageCount = normCapacity >> log2(pageSize)
rd = log2(pageCount)
pageCount是normCapacity需要的page数量。 反向深度的含义是，d=0对应二叉树的最大深度maxOrder, d=1对应maxOrder -1，依次类推。因此maxOrder - rd会得到最大深度d，d <= maxOrder。
第3行，如果能够根据d找到一个合适的节点，就会把这个节点记录为已经使用的状态，然后返回这个节点的索引id, id的取值区间是[0, 1 << maxOrder)。
第7行，重新计算剩余内存数。
rungLength方法用于计算节点id的内存长度:
private int runLength(int id) { // represents the size in #bytes supported by node 'id' in the tree return 1 << log2ChunkSize - depth(id); }
log2ChunkSize=log2(chunkSize)在构造方法中初始化。 有性质(6)可以得到节点id的长度 length = (1 << maxOrder - depth(id)) * pageSize，它和代码中表达式是等价的，推导过程如下:
已知:
log2ChunkSize = log2(chunkSize)
chunkSize = (1 << maxOrder) * pageSize
pageSize = 2k = 1 << k
=> chunkSize = (1 << maxOrder) * 2k
= 2maxOrder * 2k
= 2maxOrder + k
=> log2ChunkSize = log2(chunkSize)
= log2(2maxOrder + k)
= maxOrder + k
=> log2ChunkSize - depth(id) = maxOrder + k - depth(id)
=> 1 << log2ChunkSize - depth(id) = 1 << maxOrder + k - depth(id)
= (1 << maxOrder - depth(id)) * (1 << k)
= (1 << maxOrder - depth(id)) * pageSize
如果需要的内存>=pageSize, 就会调用allocateNode方法，这个方法的作用是从二叉树中分配一个节点，返回值id是这个节点的索引。
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 }
allocateNode方法的功能是从memoryMap树中深度[1, d]的节点中找出一个没有被分配出去的节点，然后把这个节点记录为已分配的状态。寻找顺序是自上而下，从左到到右。
第2行，从第一个节点开始，这个节点是二叉树的根节点。
第3行，计算一个32位initial，它的[0, d)位都是0，[d, 31]位都是1。
第4-6行，检查是否可以分配一个深度<=d节点， 如果不能分配内存失败，返回-1。 val == maxOrder + 1时表示这个节点的内存已经被分配完了，val在[0, maxOrder]区间内时，表示可以分配一个深度在[val, maxOder]区间内的节点。所以在第5行检查到val>d时表示不能分配到内存了。
8-15行，能够运行到第8行，说明在这个chunk中，二叉树中一定至少有一个节点满足深度等于d, 且没有任何子节点被分配出去的节点。循环，满足 val < d或(id & initial) == 0会增加一个深度继续寻找。也就是说如果满足val == d 且 (id & initial) == 1时，表示找到了符合调条件的节点了。第9行，增加一个深度。 第10，11行检查左节点。 12,13行检查右节点。
19行, 把选中的节点id, 设置成unusable(maxOrder+1)状态。
20行，更新所有父节点的值。
这个方法展示了已知memoryMap中索引为id的值val = memoryMap[id], 找到一个深度为d的空闲节点的算法。前面已经讲过val值的三种情况，其中第2中情况的时候，表示只有节点id下面只能找到深度>=val的空闲节点，索引d<val情况下，无法找到满足深度等于d的空闲节点。影响memoryMapy[id]值的算法在updateParentsAlloc中实现:
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 }
3行，得到id的父节点。
4-6行，取memoryMap中，取节点id和它的兄弟节点的值中交小的一个，如果相等的话就随意取一个。
7行，把上一步中的取值设置到父节点上。
8，2行，深度减1，重复这个过程直到根节点为止。
分配一个小于pageSize的子页subpage
当需要分配的内存小于pageSize时，仍然会分配一个page，因为PoolChunk能分配的最小内存单元是一个page。这时候只需分配一个也节点就可以了。
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 }
6-10行，分配一个深度d=maxOrder的叶节点。
17,18行，从subpages取出一个PoolSubpage缓存。subpages在构造方法中初始化，subpages = new PoolSubpage[maxSubpageAllocs], maxSubpageAllocs = 1 << maxOrder。subpages的长度就是chunk中的page数量。
19-24行，如果缓存中没有，创建一个新的。如果有直接初始PoolSubpage。
25行，分配一个子页。
关于PoolSubpage子页面管理的功能，后面会详细分析，这里只涉及和PoolChunk相关的内容。
释放内存
分配内存成功后会返回一个long型的handle，64位的handle被分为两部分，[0, 32)位是二叉树中的节点索引，可以使用memoryMapIdx(handle)方法取出。[32, 64)位是PoolSubpage中子页面的索引，可以使用bitMapIdx(handler)方法取出。释放一个handle时，可能需要同时释放二叉树中的节点和PoolSubpage中子页面，free(int handle)方法实现了这个内存释放过程:
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 }
2,3行，分别取出二叉树的节点id和PoolSubpage中子页的id。
5-17行，释放PoolSubpage子页。子页内存被释放之后，subpages数组中仍然保存着PoolSubpages对象。13行只有subpage中所有的子页都释放完了才会释放subpage持有的page。
18-20行，释放二叉树中的节点。调用setValue把被释放的节点memoryMap值设置成它原本的深度depth(memoryMapIdx)。 调用updateParentsFree, 修改memoryMap记录，这个方法实现了updateParentsAlloc的逆过程。
updateParentsFree释放二叉树节点的关键，如果一个节点被释放，它的父节点在memoryMap值可能会发生变化。这个方法的实现如下:
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 }
第2行，计算节点id的子节点深度logChild。
第3行，确保id不是根节点。
第4行，得到父节点id。
第5，6行，得到节点id及其兄弟节点memoryMap值: val1, val2。
第7行，把logChild变成id的深度。
第9,10行, 如果id及其兄弟节点的指定都是depth(id)，表示这两个节都已经完全释放，把父节点的指定还原成depth(parentId) == logChild -1 。
第12,13行，如果id及其兄弟节点至少有一个没有完全释放，把较小的值设置到父节点上。
第16行，深度上移，继续上面的过程。
使用分配的内存初始化PooledByteBuf
使用allocate分配内存得到一个handle之后，需要调用PooledByteBuf的init方法使用handle对应的内存初始化。初始化的关键是计算出handle对应的内存在memory中的偏移量和长度。前面讲的lenthRun可以计算出内存的长度，剩下的就是计算内存偏移量方法runOffset。PoolChunk的initBuf方法用来初始化一个PooledByteBuf对象:
1 void initBuf(PooledByteBuf<T> buf, long handle, int reqCapacity) { 2 int memoryMapIdx = memoryMapIdx(handle); 3 int bitmapIdx = bitmapIdx(handle); 4 if (bitmapIdx == 0) { 5 byte val = value(memoryMapIdx); 6 assert val == unusable : String.valueOf(val); 7 buf.init(this, handle, runOffset(memoryMapIdx) + offset, reqCapacity, runLength(memoryMapIdx), 8 arena.parent.threadCache()); 9 } else { 10 initBufWithSubpage(buf, handle, bitmapIdx, reqCapacity); 11 } 12 }
第2,3行，在分析free代码中解释过。
第4-8行，表示这块内存是二叉树中的一个节点，直接使用init方法初始化。runOffset的算法是 (memoryMapIdx ^ 1 << depth(memoryMapIdx)) * runLength(memoryMapIdx), 根据性质(5)可知，memoryMapIdx ^ depth(memoryMapIdx) 是节点memoryMepIdx在深度为depth(memoryMapIdx)层上的偏移量doffset, 即这一层前面还有doffset个节点，根据性质(4)可知每个节点的内存大小是runLength(memoryMapIdx)，所以doffset * runLength(memoryMapIdx)是节点memoryMapIdx在chunk内存上的偏移量。最后还要再加上一个offset，它是chuk在memory上的偏移量。
第10行，表示这块内存是一个subpage，使用initBufWithSubpage初始化。
1 void initBufWithSubpage(PooledByteBuf<T> buf, long handle, int reqCapacity) { 2 initBufWithSubpage(buf, handle, bitmapIdx(handle), reqCapacity); 3 } 4 5 private void initBufWithSubpage(PooledByteBuf<T> buf, long handle, int bitmapIdx, int reqCapacity) { 6 assert bitmapIdx != 0; 7 8 int memoryMapIdx = memoryMapIdx(handle); 9 10 PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)]; 11 assert subpage.doNotDestroy; 12 assert reqCapacity <= subpage.elemSize; 13 14 buf.init( 15 this, handle, 16 runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize + offset, 17 reqCapacity, subpage.elemSize, arena.parent.threadCache()); 18 }
关键部分在第二个重载方法。的第14-17行。这个计算内存偏移量的算法是runOffst(memoryMapIdx) + offset + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize，它可以拆分成两部分:
memoryMapIdx表示的page在内存中的偏移量pageOffset = runOffset(memoryMapIdx) + offset
子页面subpage在page中的偏移量: subpOffset = (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize
其中subpOffset是个陌生的东西，会在后面PoolSubpage相关章节详细分析。