Netty源码分析--内存模型（下）

Posted 2022-08-02 huxipeng

     这一节我们一起看下分配过程

1     PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) 
2         PooledByteBuf<T> buf = newByteBuf(maxCapacity); // 初始化一块容量为 2^31 - 1的ByteBuf
3         allocate(cache, buf, reqCapacity); // reqCapacity = 1024  进入分配逻辑
4         return buf;
5     

 1     private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) 
 2         final int normCapacity = normalizeCapacity(reqCapacity); // 对请求的大小进行校准，一会具体看
 3         if (isTinyOrSmall(normCapacity))  // capacity < pageSize 判断校准之后的请求大小 是否小于 8k
 4             int tableIdx;
 5             PoolSubpage<T>[] table;
 6             boolean tiny = isTiny(normCapacity); // 判断请求大小是否小于512B
 7             if (tiny)  // < 512
 8                 if (cache.allocateTiny(this, buf, reqCapacity, normCapacity))  // 从缓存中进行分配，如果是第一次这里肯定是没有对应大小缓存的，除非之前有过释放，然后内存空间被缓存起来了。
 9                     // was able to allocate out of the cache so move on
10                     return;
11                 
12                 tableIdx = tinyIdx(normCapacity); // tiny数组中获取对应的下标 
13                 table = tinySubpagePools; // tiny subPage数组
14              else  // 512 <=  normCapacity  < 8K
15                 if (cache.allocateSmall(this, buf, reqCapacity, normCapacity))  // 一样从缓存中进行分配，这个上一节详细的介绍过
16                     // was able to allocate out of the cache so move on
17                     return;
18                 
19                 tableIdx = smallIdx(normCapacity); // 获取small subPage数组下标  
20                 table = smallSubpagePools; // subPage 数组
21             
22 
23             final PoolSubpage<T> head = table[tableIdx];// 取得对应下标的subPage
24 
25             /**
26              * Synchronize on the head. This is needed as @link PoolChunk#allocateSubpage(int) and
27              * @link PoolChunk#free(long) may modify the doubly linked list as well.
28              */
29             synchronized (head)  // 锁定防止其他操作修改head结点
30                 final PoolSubpage<T> s = head.next;
31                 if (s != head) // 如果双向链表已经初始化，那么从subpage中进行分配
32                     assert s.doNotDestroy && s.elemSize == normCapacity; // 断言确保 当前的这个subPage中的elemSize大小必须和校对后的请求大小一样
33                     long handle = s.allocate(); // 从subPage中进行分配
34                     assert handle >= 0;
35                     s.chunk.initBufWithSubpage(buf, handle, reqCapacity);
36                     incTinySmallAllocation(tiny);
37                     return;
38                 
39             
40             synchronized (this) 
41                 allocateNormal(buf, reqCapacity, normCapacity); // 双向循环链表还没初始化，使用normal分配
42             
43 
44             incTinySmallAllocation(tiny);
45             return;
46         
47         if (normCapacity <= chunkSize) 
48             if (cache.allocateNormal(this, buf, reqCapacity, normCapacity)) 
49                 // was able to allocate out of the cache so move on
50                 return;
51             
52             synchronized (this) 
53                 allocateNormal(buf, reqCapacity, normCapacity);
54                 ++allocationsNormal;
55             
56          else 
57             // Huge allocations are never served via the cache so just call allocateHuge
58             allocateHuge(buf, reqCapacity);
59         
60     

    int normalizeCapacity(int reqCapacity) 
        if (reqCapacity < 0) 
            throw new IllegalArgumentException("capacity: " + reqCapacity + " (expected: 0+)");
        

        if (reqCapacity >= chunkSize) 
            return directMemoryCacheAlignment == 0 ? reqCapacity : alignCapacity(reqCapacity);
        

        if (!isTiny(reqCapacity))  //  如果是 >=512 的内存申请，那么就规范到2的次方大小 ，比如 1000 就会 变成 1024，自己可以把这段位操作，自己main方法试一下。
            // Doubled

            int normalizedCapacity = reqCapacity;
            normalizedCapacity --;
            normalizedCapacity |= normalizedCapacity >>>  1;
            normalizedCapacity |= normalizedCapacity >>>  2;
            normalizedCapacity |= normalizedCapacity >>>  4;
            normalizedCapacity |= normalizedCapacity >>>  8;
            normalizedCapacity |= normalizedCapacity >>> 16;
            normalizedCapacity ++;

            if (normalizedCapacity < 0) 
                normalizedCapacity >>>= 1;
            
            assert directMemoryCacheAlignment == 0 || (normalizedCapacity & directMemoryCacheAlignmentMask) == 0;

            return normalizedCapacity;
        

        if (directMemoryCacheAlignment > 0) 
            return alignCapacity(reqCapacity);
        

        // Quantum-spaced
        if ((reqCapacity & 15) == 0)  // 如果 < 512 判断如果是16的倍数，那么直接返回
            return reqCapacity;
        

        return (reqCapacity & ~15) + 16; // 如果不是则对齐成16的倍数
    

 private 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)) 
         return;
       // 先尝试从chunk链表中进行分配， 这里有一个需要注意的地方
     // 分配顺序 先分配 q50 也就是 50% ~ 100% 的，然后是 q25 也就是 25% ~ 75% 的， 然后是 q000 0%~50% 、 qInit 0%~25% 、 q75 75% ~ 100%

     // Add a new chunk.
     PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize); // 新建一个Chunk 
     long handle = c.allocate(normCapacity); // 内存分配
     assert handle > 0;
     c.initBuf(buf, handle, reqCapacity);
     qInit.add(c); // 添加到初始化的chunk链表
 

上面的分配顺序，大家想一下为什么不是从q000开始分配呢？我找了一段分析的很好的。

在分析PoolChunkList的时候，我们知道一个chunk随着内存的不停释放，它本身会不停的往其所在的chunk list的prev list移动，直到其完全释放后被回收。 如果这里是从q000开始尝试分配，虽然分配的速度可能更快了（因为分配成功的几率更大），但一个chunk在使用率为25%以内时有更大几率再分配，也就是一个chunk被回收的几率大大降低了。这样就带来了一个问题，我们的应用在实际运行过程中会存在一个访问高峰期，这个时候内存的占用量会是平时的几倍，因此会多分配几倍的chunk出来，而等高峰期过去以后，由于chunk被回收的几率降低，内存回收的进度就会很慢（因为没被完全释放，所以无法回收），内存就存在很大的浪费。

为什么是从q050开始尝试分配呢，q050是内存占用50%~100%的chunk，猜测是希望能够提高整个应用的内存使用率，因为这样大部分情况下会使用q050的内存，这样在内存使用不是很多的情况下一些利用率低(<50%)的chunk慢慢就会淘汰出去，最终被回收。然而为什么不是从qinit中开始呢，这里的chunk利用率低，但又不会被回收，岂不是浪费？q075,q100由于使用率高，分配成功的几率也会更小，因此放到最后。

上面也有提到新建一个chunk，这里我想特殊说明两个byte数组，数组的大小是4096

        memoryMap = new byte[maxSubpageAllocs << 1];
        depthMap = new byte[memoryMap.length];
        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 ++;
            
        

循环之后的数组是怎么样子的呢？

MemoryMap存放分配信息，depthMap存放节点的高度信息 ；初始状态时，memoryMap和depthMap相等，depthMap的值初始化后不再改变，memoryMap的值则随着节点分配而改变。

下标代表了平衡二叉树的节点序号，值代表该节点的高度值。

 比如MemoryMap[1024] = 10, MemoryMap[2048] = 11. 好了，这个说完了，我们继续看：

1     long allocate(int normCapacity) 
2         if ((normCapacity & subpageOverflowMask) != 0)  // >= pageSize  >= 8K
3             return allocateRun(normCapacity);
4          else 
5             return allocateSubpage(normCapacity); // < 8K
6         
7     

 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); // 根据请求内存大小，获取下标，然后获取对应的subPage
 5         synchronized (head) 
 6             int d = maxOrder; // 最大层数 11
 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;// 16M（初始化值，后面会逐渐减少） - 8K
16 
17             int subpageIdx = subpageIdx(id); // 2048 对应的偏移量是 0 ， 2049 对应的偏移量是 1 。。。4095 对应偏移量是 2047 
18             PoolSubpage<T> subpage = subpages[subpageIdx]; // 取对应下标的 subpage
19             if (subpage == null) 
20                 subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity); // 创建PoolSubpage
21                 subpages[subpageIdx] = subpage;
                   // 新建或初始化subpage并加入到chunk的subpages数组，同时将subpage加入到arena的subpage双向链表中，最后完成分配请求的内存。
                   // 代码中，subpage != null的情况产生的原因是：subpage初始化后分配了内存，但一段时间后该subpage分配的内存释放并从arena的双向链表中删除，
                   // 此时subpage不为null，当再次请求分配时，只需要调用init()将其加入到areana的双向链表中即可。
    
22              else 
23                 subpage.init(head, normCapacity);
24             
25             return subpage.allocate(); // 最后结果是一个long整数，其中低32位表示二叉树中的分配的节点，高32位表示subPage中分配的具体位置
26         
27     

 修改节点对应的层数，底下这个方法涉及了很多的位运算，这里大家要仔细琢磨。

 1     private int allocateNode(int d)  // d = 11
 2         int id = 1;
 3         int initial = - (1 << d); // -2048
 4         byte val = value(id); // memoryMap[id] , 一般1对应的层数是 0 ,当第一个节点被分配完成后，这个节点的值会变成12
 5         if (val > d)  // unusable // 如果分配完成，那么无法进行分配，那么就是 12 > 11 ，直接返回 -1 
 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; // 左移1位
10             val = value(id); // 取到节点对应的层数
11             if (val > d)  // 如果当前的值大于层数，也就是说左子节点用完了
12                 id ^= 1; // 当前节点值 +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 //将该节点的值设置为12，不可用
20         updateParentsAlloc(id); // 将该节点对应的所有父节点，层数全部 + 1
21         return id; 
22     

 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); // 如果 id = 2048 那么 id ^ 1 = 2049 ，  如果 id = 2049 这里  id ^ 1 = 2048 这里一定要注意！！！
 6             byte val = val1 < val2 ? val1 : val2; 
 7             setValue(parentId, val);  // 修改父级对应层数
 8             id = parentId; 
 9         
10     

上面这个方法很有意思，大家要仔细体会，我举个例子

 最开始是这样的，我们开始第一次分配，那么就是2048号节点，分配完成后会变成

 

 这时候由于2048变成了12，不可用状态，那么取右节点进行分配

    下一次再进行分配的时候， 发现 1024 是 12 ，那么取右子节点， 是 1025 = 10 ，因为 10 < 11 所以再次进入循环，找到2050节点，对应层数是 11 < 12 ，则分配 2050节点分配后变成

  

 然后依次类推。

1     private long allocateRun(int normCapacity) 
2         int d = maxOrder - (log2(normCapacity) - pageShifts); // 计算满足需求的节点的高度 比如 16K , 那么 就是 d= 11 - (14 - 13) = 10
3         int id = allocateNode(d); // 从第十层找空闲节点， 原理跟上面相同，就不说了
4         if (id < 0) 
5             return id;
6         
7         freeBytes -= runLength(id); // 分配后剩余的字节数
8         return id;
9     

说到这里，内存模型的重点部分算是说完了。如果整个Netty系列有任何不对的地方，欢迎大家指出，我会虚心改正。

      后面呢，我将打算将Spring的源码的关键部分进行分析，期间也会穿插着对Redis的底层数据结构等重要内容进行分析。有需要的童鞋们，记得关注我，随时可以获取最新的消息。

谢谢。