Netty源码解析 -- PoolSubpage实现原理

前面文章说了PoolChunk如何管理Normal内存块，本文分享PoolSubpage如何管理Small内存块。
源码分析基于Netty 4.1.52

内存管理算法

PoolSubpage负责管理Small内存块。一个PoolSubpage中的内存块size都相同，该size对应SizeClasses#sizeClasses表格的一个索引index。
新建立的PoolSubpage都必须加入到PoolArena#smallSubpagePools[index]链表中。
PoolArena#smallSubpagePools是一个PoolSubpage数组，数组中每一个元素都是一个PoolSubpage链表，PoolSubpage之间能够经过next，prev组成链表。
感兴趣的同窗能够参考《内存对齐类SizeClasses》。

注意，Small内存size并不必定小于pageSize(默认为8K)
默认Small内存size <= 28672(28KB)
关于Normal内存块，Small内存块，pageSize，可参考《PoolChunk实现原理》。

PoolSubpage实际上就是PoolChunk中的一个Normal内存块，大小为其管理的内存块size与pageSize最小公倍数。
PoolSubpage使用位图的方式管理内存块。
PoolSubpage#bitmap是一个long数组，其中每一个long元素上每一个bit位均可以表明一个内存块是否使用。

内存分配

分配Small内存块有两个步骤

1. PoolChunk中分配PoolSubpage。

若是PoolArena#smallSubpagePools中已经有对应的PoolSubpage缓冲，则不须要该步骤。

1. PoolSubpage上分配内存块

PoolChunk#allocateSubpage

private long allocateSubpage(int sizeIdx) {
    // #1
    PoolSubpage<T> head = arena.findSubpagePoolHead(sizeIdx);
    synchronized (head) {
        //allocate a new run
        // #2
        int runSize = calculateRunSize(sizeIdx);
        //runSize must be multiples of pageSize
        // #3
        long runHandle = allocateRun(runSize);
        if (runHandle < 0) {
            return -1;
        }
        // #4
        int runOffset = runOffset(runHandle);
        int elemSize = arena.sizeIdx2size(sizeIdx);

        PoolSubpage<T> subpage = new PoolSubpage<T>(head, this, pageShifts, runOffset,
                           runSize(pageShifts, runHandle), elemSize);

        subpages[runOffset] = subpage;
        // #5
        return subpage.allocate();
    }
}

#1 这里涉及修改PoolArena#smallSubpagePools中的PoolSubpage链表，须要同步操做
#2 计算内存块size和pageSize最小公倍数
#3 分配一个Normal内存块，做为PoolSubpage的底层内存块，大小为Small内存块size和pageSize最小公倍数
#4 构建PoolSubpage
runOffset，即Normal内存块偏移量，也是该PoolSubpage在整个Chunk中的偏移量
elemSize，Small内存块size
#5 在subpage上分配内存块

PoolSubpage(PoolSubpage<T> head, PoolChunk<T> chunk, int pageShifts, int runOffset, int runSize, int elemSize) {
    // #1
    this.chunk = chunk;
    this.pageShifts = pageShifts;
    this.runOffset = runOffset;
    this.runSize = runSize;
    this.elemSize = elemSize;
    bitmap = new long[runSize >>> 6 + LOG2_QUANTUM]; // runSize / 64 / QUANTUM
    init(head, elemSize);
}

void init(PoolSubpage<T> head, int elemSize) {
    doNotDestroy = true;
    if (elemSize != 0) {
        // #2
        maxNumElems = numAvail = runSize / elemSize;
        nextAvail = 0;
        bitmapLength = maxNumElems >>> 6;
        if ((maxNumElems & 63) != 0) {
            bitmapLength ++;
        }

        for (int i = 0; i < bitmapLength; i ++) {
            bitmap[i] = 0;
        }
    }
    // #3
    addToPool(head);
}

#1 bitmap长度为runSize / 64 / QUANTUM，从《内存对齐类SizeClasses》能够看到，runSize都是2^LOG2_QUANTUM的倍数。

#2
elemSize：每一个内存块的大小
maxNumElems：内存块数量
bitmapLength：bitmap使用的long元素个数，使用bitmap中一部分元素足以管理所有内存块。
(maxNumElems & 63) != 0，表明maxNumElems不能整除64，因此bitmapLength要加1，用于管理余下的内存块。
#3 添加到PoolSubpage链表中

前面分析《Netty内存池与PoolArena》中说过，在PoolArena中分配Small内存块时，首先会从PoolArena#smallSubpagePools中查找对应的PoolSubpage​。若是找到了，直接从该PoolSubpage​上分配内存。不然，分配一个Normal内存块，建立PoolSubpage​，再在上面分配内存块。

PoolSubpage#allocate

long allocate() {
    // #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 没有可用内存块，分配失败。一般PoolSubpage分配完成后会从PoolArena#smallSubpagePools中移除，再也不在该PoolSubpage上分配内存，因此通常不会出现这种场景。
#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 转换为最终的handle

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位。

内存释放

释放Small内存块可能有两个步骤

1. 释放PoolSubpage的上内存块
2. 若是PoolSubpage中的内存块已所有释放，则从Chunk中释放该PoolSubpage，同时从PoolArena#smallSubpagePools移除它。

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 逻辑到这里，是处理全部内存块已经彻底释放的场景。
PoolArena#smallSubpagePools链表组成双向链表，链表中只有head和当前PoolSubpage时，当前PoolSubpage的prev，next都指向head。
这时当前​PoolSubpage是PoolArena中该链表最后一个PoolSubpage，不释放该PoolSubpage，以便下次申请内存时直接从该PoolSubpage上分配。
#5 从PoolArena中移除，并返回false，这时PoolChunk会将释放对应Page节点。

void free(long handle, int normCapacity, ByteBuffer nioBuffer) {
    if (isSubpage(handle)) {
        // #1
        int sizeIdx = arena.size2SizeIdx(normCapacity);
        PoolSubpage<T> head = arena.findSubpagePoolHead(sizeIdx);

        PoolSubpage<T> subpage = subpages[runOffset(handle)];
        assert subpage != null && subpage.doNotDestroy;

        synchronized (head) {
            // #2
            if (subpage.free(head, bitmapIdx(handle))) {
                //the subpage is still used, do not free it
                return;
            }
        }
    }

    // #3
    ...
}

#1
查找head节点，同步
#2
调用subpage#free释放Small内存块
若是subpage#free返回false，将继续向下执行，这时会释放PoolSubpage整个内存块，不然，不释放PoolSubpage内存块。
#3 释放Normal内存块，就是释放PoolSubpage整个内存块。该部份内容可参考《PoolChunk实现原理》。

若是您以为本文不错，欢迎关注个人微信公众号，系列文章持续更新中。您的关注是我坚持的动力！