netty源码阅读之ByteBuf之内存page级别内存的分配

我不是女神ヾ 2022-05-14 13:16 290阅读 0赞

我们继续回到PoolArena的分配内存的方法：

private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) {
            final int normCapacity = normalizeCapacity(reqCapacity);
            if (isTinyOrSmall(normCapacity)) { // capacity < pageSize
                int tableIdx;
                PoolSubpage<T>[] table;
                boolean tiny = isTiny(normCapacity);
                if (tiny) { // < 512
                    if (cache.allocateTiny(this, buf, reqCapacity, normCapacity)) {
                        // was able to allocate out of the cache so move on
                        return;
                    }
                    tableIdx = tinyIdx(normCapacity);
                    table = tinySubpagePools;
                } else {
                    if (cache.allocateSmall(this, buf, reqCapacity, normCapacity)) {
                        // was able to allocate out of the cache so move on
                        return;
                    }
                    tableIdx = smallIdx(normCapacity);
                    table = smallSubpagePools;
                }
    
                final PoolSubpage<T> head = table[tableIdx];
    
                /**
                 * Synchronize on the head. This is needed as {@link PoolChunk#allocateSubpage(int)} and
                 * {@link PoolChunk#free(long)} may modify the doubly linked list as well.
                 */
                synchronized (head) {
                    final PoolSubpage<T> s = head.next;
                    if (s != head) {
                        assert s.doNotDestroy && s.elemSize == normCapacity;
                        long handle = s.allocate();
                        assert handle >= 0;
                        s.chunk.initBufWithSubpage(buf, handle, reqCapacity);
    
                        if (tiny) {
                            allocationsTiny.increment();
                        } else {
                            allocationsSmall.increment();
                        }
                        return;
                    }
                }
                allocateNormal(buf, reqCapacity, normCapacity);
                return;
            }
            if (normCapacity <= chunkSize) {
                if (cache.allocateNormal(this, buf, reqCapacity, normCapacity)) {
                    // was able to allocate out of the cache so move on
                    return;
                }
                allocateNormal(buf, reqCapacity, normCapacity);
            } else {
                // Huge allocations are never served via the cache so just call allocateHuge
                allocateHuge(buf, reqCapacity);
            }
        }

之前的一篇文章我们通过cache.allocateTiny()介绍了[缓存的分配流程][Link 1]，这篇文章，我们以allocateNormal为例，介绍page级别的内存分配。

一个chunk是16M，一个page是8k，normal大小的内存，介于8k到16M之间，allocateNormal方法的名称由此来。

进入allocateNormal这个方法：

private synchronized void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
            if (q050.allocate(buf, reqCapacity, normCapacity) || q025.allocate(buf, reqCapacity, normCapacity) ||
                q000.allocate(buf, reqCapacity, normCapacity) || qInit.allocate(buf, reqCapacity, normCapacity) ||
                q075.allocate(buf, reqCapacity, normCapacity)) {
                ++allocationsNormal;
                return;
            }
    
            // Add a new chunk.
            PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize);
            long handle = c.allocate(normCapacity);
            ++allocationsNormal;
            assert handle > 0;
            c.initBuf(buf, handle, reqCapacity);
            qInit.add(c);
        }

这个allocateNormal的流程，可以分为以下三段分析：

1、尝试在现有的chunk上面分配内存。

2、如果现有的chunk分配不成功，那就创建一个chunk进行内存分配

3、初始化PooledByteBuf，也就是将PooledByteBuf需要的内存指向chunk的内存。

### 一、尝试在现有的chunk上面进行分配，代码是下面这一段： ###

if (q050.allocate(buf, reqCapacity, normCapacity) || q025.allocate(buf, reqCapacity, normCapacity) ||
                q000.allocate(buf, reqCapacity, normCapacity) || qInit.allocate(buf, reqCapacity, normCapacity) ||
                q075.allocate(buf, reqCapacity, normCapacity)) {
                ++allocationsNormal;
                return;
            }

第一次分配内存的时候，肯定是不会走到这一步的，因为chunk还没有实例化，所以会直接走下面一步，但是我们从这里开始分析，进入q050.allocate(buf,reqCapacity,normCapacity)：

boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
            if (head == null || normCapacity > maxCapacity) {
                // Either this PoolChunkList is empty or the requested capacity is larger then the capacity which can
                // be handled by the PoolChunks that are contained in this PoolChunkList.
                return false;
            }
    
            for (PoolChunk<T> cur = head;;) {
                long handle = cur.allocate(normCapacity);
                if (handle < 0) {
                    cur = cur.next;
                    if (cur == null) {
                        return false;
                    }
                } else {
                    cur.initBuf(buf, handle, reqCapacity);
                    if (cur.usage() >= maxUsage) {
                        remove(cur);
                        nextList.add(cur);
                    }
                    return true;
                }
            }
        }

我们从q050的这个chunklist的第一个chunk开始进行分配，也就是调用这里long handle = cur.allocate(normCapacity);，如果handle是小于0的，那就说明分配不成功，继续寻找下一个节点进行分配；如果分配成功，就调用cur.initBuf()进行分配，也就是我们第三点要分析的，分配完之后，如果使用率超过最大使用率，就放到下一个chunklist里面去。这里最终不会调用到initBuf方法，cur==null，直接return false。

### 二、创建一个chunk进行分配 ###

这里的代码就是一下两段：

// Add a new chunk.
            PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize);
            long handle = c.allocate(normCapacity);

这一回，我们以debug的方式，分析内存的分配，首先这里是用户端代码：

public class Test {
        public static void main(String[] args) {
            int page = 1024*8;
            PooledByteBufAllocator allocator = PooledByteBufAllocator.DEFAULT;
            ByteBuf byteBuf = allocator.directBuffer(2 * page);
        }
    }

debug到达这里：

![70][]

查看这里的变量：

![70 1][]

pagesize=8192，也就是我们刚刚要分配的8k内存

maxOrder=11，代表一共有11层，稍后我们会分析。

pageShifts=13，2^13=8192，稍后我们分析

chunkSize=15777216，也就是16M，也就是一个chunkSize的大小

step into进入函数：

@Override
            protected PoolChunk<ByteBuffer> newChunk(int pageSize, int maxOrder, int pageShifts, int chunkSize) {
                return new PoolChunk<ByteBuffer>(
                        this, allocateDirect(chunkSize),
                        pageSize, maxOrder, pageShifts, chunkSize);
            }

看allocateDirect(chunkSize)：

private static ByteBuffer allocateDirect(int capacity) {
                return PlatformDependent.useDirectBufferNoCleaner() ?
                        PlatformDependent.allocateDirectNoCleaner(capacity) : ByteBuffer.allocateDirect(capacity);
            }

我们看PlatformDependent.useDirectBufferNoCleaner()的值，返回true

![70 2][]

我们知道使用的是这种规格就行了，它内部就是使用的jdk方法创建一个直接内存ByteBuffer。

然后继续进入PoolChunk的构造函数：

![70 3][] 这里使用的是pooled，所以unpooled为false。需要哪一块arena，哪一块memory都传入进去，之前讲解过的pageSize和pageShifts和maxOrder和chunkSIze也传进去。unusable为12，比最大的层数大1，通过这个条件，如果某个值赋值为12，我们就知道他不可用，后面分析我们就知道了。

1右移11位，maxSubpageAllocs是2048

maxSubpageAllocs右移1位是4096，所以整个memoryMap和depthMap的大小为4096大小，我们看这一段：

int memoryMapIndex = 1;
            for (int d = 0; d <= maxOrder; ++ d) { // move down the tree one level at a time
                int depth = 1 << d;
                for (int p = 0; p < depth; ++ p) {
                    // in each level traverse left to right and set value to the depth of subtree
                    memoryMap[memoryMapIndex] = (byte) d;
                    depthMap[memoryMapIndex] = (byte) d;
                    memoryMapIndex ++;
                }
            }

我们用下面这个图辅助我们理解：

![70 4][]

一个memoryMap是一个完全二叉树，里面有11层，对应maxOrder；memoryMapIndex就是从上往下、从左往右数下去的第几个节点；d就是树的深度，p就是树的每一层的数量。depthMap和memoryMap一样分析。最终我们会分配到一个memoryMap，一共有4096个节点，每个节点的值为树的深度，也就是memoryMap=\{0,1,1,2,2,2,2,3,3,3,3,3,3,3,3，...，...11\}。

然后， 每一层有代表的数据的大小：

![70 5][]

最后我们总结一下：

![846961-20180726234953241-661643120.png][]

[参考][Link 2]

上面代码的逻辑及就是我们刚刚说的，最终我们会分配到一个memoryMap，一共有4096个节点，每个节点的值为树的深度，也就是memoryMap=\{0,1,1,2,2,2,2,3,3,3,3,3,3,3,3，...，...11\}。

创建完chunk之后，我们开始为chunk分配数据，也就是一开始的long handle = c.allocate(normCapacity);,断点到这里：

![70 6][]

normCapacity大小为16M，对的。

进入：

![70 7][]

run呢，我们大致解释为一个片段吧，一个片段其实就是page。这里我们介绍的是page的，所以进入了allocateRun里面，如果下一篇文章介绍subpage的分配，那就是进入allocateSubpage。然后进入allocateRun：

![70 8][]

通过一定的算法，计算出d为10，也就是在第十层分配：

![70 9][]

memoryMap的第10层为0~16k，16k~32k的块，这里我们知道，第一块就可以用来分配内存了。

我们进入allocateNode(d)吧：

![70 10][]

d=10，我们从id为1开始找，找到合适的id，也就是memory数组里面的第id位，这个id位就是我们可以分配的内存，通过上面的图，我们可以计算出着16k的内存适合分配在memory的第1024位，也是正确的。看value(id)吧，

private byte value(int id) {
            return memoryMap[id];
        }

也就是读出memory第id位的值。

下面一段代码：

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

把我们这个memoryMap里面，第id位也就是1024位的值设置为unusable，也就是设置值为12，12永远小于最大层数11，所以这个做法说的过去。以后判断到位12的位，就说明使用过了，方便查找未使用的片段。

下面这段代码呢，也是挺有意思的：

updateParentsAlloc(id);

继续深入看：

private void updateParentsAlloc(int id) {
            while (id > 1) {
                int parentId = id >>> 1;
                byte val1 = value(id);
                byte val2 = value(id ^ 1);
                byte val = val1 < val2 ? val1 : val2;
                setValue(parentId, val);
                id = parentId;
            }
        }

我们知道，memoryMap是以二叉树的形式组成的，这里做的就是将父节点也标记为使用，然后再递归调用父节点的父节点也标记为不能使用（也就是已使用），通过分析上面的代码可以很容易的得出这个结论，函数名也是这个意思。为什么要这样做呢。如果不这样做，下次要分配一块8M的内存，到了第二层发现没有被使用，直接使用了第二层的第一块，这是不符合要求的。

这个新建chunk并且分配的逻辑。

### 三、初始化PooledByteBuf ###

这里的作用就是将PooledByteBuf需要的内存指向chunk的内存。最后一步是从这里开始的：

c.initBuf(buf, handle, reqCapacity);

继续断点调试：

![70 11][]

看到这个handle为1024，就惊喜地诠释了一个chunk和一个handle可以唯一确定一块内存，handle的值其实是chunk的memoryMap数组的索引，惊不惊喜意不意外？

然后我么看bitmapIdx(handle)是做什么的：

private static int bitmapIdx(long handle) {
            return (int) (handle >>> Integer.SIZE);
        }

Integer.SIZE就是整形数据占用空间的大小，也就是32.

handle为1024，右移32必须等于0了，只要是normal类型数据，也就是page类型的handle，右移之后一定都是等于0的。而subpage也就是我们下一篇文章要分析的，就是不等于0的。

所以我们会进入bitmap==0的这个分支，而subpage的会进入initBufWithSubpage的分支：

![70 12][]

我们先稍微看下下面两个表达式的值：

![70 13][]

runOffset为0，表明没有偏移量。

![70 14][]

runLength是16k，表明我们第1024个节点的大小为16k。

进入buf.init()：

![70 15][]

继续：

![70 16][]

我们看到，这里就是把chunk的内存分配给bytebuf，this.chunk=chunk，代表指向哪一块chunk，handle，指向的chunk里面的哪一个位置，memory，哪一块内存，偏移量offset为0（page级别分配没有偏移量，subpage才有），大小length16384也就是16k。分配完毕回来这里：

![70 17][] initMemoryAddress()，由于是unsafe的DirectByteBuf，所以要初始化memoryAddress：

private void initMemoryAddress() {
            memoryAddress = PlatformDependent.directBufferAddress(memory) + offset;
        }

前面也已经讲解过。

最后回到这里，把分配的内存加到qInit这个chunklist里面。

void add(PoolChunk<T> chunk) {
            if (chunk.usage() >= maxUsage) {
                nextList.add(chunk);
                return;
            }
            add0(chunk);
        }

[Link 1]: https://blog.csdn.net/fst438060684/article/details/82459573
[70]: /images/20220514/7315284298ff4e03bb88d1ebaeb161aa.png
[70 1]: /images/20220514/fe870852be744da587ce7e2d5dfa4760.png
[70 2]: /images/20220514/e0e7e2a0cd4e4909bd468c6f5e0c3d51.png
[70 3]: /images/20220514/cb33135dc34e4e9581236e25ed6affd2.png
[70 4]: /images/20220514/c87d707e217c41c2a344306b117319b3.png
[70 5]: /images/20220514/06554dcc7de947acae105e7340098de1.png
[846961-20180726234953241-661643120.png]: /images/20220514/ae719398947847a79003f1d6f1eb0b0b.png
[Link 2]: https://www.cnblogs.com/sunshine-2015/p/9375157.html
[70 6]: /images/20220514/3ee9fcfe72174b639fa6eaf73cc22c1c.png
[70 7]: /images/20220514/c1697287fe9047399d25fd725bd572ec.png
[70 8]: /images/20220514/67c7c32e630141a2879179604189cbbe.png
[70 9]: /images/20220514/ef061137954e43e8a45e69d9aee4110e.png
[70 10]: /images/20220514/8451bba701ca415892dfff0b88dc703c.png
[70 11]: /images/20220514/1e4b379e8ae140279d12035db16a96e1.png
[70 12]: /images/20220514/bf3fb41528f444db9cef0b4d5e070bae.png
[70 13]: /images/20220514/719a1aa068244fbb953904a249b471ff.png
[70 14]: /images/20220514/f59258983bd24453b08e4d04f2d87994.png
[70 15]: /images/20220514/3e34c831d0264966b609a0aae2553b58.png
[70 16]: /images/20220514/70bf2a93f2fe457c81fb386e5799ac82.png
[70 17]: /images/20220514/299497f9e12f4ca29ec8835a539fe575.png