Netty解析二十三：Netty内存分配PoolSubpage

上一篇解析了Netty内存分配的伙伴模型和PoolChunk的原理，如果需要分配的内存大于等于pageSize则会直接通过Chunk进行分配，不过不会产生Page对象；如果小于pageSize也就是分为tiny/small会通过Chunk然后分配Subpage，由Subpage来管理具体内存。

PoolSubpage解析

Page按固定大小（elemSize）切分一小块一小块的，这样的page叫做SubPage。在chunk中Page是一个虚拟的概念，由二叉树控制，当需要的内存空间小于PageSize的时候就需要对Page进行切割，此时就需要一个对象来管理这些被切割内存的分配。

分配过程是：chunk.allocate->chunk.allocateSubpage

继承关系以及成员变量

final class PoolSubpage<T> implements PoolSubpageMetric {

    // 持有的chunk
    final PoolChunk<T> chunk;
    // 在chunk中page的位置
    private final int memoryMapIdx;
    // chunk的offzet
    private final int runOffset;
    // chunk的pageSize
    private final int pageSize;
    // 位表，long中的每一位代表subPage中的一块内存使用情况
    private final long[] bitmap;
    // 构成链表
    PoolSubpage<T> prev;
    PoolSubpage<T> next;
    // 是否销毁
    boolean doNotDestroy;
    // SubPage中每一块的大小
    int elemSize;
    // 最大块个数
    private int maxNumElems;
    // 位表长度
    private int bitmapLength;
    // 下一个可以的块的位置
    private int nextAvail;
    // 可用块个数
    private int numAvail;
}

构造器

PoolSubpage也有两个构造器，一个是用于构造PoolSubpage链表的表头节点的，另一个是正常构造器。

链表头节点构造器

PoolSubpage(int pageSize) {
    chunk = null;
    memoryMapIdx = -1;
    runOffset = -1;
    elemSize = -1;
    this.pageSize = pageSize;
    bitmap = null;
}

普通构造器

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;
    // 一位代表一个内存块的分配情况，long是64位所以代表64块，而pageSize是通过PoolArena#normalizeCapacity方法优化过的，最小值为16
    // 所以最大的bitmap就是 pageSize / 16 / 64
    bitmap = new long[pageSize >>> 10]; // pageSize / 16 / 64
    // 初始化的核心方法
    init(head, elemSize);
}

初始init

在chunk分配Subpage对象的可以看到init方法和allocate方法是核心。

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();

void init(PoolSubpage<T> head, int elemSize) {
    // 设置为未销毁
    doNotDestroy = true;
    // 设置按多大进行分割的
    this.elemSize = elemSize;
    if (elemSize != 0) {
        // 计算块数，Page大小 / 每一小块的大小
        maxNumElems = numAvail = pageSize / elemSize;
        // 下一个可用的块
        nextAvail = 0;
        // 位表的长度 = 最大块数 / 64(long的位数)
        bitmapLength = maxNumElems >>> 6;
        // 最大块数 % 64(long的位数) > 0
        if ((maxNumElems & 63) != 0) {
            bitmapLength ++;
        }
        // 位表初始值0表示块还未使用
        for (int i = 0; i < bitmapLength; i ++) {
            bitmap[i] = 0;
        }
    }
    // 添加到池中
    addToPool(head);
}

添加到池中addToPool

PoolArena会管理chunk和Subpage，PoolArena中有两个PoolSubpage链表，一个存tiny大小的SubPage，另一个存small大小的Subpage。所谓的添加到池中就是将当前Subpage插入到链表的head后面一位，这样PoolArena就能通过链表来管理者SubPage。

private void addToPool(PoolSubpage<T> head) {
    assert prev == null && next == null;
    prev = head;
    next = head.next;
    // 当前节点插入到head都后面
    next.prev = this;
    head.next = this;
}

分配内存allocate

分配内存的逻辑计算计算哪个块可用bitmapIdx，然将对于的位置1。

long allocate() {
    if (elemSize == 0) {
        // 分配bitmap 0的块
        return toHandle(0);
    }
    // 没有可用的块或者已经销毁了
    if (numAvail == 0 || !doNotDestroy) {
        return -1;
    }
    // 下一个可用块
    final int bitmapIdx = getNextAvail();
    // bitmapIdx / 64 代表bitmapIdx在bitmap[]的序号
    int q = bitmapIdx >>> 6;
    // bitmapIdx % 64 代表bitmap[q]的哪一位（long的哪一位）
    int r = bitmapIdx & 63;
    // 判断bitmap[q]的第q位是不是为0，为0表示还没有被使用
    assert (bitmap[q] >>> r & 1) == 0;
    // 将bitmap[q]的第q位置位1，表示被使用
    bitmap[q] |= 1L << r;
    // 可用块数减一
    if (-- numAvail == 0) {
        // 可用块为0则移出pool
        removeFromPool();
    }
    // 计算这个位置的handle，高32是描述bitmapIdx，低32位描述chunk的page的序号
    return toHandle(bitmapIdx);
}

如果块大小为0，则直接分配第一块；
如果没有可用的块或者已经销毁了，直接返回-1；
获取下一个可用块的bitmapIdx；
校验bitmapIdx所在的块可用；
可用块数numAvail减一；
如果numAvail为0了，则将当前Subpage移出pool（链表）；
计算这个块的handle数（0x4000000000000000L | (long) bitmapIdx << 32 | memoryMapIdx）；

释放内存free

如果这个Subpage还再被使用则返回true;

如果这个Subpage没有被它的chunk使用且被释放了则返回false。

boolean free(PoolSubpage<T> head, int bitmapIdx) {
    if (elemSize == 0) {
        return true;
    }
    int q = bitmapIdx >>> 6;
    int r = bitmapIdx & 63;
    // 校验这个块是不是未释放
    assert (bitmap[q] >>> r & 1) != 0;
    // 块所在的位置0
    bitmap[q] ^= 1L << r;
    // 将下一个可用的块设置为刚释放的这个块
    setNextAvail(bitmapIdx);
    // 可用块数加一
    if (numAvail ++ == 0) {
        // 如果之前可用块数为0，则将当前节点加入链表
        addToPool(head);
        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;
        }
        // 如果在这个Subpage前面还有Subpage那就移出池
        // Remove this subpage from the pool if there are other subpages left in the pool.
        doNotDestroy = false;
        removeFromPool();
        return false;
    }
}

如果块大小为0，则返回true；
校验这个块是不是未释放；
块所在的位（bitmap）置0；
将下一个可用的块设置为刚释放的这个块（nextAvail = bitmapIdx）；
可用块数加一；
如果之前可用块数为0，则将当前节点加入链表，返回true；
如果还有块在被使用，则返回true；
全部块都可用：
1. 如果当前节点的前一个节点和后一个节点都相等（除头节点就只有一个节点），则返回true；
2. doNotDestroy置为false；
3. 将当前节点移出pool；
4. 返回false；

工具方法

生成带page的序号toHandle()

private long toHandle(int bitmapIdx) {
    // 0x4000000000000000L 第二位为1，其它都为0
    // 高32位为bitmapIdx，低32位chunk中page的序号
    // bitmapIdx和memoryMapIdx实际都是很小的，用32位表示绰绰有余
    return 0x4000000000000000L | (long) bitmapIdx << 32 | memoryMapIdx;
}

获取下一个可用的块getNextAvail()

getNextAvail有两种情况：

nextAvail存有值（>=），那么这个位置就是空闲的，返回这个值，并置为this.nextAvail = -1；
使用findNextAvail方法遍历bitmap数组查找那个位置是空闲的。

第一种情况是和内存释放free()方法相匹配的，释放内存块后会将对应的位置（bitmapIdx）设置为下一个可用的位置nextAvail，这样节约零碎空间遍历的时间成本。

第二种是常规的遍历查找，相对来说就会花时间。

private int getNextAvail() {
    int nextAvail = this.nextAvail;
    // 大于等于0就代表是可以的块
    if (nextAvail >= 0) {
        // 返回前置-1，下次就需要再计算
        this.nextAvail = -1;
        return nextAvail;
    }
    return findNextAvail();
}

查找下一个可用的块findNextAvail()

findNextAvail确定bitmap数组哪一位还有空余空间，然后交由findNextAvail0查找哪一位还有未分配。

private int findNextAvail() {
    final long[] bitmap = this.bitmap;
    final int bitmapLength = this.bitmapLength;
    // 遍历bitmap
    for (int i = 0; i < bitmapLength; i ++) {
        long bits = bitmap[i];
        // 没有全部为1，即还用空
        if (~bits != 0) {
            // 查找到哪一个
            return findNextAvail0(i, bits);
        }
    }
    return -1;
}

findNextAvail0遍历64位字节，确定哪一位有空余，遍历从低位开始。

private int findNextAvail0(int i, long bits) {
    final int maxNumElems = this.maxNumElems;
    // i*64,表示数组中位置
    final int baseVal = i << 6;
    // 遍历每一位，确定位置
    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;
}

removeFromPool()

在Subpage的所有块都被使用时会将当前Subpage移出链表。

private void removeFromPool() {
    assert prev != null && next != null;
    prev.next = next;
    next.prev = prev;
    next = null;
    prev = null;
}