Netty学习之旅----源码分析内存分配与释放原理

 2020-01-09 16:26  阅读(1613)
文章分类:架构设计

Netty 内存分配优先从线程本地缓冲池中分配,然后才是从 PoolChunk 中分配。线程池缓存的不仅是使用的内存,比如分配的内存区域(PoolSubpage),PooledByteBuf 对象本身也被缓存,关于本地线程分配缓存,下文会有专题进行研究与学习。

1、源码分析Netty内存分配(不考虑线程本地分配)

1.1 入口方法:PooledByteBufAllocator#newHeapBuffer

protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
            PoolThreadCache cache = threadCache.get(); //@1
            PoolArena<byte[]> heapArena = cache.heapArena; //@2
            ByteBuf buf;
            if (heapArena != null) {
                buf = heapArena.allocate(cache, initialCapacity, maxCapacity); //@3
            } else {
                buf = new UnpooledHeapByteBuf(this, initialCapacity, maxCapacity);
            }
            return toLeakAwareBuffer(buf);
    }

内存分配的入口为PooleadByteBufAllocator.newHeapBuffer。

代码@1:线程的分配缓冲区,理解为一个TheadLocal变量,为了提高并发减少锁竞争。由于第一次分配时,线程本地缓存池目前为空,故在这里先不关注,下文专门出一个专题。

代码@2:在讲解PoolArena的时候,会初始化一个PoolArena[]数组,每个线程轮询PoolArena数组,在整个线程运行过程中,只会与一个特定的PoolArena打交道。

代码@3:进入到PoolArena的allocate方法去分配内存。

1.2 PoolArena 的 allocate 方法

PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) {
            PooledByteBuf<T> buf = newByteBuf(maxCapacity); //@1
            allocate(cache, buf, reqCapacity); //@2
            return buf;
    }

1.2.1 PoolArena allocatede 代码@1:newByteBuf:先创建一个PooledByteBuf,这里有个亮点如下。

static PooledHeapByteBuf newInstance(int maxCapacity) {
            PooledHeapByteBuf buf = RECYCLER.get();
            buf.setRefCnt(1);
            buf.maxCapacity(maxCapacity);
            return buf;
        }

这里有个RECYLER.get(),在这里首先先知道是Netty关于ByteBuf的一个轻量级的对象池的实现,Netty会在本地线程变量中重复利用释放掉的ByteBuf,关于本地线程缓存机制,请看下文关于本地线程内存分配专题讲解。创建一个PooledByteBuf对象后,然后进入到PoolArena.allocate方法,为创建的PooledByteBuf对象分配内存(PooledByteBuf内部缓存区的)。

代码@2,PoolArena.allocate(cache, buf, reqCapacity)。

1.2.2 PoolArena关于allocate代码@2 allocate方法详解

private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) {
            final int normCapacity = normalizeCapacity(reqCapacity); //@1
            if (isTinyOrSmall(normCapacity)) { // capacity < pageSize //@2
                int tableIdx;
                PoolSubpage<T>[] table;
                if (isTiny(normCapacity)) { // < 512 //@21
                    if (cache.allocateTiny(this, buf, reqCapacity, normCapacity)) {//@211
                        // was able to allocate out of the cache so move on
                        return;
                    }
                    tableIdx = tinyIdx(normCapacity);
                    table = tinySubpagePools;
                } else { //@22
                    if (cache.allocateSmall(this, buf, reqCapacity, normCapacity)) {
                         //@221
                        // was able to allocate out of the cache so move on
                        return;
                    }
                    tableIdx = smallIdx(normCapacity);
                    table = smallSubpagePools;
                }
                synchronized (this) { //@3
                    final PoolSubpage<T> head = table[tableIdx];
                    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);
                        return;
                    }
                }
            } else if (normCapacity <= chunkSize) { // @4
                if (cache.allocateNormal(this, buf, reqCapacity, normCapacity)) {
                    // was able to allocate out of the cache so move on
                    return;
                }
            } else { //@5
                // Huge allocations are never served via the cache so just call allocate // Huge
                allocateHuge(buf, reqCapacity);
                return;
            }
            allocateNormal(buf, reqCapacity, normCapacity); //@6
        }

代码@1:先对reqCapacity进行处理,将reqCapacity转换成一个合适的容量tiny,small内存,tiny内存为16,32,48,到512,tiny内存为[0,512),而small内存为1024,2048,4096成倍增长,直到(pageSize-1),该方法已经在源码分析内存分配前置篇中有过解读,在此不做重复讲解。

代码@2:申请内存小于pageSize。

代码@4:申请内存小于chunkSize。

代码@5:申请内存大于chunkSize。

代码@21,@22:从PoolArena中会缓存将tiny,small内存按照内存大小进行组织。相同大小的tiny,small会以链表的形式存储。首先尝试从本地线程申请,申请失败后,才会从全局内存中进行分配。对于第一次内存申请,最后会走代码@6。由于已经对PoolArena,PoolChunk的数据结构等都讲解清楚了,故该方法不难理解,故先重点关注allocateNormal方法。

1.2.2.1关于PoolArena allocate的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)
                  || q100.allocate(buf, reqCapacity, normCapacity)) {
                return;
            } // @1
            // Add a new chunk.
            PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSiz e); //@2
            long handle = c.allocate(normCapacity); //@3
            assert handle > 0;
            c.initBuf(buf, handle, reqCapacity); //@4
            qInit.add(c); //@5
        }

代码@1:首先从PoolArena的PoolChunkList中去选择一个合适的PoolChunk进行分配,前文已经详细介绍PoolArena按照PoolChunk的使用率区间来组织。PoolArena按照使用率区间有如下PoolChunkList,qInt(Integer.MIN_VALUE,25)--->q000(1,50)--->q025(25-75)-->q050(50-100)-->q075(75,100)--->q100(100-Integer.MAX_VALUE) 。然后每个PoolChunkList维护一个PoolChunk的链表,这里的PoolChunk的使用率区间是一致的,当随着使用率的增加,PoolChunk沿着上面的链表从一个PoolChunkList进入到下一个PoolChunkList。下面将罗列出这块代码,PoolChunkList的allocate方法,只是负责选出一个PoolChunk,然后具体的内存分配还是有PoolChunk来分配,故不做重点讲解。

代码@2:创建一个新的PoolChunk,创建具体的内存,堆PoolArena的创建有不同的子类PoolArena.HeapArena和PoolArena.DirectArena实现。

代码@3:从PoolChunk中分配内存,重点关注。根据下文的讲解,原来PoolChunk的allocate方法返回只是一个long类型的数据,其高32位保存的是待分配的PoolSubpage的bitmapIdx,低32位保存的是PoolChunk的memoryMap的下标id。当然,目前所有的分析还是申请的内存小于PageSize。也就是在PoolSubpage中完成内存分配。

代码@4:将分配的内存与PooledByteBuf对象建立起连接。这步是真正为PooledByteBuf关联内存。详细解读请看关注下文2.4。

代码@5:将创建的PoolChunk加入到PoolArena的PoolChunkList中。

1.3 PoolChunk的allocate分配方法详解

long allocate(int normCapacity) {
            if ((normCapacity & subpageOverflowMask) != 0) { // >= pageSize
                return allocateRun(normCapacity); //@2
            } else {
                return allocateSubpage(normCapacity); //@1
            }
        }

代码@1:分配内存大于pageSize,这里返回的是memoryMap的id。

代码@2:分配内存小于PageSize,这里返回的是一个long类型的变量,高32位表示PoolSubpage中的bitmap中的bitmapIdx,表示第多少个内存区域,低32位表示memoryMap的id。

1.3.1 PoolChunk allocateSubpage

分配小于pageSize的内存。

/**
         * Create/ initialize a new PoolSubpage of normCapacity
         * Any PoolSubpage created/ initialized here is added to subpage pool in the PoolArena that owns this PoolChunk
         *
         * @param normCapacity normalized capacity
         * @return index in memoryMap
         */
        private long allocateSubpage(int normCapacity) {
            int d = maxOrder; // subpages are only be allocated from pages i.e., leaves
            int id = allocateNode(d); //@1
            if (id < 0) {
                return id;
            }
            final PoolSubpage<T>[] subpages = this.subpages;
            final int pageSize = this.pageSize;
            freeBytes -= pageSize;
            int subpageIdx = subpageIdx(id); // @2
            PoolSubpage<T> subpage = subpages[subpageIdx];
            if (subpage == null) {
                subpage = new PoolSubpage<T>(this, id, runOffset(id), pageSize, normCapacity); // 代码@3
                subpages[subpageIdx] = subpage;
            } else { //@4
                subpage.init(normCapacity);
            }
            return subpage.allocate(); //代码@5
        }

首先讲解一下整体分配思路,先根据PoolChunk内部维护的各个PoolChunk的占用情况,返回一个可以PoolChunk的id,这里的id就是memoryMap数组的下标。然后从PoolChunk维护的PoolSubpage数组对象中获取一个与之对应的PoolSubpage,如果为空,先创建一个PoolSubpage。然后再在PoolSubpage中分配内存。如果已经存在PoolSubpage,则直接分配。

代码@1:首先,我们需要从meomryMap中寻找一个可供分配的PoolSubpage。memoryMap就是记录整个PoolSubpage的占用情况。

代码@2:根据我们现在掌握的知识应该知道,memoryMap[ 2的maxOrder]存放的是第一个PoolSubpage,memoryMap[ 2的 (maxOrder+1)的幂再减去1]存放的是最后一个PoolSubpage,所以接下来我们需要将id转换为实际的PoolSubpage。从上面的讲解,其实 id 与 2的maxOrder的偏移,,比如基数2048对应0,2049对应1,基数是2048,根据id求数组中的下标应该清楚了吧,给出如下三个算法实现,请看下午的关于subpageIdx方法的详解。

代码@3:如果id对应的PoolSubpage没有被初始化,则新建一个PoolSubpage。在内存分配前置篇的时候特意讲解了PoolSubpage几个关键属性,当然包括这里的runOffset;在这里再重复一遍:在整个Netty内存管理中,其实真正持有内存的类是PoolChunk, 如果是直接内存,就是 java.nio.ByteBuffer memory;如果是堆内存的话,就是byte[] memory,我们以堆内存来讲解,比较直观,一个PoolChunk一旦创建,就会分配内存 byte[] memory = new byte[chunkSize];然后PoolChunk由一个一个的PoolSubpage组成,也就是PoolSubpage[] subpages;那们我们如果保存 memoryMap[id]位置代表的PoolSubPage的内存呢?使用一个偏移量,相对于PoolChunk的byte[] meomry的偏移量。那怎么计算呢?先得出id所在平衡二叉树的深度,就能得到该层有多个节点,所谓的偏移量,就是针对同级的。当然,真正的偏移量主要还是针对PoolSubpage,整个Netty的内存管理,真正涉及到内存的只有PoolChunk,PoolSubpage维护一个偏移量,其实就是一个指针,表示PoolChunk的T memory的(从runOffset 至 (runOffset+pageSize)被这个PoolSubpage占用。

代码@5:PoolSubpage 的具体分配方法。

1.3.1.1 关于代码@1PoolChunk allocateSubpage 的allocateNode的详细解读如下

private int allocateNode(int d) {
            int id = 1;
            int initial = - (1 << d); // has last d bits = 0 and rest all = 1
            byte val = value(id);
            if (val > d) { // unusable
                return -1;
            }
            while (val < d || (id & initial) == 0) { // id & initial == 1 << d for all ids at depth d, for < d it is 0 //@1 
                id <<= 1; //@2
                val = value(id); //@3
                if (val > d) { //@4
                    id ^= 1;
                    val = value(id);
                }
            }
            byte value = value(id);
            assert value == d && (id & initial) == 1 << d : String.format("val = %d, id & initial = %d, d = %d",
                    value, id & initial, d);
            setValue(id, unusable); // mark as unusable //@5
            updateParentsAlloc(id); //@6
            return id;
        }
    private byte value(int id) {
            return memoryMap[id];
    }

这里再重复一次PoolChunk里面维护一颗平衡二叉树来映射PoolChunk每个PoolSubpage的是否已经被分配的情况,平衡二叉树的深度为maxOrder,也就是平衡二叉数的所有叶子节点代表所有的PoolSubpage,而整棵二叉树从根到下,从左到右被映射为一个数组memoryMap,memoryMap长度为( 2的 (maxOrder=1)),memoryMap[0]处不存放任何元素,从 id=1到 2的(maxOrder-1)幂-1,举个更直观的例子,比如maxOrder为10,则memoryMap[]的长度为2048,memoryMap[1] 存放平衡二叉树的根节点,memory[2],memory[3]存放深度为1的节点,依次类推,memoryMap[2047]存放最后一个叶子节点。具体的算法是从根节点(dept=0)开始,一直向下找,直到到达层为d。找合适的节点。

代码@2:找到id的左子树(左节点)。

代码@3:获取该id所在二叉树的深度。

代码@4:val > d 表示已经被占用,需要找id的兄弟,也就是右节点,在平衡二叉树,已知左节点,求右节点的下标,使用id^1即可,然后看右节点是否被占用,如果没有,直接返回。理解这个算法,要知道,如果memoryMap[id]不可用,则不会继续去查找其子节点,也就是说,如果memoryMap[id]可用,就代表了至少有一个节点是没有被占用的。如果全被占用,id的值会为 2的(maxOrder+1)的幂,再减去1,如果按照上面的例子,id=2047。

代码@5:将memoryMap[id]=unusable,(maxOrder+1),代表PoolSubpage已经被占用。

代码@6:递归更新父节点的占用情况,这里有个哲学,memoryMap[id]中存的值,代表可以至少可以分配的深度数,比如memoryMap[8] = 2,则表示id为8的节点可以胜任分配深度>=2的内存大小,我们知道,深度越大,节点代表的内存就越小。比如id=3的节点,最小可以分配的深度为2,那他的子节点id为(6,7)能分配的最小深度为3,那比如id=6的节点已被分配,那它的直接父节点id=3,此时就不能再分配深度为2的内存了,因为已经被占用了一半,只能分配深度为3。updateParentsAlloc方法就是按照上述思路实现的,具体情况分析如下:

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

代码@1:循环调用,知道根节点。

代码@2:找到父节点。

代码@3,@4:得到parentId两个子节点当前可分配的深度值

代码@5:父节点可分配的深度值为两个子节点可分配的深度值的最小值。

1.3.1.2 PoolChunk的allocateSubpage代码@2 的 subpageIdx方法:

private int subpageIdx(int memoryMapIdx) {
             return memoryMapIdx ^ maxSubpageAllocs; // remove highest set bit, to get offset
             //我们一般简单的实现:
             //return memoryMapIdx - maxSubpageAllocs;
            //或 return memoryMapIdx & ( ~( (1 << maxOrder) -1) );
    }

1.3.1.3 PoolChunk的allocateSubpage代码@5 poolSubpage.allocate方法详解:

/**
         * Returns the bitmap index of the subpage allocation.
         */
        long allocate() {
            if (elemSize == 0) {
                return toHandle(0);
            }
            if (numAvail == 0 || !doNotDestroy) { //@1
                return -1;
            }
            final int bitmapIdx = getNextAvail(); //@2
            int q = bitmapIdx >>> 6; //@3
            int r = bitmapIdx & 63; //@4
            assert (bitmap[q] >>> r & 1) == 0;
            bitmap[q] |= 1L << r; //@5
            if (-- numAvail == 0) {
                removeFromPool(); //@6
            }
            return toHandle(bitmapIdx); //@7
        }

代码@1:如果可用内存块小于0,则直接返回-1。

代码@2:从bitmap中,获取下一个可用的内存区域,PoolSubpage由大小相等的内存区域组成,总共maxNumElems个,每个区域大小为elemSize。如果没有可用的区域,返回-1,否则返回0-(maxNumElms)之间的一个值。

代码@3:q表示在bitmap中的下标。

代码@4:r表示 bitmapInx % 64 的余数。

代码@5:就是要将即将返回的PoolSubpag所代表的位设置为1, 1L<<<r,然后与bimap[q] | 1L<<r即可实现这样的逻辑。

代码@6:如果该PoolSubpage所有内存都被分配后,从PoolSubpage链表中移除,这是在哪个链表中呢?是PoolChunk的链表?不是,因为PoolChunk只维护一个所有的PoolSubpage数组,是移除PoolArena的PoolSubpage链表,从这里也可以看出PoolArena维护的tinyPoolSubpages[]和smallPoolSubpages[]中的PoolSubpage都是未全部用完的PoolSubpage。

代码@7:返回一个long类型的变量,该变量的后32位保存中该PoolSubpage在PoolChunk的memoryMap中的下标id,高32位保存bitmapIdx。

关于PoolSubpage allocate 代码@2详解:getNextAvail方法:

private int getNextAvail() {
            int nextAvail = this.nextAvail;
            if (nextAvail >= 0) { //@1
                this.nextAvail = -1;
                return nextAvail;
            }
            return findNextAvail();
        }

如果是第一次分配,直接返回0,即可,因为下一个可用的就是0。

如果不是第一次分配,然后需要从bitmap中获取一个可用的PoolSubpage。进入findNextAvail()方法。

private int findNextAvail() {
            final long[] bitmap = this.bitmap;
            final int bitmapLength = this.bitmapLength;
            for (int i = 0; i < bitmapLength; i ++) { //@2
                long bits = bitmap[i];
                if (~bits != 0) { //@3
                    return findNextAvail0(i, bits); //@4
                }
            }
            return -1;
        }

代码@2:遍历bitmap,根据当前的elmSize可以得知当前用来多少个long来代表总的长度。

代码@3:如果取反不为0,则说明用这个long代表的64个PoolSubpage未使用完,可用从这里获取一个,故进入到代码@4。

private int findNextAvail0(int i, long bits) {
            final int maxNumElems = this.maxNumElems;
            final int baseVal = i << 6; //@5
            for (int j = 0; j < 64; j ++) { //遍历,一个long总代表64个PoolSubpage
                if ((bits & 1) == 0) { //@6
                    int val = baseVal | j; //@8
                    if (val < maxNumElems) {
                        return val;
                    } else {
                        break;
                    }
                }
                bits >>>= 1; //@7
            }
            return -1;
        }

代码@5:baseVal = i << 6,也就是i * 64,如果i在bitmap数组中的下标为0,baseVal则等于0,如果下标为1,则baseVal=64,如果下标为2,则baseVal=128。

代码@6:用bits&1判断是否等于0,也就是判断bits的低位是否被占用,如果为1,则表示被占用,如果为0,表示未被占用,如果结果为0,则可以返回该值了,如果不为0,则无符号向右移动1位,相当与去掉最后一位,继续比较。从这里就可以看出,一个long从低位开始被标记。我们举例说明一下,比如现在 用两个long类型可以标记所有的PoolSubpage,比如bitmap[0] = (00000000 00000000 00000000 00000000 00000000 00000000 00000000 0111111),表示已经分配了7个PoolSubpage。

代码 toHandle方法详解:

private long toHandle(int bitmapIdx) {
            return 0xb000000000000000L | (long) bitmapIdx << 32 | memoryMapIdx;
        }

该值,主要是用一个long类型的值的低32位来表示 memoryMapIdx,用高32位表示 bitmapIdx,这里为什么需要与0xb000000000000000L进行或运算呢?据我的目前掌握的知识看,主要是将0映射为一个数字,主要来区分是在PoolSubpage中分配的,还是内存大于pageSize,在内存释放的时候,可以通过bitmaIdx >> 32 & 0x3FFFFFFF 得出原来的索引。

1.3.1.4 关于 PoolArena allocate代码@4 initBuf详解:

void initBuf(PooledByteBuf<T> buf, long handle, int reqCapacity) {
            int memoryMapIdx = (int) handle;//@1
            int bitmapIdx = (int) (handle >>> Integer.SIZE); //@2
            if (bitmapIdx == 0) {//@3
                byte val = value(memoryMapIdx);
                assert val == unusable : String.valueOf(val);
                buf.init(this, handle, runOffset(memoryMapIdx), reqCapacity, runLength(memoryMapIdx));
            } else {//@4
                initBufWithSubpage(buf, handle, bitmapIdx, reqCapacity);
            }
        }

首先这里对入参 long handle做一个详解:如果需要分配的内存大于pageSize,则返回的高32为为0,低32位为memoryMap的下标id。具体参照下文的allocateRun方法讲解;如果需要分配的内存小于pageSize,则返回的高32为为bitmapIdx,低32位同样表示memoryMap的id。

代码@1:从long中获取memoryMapIdx。

代码@2:从long中获取bitmapIdx。

代码@3:如果bitmapIdx为0,偏移量就是PoolSubpage的偏移量。为什么呢?bitmapIdx为0在内存申请大于pageSize和小于pageSize时都会出现:

  • 如果申请的内存数小于pageSize,bitmapIdx表示该pageSubpage是第一次分配。
  • 如果申请的内存数大于pageSize,表明该节点所以子节点都是第一次被分片。故偏移量就是memory[id]所代表的偏移量。

代码@4:如果bitmapIdx不为0,需要计算偏移量,与PoolSubpage的elemSize相关,具体请看2)PoolChunk的initBufWithSubpage方法,该方法,最终还是要调用PooledByteBuf的init方法,分配内存,这里只是需要计算偏远量。

1、PooledByteBuf的 int方法

void init(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength) {
            assert handle >= 0;
            assert chunk != null;
            this.chunk = chunk;
            this.handle = handle;
            memory = chunk.memory;               // @1
            this.offset = offset;                           // @2
            this.length = length;                         // @3
            this.maxLength = maxLength;         // @4
            setIndex(0, 0);                                   //@5
            tmpNioBuf = null;
            initThread = Thread.currentThread();//@6
        }

代码@1:PooledByteBuf的内存,直接指向PoolChunk的memory;

代码@2:offset,在memory的起始偏移量。

代码@3:memory的offset + length之间的内存被该PooledByteBuf占用,其他ByteBuf无法使用。

代码@4:maxLength的作用是什么呢?目前我的理解是,PooledByteBuf自动扩容时,只要最终长度不超过maxLength,就可以在当前的内存中完成,不需要去申请新的空间,再进行内存负责。

代码@5:初始化readIndex,writeIndex。

代码@6:记录该PooledByteBuf的初始化线程,方便本地线程池的使用。

2、PoolChunk的iinitBufWithSubpage方法详解

private void initBufWithSubpage(PooledByteBuf<T> buf, long handle, int bitmapIdx, int reqCapacity) {
            assert bitmapIdx != 0;
            int memoryMapIdx = (int) handle;
            PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
            assert subpage.doNotDestroy;
            assert reqCapacity <= subpage.elemSize;
            buf.init(
                this, handle,
                runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize, reqCapacity, subpage.elemSize);
        }

1.3.2 PoolChunk allocate关于代码@allocateRun,超过pageSize内存分配源码分析

/**
         * Allocate a run of pages (>=1)
         *
         * @param normCapacity normalized capacity
         * @return index in memoryMap
         */
        private long allocateRun(int normCapacity) {
            int d = maxOrder - (log2(normCapacity) - pageShifts);    //@1
            int id = allocateNode(d);                                                   //@2
            if (id < 0) {
                return id;
            }
            freeBytes -= runLength(id);                                            //@3
            return id;
        }

该方法的实现,就是要在平衡二叉树中要找到一个可以满足normCapacity的节点,从前文的介绍中得知,memoryMap[id]存放的是,该id能分配的最小深度代表的容量。超过pageSize的内存节点,肯定不是在叶子节点。如果让我实现查找合适id的算法,我想应该是这样的:

1、算成需要分配的内存大小是 pageSize的倍数,再直观点就是normalCapacity 是 pageSize 的倍数n,然后算成n是2的多少次幂,比如pageSize=4,而normalCapacity是16,得出的倍数是4,4是2的2次幂,最终要得到这个值。标记为r

2、然后从平衡二叉树,沿着最底层(高度为0,深度为maxOrder),向上找r级,即能知道这一层次的节点可以容纳下normaCapacity的节点。

代码@1:先算出normCapacity,2的对(2的幂)然后减去pageSize(2的幂),得出思路1的r的值,然后用maxOrder-r,就表示深度最大为maxOrder-r。

代码@2:沿着平衡二叉树,从根节点开始,寻找一个合适的节点。如果无法分配,返回-1。

allocateNode(id)在上文中已经讲解,为了增深映像,在这里将代码再次粘贴:

private int allocateNode(int d) {
            int id = 1;
            int initial = - (1 << d); // has last d bits = 0 and rest all = 1
            byte val = value(id);
            if (val > d) { // unusable
                return -1;
            }
            while (val < d || (id & initial) == 0) { // id & initial == 1 << d for all ids at depth d, for < d it is 0
                id <<= 1;
                val = value(id);
                if (val > d) {
                    id ^= 1;
                    val = value(id);
                }
            }
            byte value = value(id);
            assert value == d && (id & initial) == 1 << d : String.format("val = %d, id & initial = %d, d = %d",
                    value, id & initial, d);
            setValue(id, unusable); // mark as unusable
            updateParentsAlloc(id);
            return id;
        }

1.3.3 关于PoolArena 2.2.2 allocate方法的代码@5allocateHuge方法详解

在netty内存管理中,如果超过chunkSize的内存,为大内存,不重复使用。

private void allocateHuge(PooledByteBuf<T> buf, int reqCapacity) {
            buf.initUnpooled(newUnpooledChunk(reqCapacity), reqCapacity);
    }
    PoolArena.HeapArena:
    protected PoolChunk<byte[]> newUnpooledChunk(int capacity) {
            return new PoolChunk<byte[]>(this, new byte[capacity], capacity);
    }
    PoolChunk非池管理的PoolChunk:
    /** Creates a special chunk that is not pooled. */
        PoolChunk(PoolArena<T> arena, T memory, int size) {
            unpooled = true;
            this.arena = arena;
            this.memory = memory;
            memoryMap = null;
            depthMap = null;
            subpages = null;
            subpageOverflowMask = 0;
            pageSize = 0;
            pageShifts = 0;
            maxOrder = 0;
            unusable = (byte) (maxOrder + 1);
            chunkSize = size;
            log2ChunkSize = log2(chunkSize);
            maxSubpageAllocs = 0;
        }
    大内存的PoolChunk,不缓存使用,故内部不会再细化为PoolSubpage等数据结构。
    void initUnpooled(PoolChunk<T> chunk, int length) {
            assert chunk != null;

            this.chunk = chunk;
            handle = 0;
            memory = chunk.memory;
            offset = 0;
            this.length = maxLength = length;
            setIndex(0, 0);
            tmpNioBuf = null;
            initThread = Thread.currentThread();
        }

PooledByteBuf来初始化是,lengt,maxLenth等于需要申请的内存。看过tiny,small内存的分配后,大内存的分配比较简单,就不做每行代码的解读。不容易呀,Netty内存的分配就讲解完毕了。接下来注重分析两个方面:内存释放与PooledByteBuf动态扩容。

2、源码分析Netty内存释放

在讲解Netty内存释放之前,我们还是简单的再回顾一下ReferenceCounted,Netty的ByteBuf继承该接口,这也表明,Netty内部的内存管理是基于引用计数来进行内存的回收的。具体的实现是AbstractReferenceCounted。所以,我们在使用PooledByteBuf时,用完后要记得调用release方法。在线程本地分配时会再次强调,如果PooledByteBuf在用完后,没有调用realse方法,是无法被线程共用的。内存的释放入口,是ByteBuf的relase方法,也就是AbstractReferenceCountedByteBuf的release方法:

public final boolean release(int decrement) {
            if (decrement <= 0) {
                throw new IllegalArgumentException("decrement: " + decrement + " (expected: > 0)");
            }

            for (;;) {
                int refCnt = this.refCnt;
                if (refCnt < decrement) {
                    throw new IllegalReferenceCountException(refCnt, -decrement);
                }

                if (refCntUpdater.compareAndSet(this, refCnt, refCnt - decrement)) {
                    if (refCnt == decrement) {
                        deallocate();
                        return true;
                    }
                    return false;
                }
            }
        }

该方法的方式是循环利用CAS将引用减少,直到引用为0,则调用释放方法,ByteBuf的内存释放,不同子类,不同的实现,故deallocate方法为抽象方法;

/**
         * Called once {@link #refCnt()} is equals 0.
         */
        protected abstract void deallocate();
    3.1 PooledByteBuf deallocate方法详解
    protected final void deallocate() {
            if (handle >= 0) {
                final long handle = this.handle;
                this.handle = -1;    
                memory = null;
                boolean sameThread = initThread == Thread.currentThread(); 
                initThread = null;
                chunk.arena.free(chunk, handle, maxLength, sameThread);         //@1
                recycle();                                                                                        //@2
            }
        }

内存释放的基本实现:首先将PooledByteBuf内部的handle修改为-1,将memory设置为null,,然后释放占用的内存,然后将该PooledByteBuf放入对象池。从这里也可以看出,PooledByteBuf ,不仅ByteBuf内部持有的缓存区会被回收,连PooledByteBuf这个对象本身,也会放入到对象池,供重复利用(线程级)。

2.1关于PooledByteBuf deallocate 代码@1 PoolArena的free方法,内存释放

/**
      *  @param    chunk
      *  @param    handle,内存分配,保存着bitmapIdx,memoryMapIdx      
      *  @param    normaCapacity   申请的内存(释放的内存,取值为PooledByteBuf的 maxLenth)
      *  @param    释放该PooledByteBuf的线程释放时创建该PooledByteBuf的线程
      */
    void free(PoolChunk<T> chunk, long handle, int normCapacity, boolean sameThreads) {
            if (chunk.unpooled) {          //@1
                destroyChunk(chunk);
            } else {                                 
                if (sameThreads) {     //@2
                    PoolThreadCache cache = parent.threadCache.get();
                    if (cache.add(this, chunk, handle, normCapacity)) {
                        // cached so not free it.
                        return;
                    }
                }

                synchronized (this) {     //@3
                    chunk.parent.free(chunk, handle);
                }
            }
        }

如果是非池的内存,也就是大内存,直接释放,如果释放的线程是创建该PooledByteBuf的线程,则放入线程本地变量中,重复利用,此时不需要释放。如果不是,则释放内存。

代码@1:我们看一下PoolArena.HeapArena和DirectArena的分别实现:

PoolArena.HeapArena
    protected void destroyChunk(PoolChunk<byte[]> chunk) {
                // Rely on GC.
    }
    PoolArena.DirectArena:
    protected void destroyChunk(PoolChunk<ByteBuffer> chunk) {
                PlatformDependent.freeDirectBuffer(chunk.memory);   //释放堆外内存
    }

代码@2:放入本地线程缓存中,这个稍后重点关注。

代码@3:释放内存,这里的内存释放与第一步的内存释放不同,第一步的内存释放,是直接将内存还给JVM堆、或操作系统。而这里的是在PoolChunk中进行标记与释放而已。chunk.parent指的是该PoolChunk的PoolChunkList对象。先重点进入到PoolChunk的free方法。

2.2 关于PoolArena.free方法 代码@3的讲解,此处主要是调用PoolChunk方法进行内存的释放:

/**
         * Free a subpage or a run of pages
         * When a subpage is freed from PoolSubpage, it might be added back to subpage pool of the owning PoolArena
         * If the subpage pool in PoolArena has at least one other PoolSubpage of given elemSize, we can
         * completely free the owning Page so it is available for subsequent allocations
         *
         * @param handle handle to free
         */
        void free(long handle) {
            int memoryMapIdx = (int) handle;                        //@1
            int bitmapIdx = (int) (handle >>> Integer.SIZE);  //@2

            if (bitmapIdx != 0) { // free a subpage    //@3
                PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
                assert subpage != null && subpage.doNotDestroy;
                if (subpage.free(bitmapIdx & 0x3FFFFFFF)) {
                    return;
                }
            }
            freeBytes += runLength(memoryMapIdx);     //@4
            setValue(memoryMapIdx, depth(memoryMapIdx));  //@5
            updateParentsFree(memoryMapIdx);                        //@6
        }

代码@1:获取在memoryMap的下标。

代码@2:获取bitmapIdx。

代码@3:如果bitmapIdx不等于0,表示该内存是从PoolSubpage中直接分配的(内存申请时,小于pageSize)。此时内存的释放,直接调用PoolSubpage的free方法。

代码@4,@5,@6:如果申请的内存超过了pageSize,首先将freeBytes增加,然后在代码@5,将memoryMapIdx处的值更新为原先depth的值,代表可以重新分片深度为depth[id]的内存了,级联更新memoryMap,的父节点的释放状态。

2.3 关于 PoolArena.free方法,代码@3, PoolSubpage.free方法详解。

/**
         * @return {@code true} if this subpage is in use.
         *         {@code false} if this subpage is not used by its chunk and thus it's OK to be released.
         */
        boolean free(int bitmapIdx) {
            if (elemSize == 0) {
                return true;
            }

            int q = bitmapIdx >>> 6;                         //@1
            int r = bitmapIdx & 63;                            //@2
            assert (bitmap[q] >>> r & 1) != 0;
            bitmap[q] ^= 1L << r;                             //@3

            setNextAvail(bitmapIdx);                         //@4

            if (numAvail ++ == 0) {                          //@5 start
                addToPool();
                return true;
            }

            if (numAvail != maxNumElems) {  
                return true;
            } else {
                // Subpage not in use (numAvail == maxNumElems)
                if (prev == next) {
                    // Do not remove if this subpage is the only one left in the pool.
                    return true;
                }

                // Remove this subpage from the pool if there are other subpages left in the pool.
                doNotDestroy = false;
                removeFromPool();
                return false;
            }   //@5 end
        }

代码@1:求出bitmapIdx在bitmap数组中的下标。

代码@3:就是bitmapIdx在bitmap[q]下所对应的高位1,设置为0,表示可以再次被用来分配。

代码@4:将bitmapIdx设置为下一个可以用nextAvail,这时在nextAvail>0,分配时候就会直接先用释放的,保证内存的连续性。

代码@5:如果PoolSubpage有剩余空间后,加入到PoolArena的tinyPoolSubpages[]或smallPoolSubpages[]中。如果空间全部分配后,或一个都没分配,从PoolArena的tinyPoolSubpages[],smallPoolSubpages[]中移除。

3.3 关于PoolArena.free方法,代码@6,级联更新释放状态详解

/**
         * Update method used by free
         * This needs to handle the special case when both children are completely free
         * in which case parent be directly allocated on request of size = child-size * 2
         *
         * @param id id
         */
        private void updateParentsFree(int id) {
            int logChild = depth(id) + 1;
            while (id > 1) {
                int parentId = id >>> 1;
                byte val1 = value(id);
                byte val2 = value(id ^ 1);
                logChild -= 1; // in first iteration equals log, subsequently reduce 1 from logChild as we traverse up

                if (val1 == logChild && val2 == logChild) {
                    setValue(parentId, (byte) (logChild - 1));
                } else {
                    byte val = val1 < val2 ? val1 : val2;
                    setValue(parentId, val);
                }

                id = parentId;
            }
        }

这个算法在理解了上文的updateParentsAllocator()方法后,不难理解,只要我们明确一点,memoryMap[id]=order,表示id代表的节点,可以胜任大于等于order深度的内存分配需要。

3、PooledByteBuf内存扩容

在讲解这个问题的时候,不知道大家有没有注意到PooledByteBuf的maxLength属性?该属性有和作用,请看下文分解。

PooledByteBuf扩容算法,请看capacity(int newCapacity);

@Override
        public final ByteBuf capacity(int newCapacity) {
            ensureAccessible();

            // If the request capacity does not require reallocation, just update the length of the memory.
            if (chunk.unpooled) {                         //@1
                if (newCapacity == length) {
                    return this;
                }
            } else {
                if (newCapacity > length) {                      //@2
                    if (newCapacity <= maxLength) {
                        length = newCapacity;
                        return this;
                    }
                } else if (newCapacity < length) {              //@3
                    if (newCapacity > maxLength >>> 1) {
                        if (maxLength <= 512) {
                            if (newCapacity > maxLength - 16) {
                                length = newCapacity;
                                setIndex(Math.min(readerIndex(), newCapacity), Math.min(writerIndex(), newCapacity));
                                return this;
                            }
                        } else { // > 512 (i.e. >= 1024)
                            length = newCapacity;
                            setIndex(Math.min(readerIndex(), newCapacity), Math.min(writerIndex(), newCapacity));
                            return this;
                        }
                    }
                } else {
                    return this;
                }
            }

            // Reallocation required.
            chunk.arena.reallocate(this, newCapacity, true);      //@4
            return this;
        }

代码@1:如果是非池化的,并且新的容量等于length的话,可以直接返回,否则需要重新去申请内存。

代码@2:如果需要的内存大于length,此时newCapacity小于等于maxLength,则不需要扩容,如果需要的大小超过maxLength,则需要重新去申请内存。那这里的maxLength是什么呢?其实,如果PooledByteBuf的内存是在PoolSubpage中分配,那maxLength为PooledSubpage的内存区域中的总容量,PoolSubpage的MemoryRegion的总大小。也就是在是同一个时刻,一个PoolSubpage的MemoryRegion只能被一个PooledByteBuf所占用。

代码@3:如果需要申请的内存小于length,为了避免下一次需要扩容,再进一步判断,如果是tiny内存的话,判断newCapaciy与 maxLength-16的关系,如果不大于的话,本次也进行重新分配内存,然后维护PooledByteBuf的length,readerIndex,writerIndex。

代码@4:内存的重新分配 reallocate方法;

void reallocate(PooledByteBuf<T> buf, int newCapacity, boolean freeOldMemory) {
            if (newCapacity < 0 || newCapacity > buf.maxCapacity()) {
                throw new IllegalArgumentException("newCapacity: " + newCapacity);
            }

            int oldCapacity = buf.length;
            if (oldCapacity == newCapacity) {
                return;
            }

            PoolChunk<T> oldChunk = buf.chunk;
            long oldHandle = buf.handle;
            T oldMemory = buf.memory;
            int oldOffset = buf.offset;
            int oldMaxLength = buf.maxLength;
            int readerIndex = buf.readerIndex();
            int writerIndex = buf.writerIndex();

            allocate(parent.threadCache.get(), buf, newCapacity);    //@1
            if (newCapacity > oldCapacity) {    //@2 start 
                memoryCopy(
                        oldMemory, oldOffset,
                        buf.memory, buf.offset, oldCapacity);
            } else if (newCapacity < oldCapacity) {
                if (readerIndex < newCapacity) {
                    if (writerIndex > newCapacity) {
                        writerIndex = newCapacity;
                    }
                    memoryCopy(
                            oldMemory, oldOffset + readerIndex,
                            buf.memory, buf.offset + readerIndex, writerIndex - readerIndex);
                } else {
                    readerIndex = writerIndex = newCapacity;
                }
            } // @2end

            buf.setIndex(readerIndex, writerIndex);

            if (freeOldMemory) {
                free(oldChunk, oldHandle, oldMaxLength, buf.initThread == Thread.currentThread()); //@3
            }
        }

内存重新分配的算法如下:

首先重新申请内存,然后进行内存复制,最后释放原先的内存。

PoolArena.HeapArena的memoryCopy实现代码:

protected void memoryCopy(byte[] src, int srcOffset, byte[] dst, int dstOffset, int length) {
                if (length == 0) {
                    return;
                }

                System.arraycopy(src, srcOffset, dst, dstOffset, length);
            }

总结

Netty内存的组织,主要由PoolArena、PoolChunk、PoolSubpage,当然再加上线程本地分配(下个专题单独分析),在这里真正持有内存的单元是PoolChunk,俗称块内存,PoolChunk持有2的maxOrder次幂的PoolSubpage,PoolSubpage维护一个相对于PoolChunk的内存偏移量,每个PoolSubpage又由大小相等的内存区域构成,命名为MomeryRegion,为了方便快速的位运算,内存的大小,pageSize等大小,都是2的幂。

Netty的另一层次的代码组织,我把它称为运行时内存组织也是挺富有想象力的,PoolArena按照PoolChunk的内部的使用率区间,维护qint、q000、q025、q050、q075、q100的PoolChunk链表,并且qint,q000也是一个链表,这样动态跟踪了PoolChunk的使用,通样的道理,PoolArena内部通用按照PoolSubpage,按照MeomryRegion的大小,又维护一个一个的队列,PoolSubpage[] tinySubpagePools[],smallSubpagePools[]。

介绍这些数据存储后,重点深入源码,分析了内存的分配,释放,PooledByteBuf的动态扩容。下一个研究专题,Netty本地线程分配与PooledByteBuf本地对象池机制。


来源:https://blog.csdn.net/prestigeding/article/details/53977445

点赞(0)
版权归原创作者所有,任何形式转载请联系作者; Java 技术驿站 >> Netty学习之旅----源码分析内存分配与释放原理

相关推荐