上一篇文章分享了PoolArena如何通过PoolChunk,PoolSubpage治理内存。
本文则分享PoolChunk,PoolSubpage中如何分配内存。
源码剖析基于Netty 4.1

首先阐明PoolChunk内存组织形式。
PoolChunk的内存大小默认是16M,它将内存组织成为一颗完满二叉树。
二叉树的每一层每个节点所代表的内存大小都是均等的,并且每一层节点所代表的内存大小总和加起来都是16M。
每一层节点可分配内存是父节点的1/2。整颗二叉树的总层数为12,层数从0开始。

示意图如下

(下标 -> 层数,最大可调配块大小)

先看一下PoolChunk的构造函数

PoolChunk(PoolArena<T> arena, T memory, int pageSize, int maxOrder, int pageShifts, int chunkSize, int offset) {    unpooled = false;    this.arena = arena;    this.memory = memory;    this.pageSize = pageSize;    this.pageShifts = pageShifts;    this.maxOrder = maxOrder;    this.chunkSize = chunkSize;    this.offset = offset;    unusable = (byte) (maxOrder + 1);    log2ChunkSize = log2(chunkSize);    subpageOverflowMask = ~(pageSize - 1);    freeBytes = chunkSize;    assert maxOrder < 30 : "maxOrder should be < 30, but is: " + maxOrder;    maxSubpageAllocs = 1 << maxOrder;    // Generate the memory map.    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 ++;        }    }    subpages = newSubpageArray(maxSubpageAllocs);}

unpooled: 是否应用内存池
arena:该PoolChunk所属的PoolArena
memory:底层的内存块,对于堆内存,它是一个byte数组,对于间接内存,它是(jvm)ByteBuffer,但无论是哪种模式,其内存大小默认都是16M。
pageSize:叶子节点大小,默认为8192,即8K。
maxOrder:示意二叉树最大的层数,从0开始。默认为11。
chunkSize:整个PoolChunk的内存大小,默认为16777216,即16M。
offset:为了内存对齐而应用的偏移数量,默认为0。
unusable:示意节点已被调配,不必了,默认为12。
freeBytes:闲暇内存字节数。
每个PoolChunk都要按内存使用率关联到一个PoolChunkList上,内存使用率正是通过freeBytes计算。
maxSubpageAllocs:叶子节点数量,默认为2048,即2^11。

log2ChunkSize:用于计算偏移量,默认为24。
subpageOverflowMask:用于判断申请内存是否为PoolSubpage,默认为-8192。
pageShifts:用于计算分配内存所在二叉树层数,默认为13。

memoryMap:初始化内存治理二叉树,将每一层节点值设置为层数d。
应用数组保护二叉树,第d层的开始下标为 1<<d。(数组第0个元素不应用)。
depthMap:保留二叉树的层数,用于通过地位下标找到其在整棵树中对应的层数。
留神:depthMap的值代表二叉树的层数,初始化后不再变动。
memoryMap的值代表以后节点最大可申请内存块,在分配内存过程中一直变动。
节点最大可申请内存块能够通过层数d计算,即2 ^ (pageShifts + maxOrder - d)

PoolChunk#allocate

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

若申请内存大于pageSize,调用allocateRun办法调配Chunk级别的内存。
否则调用allocateSubpage办法调配PoolSubpage,再在PoolSubpage上调配所需内存。

PoolChunk#allocateRun

private long allocateRun(int normCapacity) {    // #1    int d = maxOrder - (log2(normCapacity) - pageShifts);    // #2    int id = allocateNode(d);    if (id < 0) {        return id;    }    // #2    freeBytes -= runLength(id);    return id;}

#1 计算应该在哪层调配分配内存
maxOrder - (log2(normCapacity) - pageShifts),如16K, 即2^14,计算结果为10,即在10层调配。
#2 缩小闲暇内存字节数。

PoolChunk#allocateNode,在d层调配一个节点

private int allocateNode(int d) {    int id = 1;    int initial = - (1 << d); // has last d bits = 0 and rest all = 1    // #1    byte val = value(id);    if (val > d) { // unusable        return -1;    }    // #2    while (val < d || (id & initial) == 0) { // id & initial == 1 << d for all ids at depth d, for < d it is 0        // #3        id <<= 1;        val = value(id);        // #4        if (val > d) {            // #5            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);    // #6    setValue(id, unusable); // mark as unusable    // #7    updateParentsAlloc(id);    return id;}

#1 memoryMap[1] > d,第0层的可分配内存有余,表明该PoolChunk内存不能满足调配,调配失败。
#2 遍历二叉树,找到满足内存调配的节点。
val < d,即该节点内存满足调配。
id & initial = 0,即 id < 1<<d, d层之前循环继续执行。这里并不会呈现val > d的场景,但会呈现val == d的场景,如
PoolChunk以后可分配内存为2M,即memoryMap[1] = 3,这时申请2M内存,在0-2层,都是val == d。可参考前面的实例。
#3 向下找到下一层下标,留神,子树左节点的下标是父节点下标的2倍。
#4 val > d,示意以后节点不能满足调配
#5 id ^= 1,查找同一父节点下的兄弟节点,在兄弟节点上分配内存。
id ^= 1,当id为偶数,即为id+=1, 当id为奇数,即为id-=1
因为后面通过id <<= 1找到下一层下标都是偶数,这里等于id+=1。
#6
因为一开始判断了PoolChunk内存是否足以调配,所以这里肯定能够找到一个可调配节点。
这里标注找到的节点已调配。
#7 更新找到节点的父节点最大可分配内存块大小

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

#1 向父节点遍历,直到根节点
#2 id >>> 1,找到父节点
取以后节点和兄弟节点中较小值,作为父节点的值,示意父节点最大可分配内存块大小。

如memoryMap[1] = 0,示意最大可分配内存块为16M。
调配8M后,memoryMap[1] = 1,示意以后最大可分配内存块为8M。

上面看一则实例,大家能够联合实例了解下面的代码

上面看一下如何调配PoolSubpage。
PoolChunk#PoolSubpage

private long allocateSubpage(int normCapacity) {    // #1    PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity);    synchronized (head) {        // #2        int d = maxOrder; // subpages are only be allocated from pages i.e., leaves        int id = allocateNode(d);        if (id < 0) {            return id;        }        // #3        final PoolSubpage<T>[] subpages = this.subpages;        final int pageSize = this.pageSize;        freeBytes -= pageSize;        int subpageIdx = subpageIdx(id);        PoolSubpage<T> subpage = subpages[subpageIdx];        // #4        if (subpage == null) {            subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity);            subpages[subpageIdx] = subpage;        } else {            subpage.init(head, normCapacity);        }        return subpage.allocate();    }}

#1 新结构的PoolSubpage,须要搁置到PoolArena#tinySubpagePools或smallSubpagePools数组中。
这里找到PoolArena中tinySubpagePools或smallSubpagePools数组中对应链表的head节点。
能够看到这里应用了head节点执行同步操作,这是有必要的,否则可能多个线程同时批改链表。
#2 调配一个叶子节点,PoolSubpage实际上就是PoolChunk一个叶子节点。
#3 在subpages数组中找到对应的PoolSubpage。
因为id是整个二叉树(memoryMap)的节点下标,而subpages只是叶子节点数组,所以要去掉高位,默认取低10个bit位的值(小于2048)。
#4 初始化PoolSubpage。

PoolSubpage

上面看看PoolSubpage如何治理Tiny,Small级别的内存块。
先阐明PoolSubpage如何组织内存。
PoolSubpage实际上就是PoolChunk中的一个Page节点,默认大小为8192。
它中将内存划分为若干内存块,每个内存块大小是雷同的,并应用位图的形式治理这些内存块。
PoolSubpage#bitmap是一个long数组,其中每个long元素上每个bit位都能够代表一个内存块是否应用。

PoolSubpage

PoolSubpage(PoolSubpage<T> head, PoolChunk<T> chunk, int memoryMapIdx, int runOffset, int pageSize, int elemSize) {    this.chunk = chunk;    this.memoryMapIdx = memoryMapIdx;    this.runOffset = runOffset;    this.pageSize = pageSize;    bitmap = new long[pageSize >>> 10]; // pageSize / 16 / 64    init(head, elemSize);}void init(PoolSubpage<T> head, int elemSize) {    doNotDestroy = true;    this.elemSize = elemSize;    if (elemSize != 0) {        // #1        maxNumElems = numAvail = pageSize / elemSize;        nextAvail = 0;        bitmapLength = maxNumElems >>> 6;        if ((maxNumElems & 63) != 0) {            bitmapLength ++;        }        for (int i = 0; i < bitmapLength; i ++) {            bitmap[i] = 0;        }    }    addToPool(head);}

chunk:该PoolSubpage是哪儿PoolChunk的Page节点
memoryMapIdx:对应Page节点在PoolChunk#memoryMap的索引
runOffset:对应Page节点在PoolChunk#memory的偏移量
pageSize:PoolSubpage整体内存大小,默认为
bitmap:内存治理位图数组,数组长度默认为pageSize >>> 10,即pageSize / 16 / 64,
因为每个内存块最小为16字节,每个long有64位,所以这个长度足够了。
elemSize:每个内存块的大小
maxNumElems:内存块总数量。
bitmapLength:bitmap应用的元素个数,应用bitmap中一部分元素足以治理全副内存块。
maxNumElems >>> 6,即 maxNumElems / 64
(maxNumElems & 63) != 0,代表maxNumElems不能整除64,所以bitmapLength要加1,用于治理余下的内存块。

每个PoolSubpage都要加到PoolArena#tinySubpagePools或smallSubpagePools数组中,这个两个数组元素都是链表,由内存块大小雷同的PoolSubpage组成。
addToPool办法正是将PoolSubpage增加到PoolArena中,从这个办法能够看到,结构的PoolSubpage会插入到head前面,并且这些链表会组成环,即最初一个节点的next指向head。

PoolSubpage#allocate

long allocate() {    if (elemSize == 0) {        return toHandle(0);    }    // #1    if (numAvail == 0 || !doNotDestroy) {        return -1;    }    // #2    final int bitmapIdx = getNextAvail();    // #3    int q = bitmapIdx >>> 6;    int r = bitmapIdx & 63;    assert (bitmap[q] >>> r & 1) == 0;    bitmap[q] |= 1L << r;    // #4    if (-- numAvail == 0) {        removeFromPool();    }    // #5    return toHandle(bitmapIdx);}

#1 没有可用内存块,调配失败
#2 获取下一个可用内存块的bit下标
#3 设置对应bit为1,即已应用
bitmapIdx >>> 6,获取该内存块在bitmap数组中第q元素
bitmapIdx & 63,获取该内存块是bitmap数组中第q个元素的第r个bit位
bitmap[q] |= 1L << r 将bitmap数组中第q个元素的第r个bit位设置为1,示意曾经应用
#4 所有内存块已调配了,则将其从PoolArena中移除。
#5 toHandle 转换为最终的标记值

PoolSubpage#getNextAvail

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

nextAvail为初始值或free时开释的值。
如果nextAvail存在,设置为不可用后间接返回该值。
如果不存在,调用findNextAvail查找下一个可用内存块。

private int findNextAvail() {    final long[] bitmap = this.bitmap;    final int bitmapLength = this.bitmapLength;    // #1    for (int i = 0; i < bitmapLength; i ++) {        long bits = bitmap[i];        if (~bits != 0) {            return findNextAvail0(i, bits);        }    }    return -1;}private int findNextAvail0(int i, long bits) {    final int maxNumElems = this.maxNumElems;    final int baseVal = i << 6;    // #2    for (int j = 0; j < 64; j ++) {        if ((bits & 1) == 0) {            int val = baseVal | j;            if (val < maxNumElems) {                return val;            } else {                break;            }        }        bits >>>= 1;    }    return -1;}

#1 遍历bitmap,~bits != 0,示意存在一个bit位不为1,即存在可用内存块。
#2 遍历64个bit位,
(bits & 1) == 0,查看最低bit位是否为0(可用),为0则返回val。
val等于 (i << 6) | j,即i * 64 + j,该bit位在bitmap中是第几个bit位。
bits >>>= 1,右移一位,解决下一个bit位。

最初解析一下PoolSubpage#toHandle办法

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

低32个bit位存储memoryMapIdx,即对应Page节点在PoolChunk#memoryMap的索引。
高33 ~ 61个bit存储bitmapIdx,即调配了PoolSubpage第几个内存块。
而0x4000000000000000L,即2^62。这里设置第62个bit位为1,代表内存块调配在PoolSubpage中。因为bitmapIdx可能为0,仅应用高32位示意bitmapIdx,无奈辨别bitmapIdx为0和没有调配在PoolSubpage的场景,所以置62位为1做标记。

内存开释

PoolChunk#free

void free(long handle) {    // #1    int memoryMapIdx = memoryMapIdx(handle);    int bitmapIdx = bitmapIdx(handle);    // #2    if (bitmapIdx != 0) { // free a subpage        PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];        assert subpage != null && subpage.doNotDestroy;        PoolSubpage<T> head = arena.findSubpagePoolHead(subpage.elemSize);        synchronized (head) {            if (subpage.free(head, bitmapIdx & 0x3FFFFFFF)) {                return;            }        }    }    freeBytes += runLength(memoryMapIdx);    setValue(memoryMapIdx, depth(memoryMapIdx));    updateParentsFree(memoryMapIdx);}

#1 获取memoryMapIdx和bitmapIdx
#2 bitmapIdx不为0,即调配在PoolSubpage。
找到对应的PoolSubpage,调用PoolSubpage#free是否内存。
bitmapIdx & 0x3FFFFFFF,后面说了,高33 ~ 61个bit才是真正的bitmapIdx,这里取出该值。
PoolSubpage#free办法返回false,即代表该PoolSubpage所有内存块已开释,能够开释对应Page节点。
#3 解决到这里,就是开释Chunk级别的内存块了。
减少闲暇内存字节数。
设置二叉树中对应的节点为未调配
对应批改该节点的父节点。

PoolSubpage#free

boolean free(PoolSubpage<T> head, int bitmapIdx) {    if (elemSize == 0) {        return true;    }    // #1    int q = bitmapIdx >>> 6;    int r = bitmapIdx & 63;    assert (bitmap[q] >>> r & 1) != 0;    bitmap[q] ^= 1L << r;    setNextAvail(bitmapIdx);    // #2    if (numAvail ++ == 0) {        addToPool(head);        return true;    }    // #3    if (numAvail != maxNumElems) {        return true;    } else {        // #4        if (prev == next) {            // Do not remove if this subpage is the only one left in the pool.            return true;        }        // #5        doNotDestroy = false;        removeFromPool();        return false;    }}

#1 将对应bit位设置为能够
#2 在PoolSubpage已齐全应用时开释了内存,这时重新加入到PoolArena中。
#3 未齐全开释,即还存在已分配内存块,返回true
#4 解决到这里,是所有内存块曾经齐全开释的场景
prev == next,只有一种场景,就是prev,next都指向head,该PoolSubpage是PoolArena中该链表最初一个PoolSubpage,这时不开释该PoolSubpage,以便下次申请内存时间接从该PoolSubpage上调配。
#5 从PoolArena中移除,并返回false,这时PoolChunk会将开释对应Page节点。

如果您感觉本文不错,欢送关注我的微信公众号,您的关注是我保持的能源!