Netty源码—7.ByteBuf原理三

PooledByteBufAllocator allocator = PooledByteBufAllocator.DEFAULT;
ByteBuf byteBuf = allocator.directBuffer(16);//分配16B内存
byteBuf.release();

abstract class PoolArena<T> implements PoolArenaMetric {
    //有32个元素：16B、32B、48B、...、480、496B
    private final PoolSubpage<T>[] tinySubpagePools;
    //有4个元素：512B、1K、2K、4K
    private final PoolSubpage<T>[] smallSubpagePools;
    ...    
    PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) {
        PooledByteBuf<T> buf = newByteBuf(maxCapacity);//创建ByteBuf对象
        allocate(cache, buf, reqCapacity);//基于PoolThreadCache对ByteBuf对象进行内存分配
        return buf;
    }
    
    private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) {
        //1.根据reqCapacity进行分段规格化
        final int normCapacity = normalizeCapacity(reqCapacity);
        if (isTinyOrSmall(normCapacity)) {//capacity < pageSize，需要分配的内存小于8K
            int tableIdx;
            PoolSubpage<T>[] table;
            boolean tiny = isTiny(normCapacity);
            if (tiny) {//capacity < 512
                if (cache.allocateTiny(this, buf, reqCapacity, normCapacity)) {
                    //was able to allocate out of the cache so move on
                    return;
                }
                //根据规格化后的需要分配的内存大小normCapacity，获取tableIdx
                tableIdx = tinyIdx(normCapacity);
                table = tinySubpagePools;
            } else {
                if (cache.allocateSmall(this, buf, reqCapacity, normCapacity)) {
                    //was able to allocate out of the cache so move on
                    return;
                }
                //根据规格化后的需要分配的内存大小normCapacity，获取tableIdx
                tableIdx = smallIdx(normCapacity);
                table = smallSubpagePools;
            }

            final PoolSubpage<T> head = table[tableIdx];

            //Synchronize on the head. 
            //This is needed as PoolChunk#allocateSubpage(int) and PoolChunk#free(long) may modify the doubly linked list as well. 
            synchronized (head) {
                final PoolSubpage<T> s = head.next;
                if (s != head) {//PoolArena的tinySubpagePools数组中有可以分配内存的PoolSubpage
                    assert s.doNotDestroy && s.elemSize == normCapacity;
                    //调用SubPage的allocate()方法进行内存分配
                    long handle = s.allocate();
                    assert handle >= 0;
                    s.chunk.initBufWithSubpage(buf, handle, reqCapacity);

                    if (tiny) {
                        allocationsTiny.increment();
                    } else {
                        allocationsSmall.increment();
                    }
                    return;
                }
            }
            allocateNormal(buf, reqCapacity, normCapacity);
            return;
        }
        ...
    }
    
    static int tinyIdx(int normCapacity) {
        return normCapacity >>> 4;
    }

    static int smallIdx(int normCapacity) {
        int tableIdx = 0;
        int i = normCapacity >>> 10;
        while (i != 0) {
            i >>>= 1;
            tableIdx ++;
        }
        return tableIdx;
    }
    
    private synchronized void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
        //1.尝试在现有的PoolChunk上分配
        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)) {
            ++allocationsNormal;
            return;
        }

        //2.创建一个PoolChunk并进行内存分配
        PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize);
        //由handle指向PoolChunk里的一块连续内存
        long handle = c.allocate(normCapacity);
        ++allocationsNormal;
        assert handle > 0;
        //3.初始化PooledByteBuf对象
        c.initBuf(buf, handle, reqCapacity);
        //4.将新建的PoolChunk添加到PoolArena的qInit这个PoolChunkList中
        qInit.add(c);
    }
    ...
}

final class PoolChunk<T> implements PoolChunkMetric {
    ...
    long allocate(int normCapacity) {
        if ((normCapacity & subpageOverflowMask) != 0) {//normCapacity >= pageSize
            //Page级别的内存分配
            return allocateRun(normCapacity);
        } else {
            //SubPage级别的内存分配
            return allocateSubpage(normCapacity);
        }
    }
    ...
}

final class PoolChunk<T> implements PoolChunkMetric {
    final PoolArena<T> arena;
    final T memory;//内存
    //一个Page的大小，比如8K
    private final int pageSize;
    //4096个元素的字节数组，表示不同规格的连续内存使用分配情况，用二叉树理解
    private final byte[] memoryMap;
    private final byte[] depthMap;
    //2048个元素的数组，表示Chunk里哪些Page是以SubPage方式存在的
    //由于一个PoolChunk是16M，会以8K为标准划分一个个的Page，所以会有16 * 1024 / 8 = 2048个Page
    private final PoolSubpage<T>[] subpages;
    //Used to mark memory as unusable
    private final byte unusable;//默认是12
    ...
    //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) {
        //Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.
        //This is need as we may add it back and so alter the linked-list structure.
        //1.定位一个SubPage对象
        //PoolArena的findSubpagePoolHead()方法会通过除以16，来获取用来存放16B的PoolSubpage节点
        PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity);
        synchronized (head) {
            //只能从层高为11的memoryMap中获取SubPage，因为只有这一层的每个结点都是8K
            int d = maxOrder;//subpages are only be allocated from pages i.e., leaves
            //1.定位一个SubPage对象：在平衡二叉树的第11层上分配一个结点
            //即通过allocateNode(d)找到一个Page在PoolChunk中的下标idx
            int id = allocateNode(d);
            if (id < 0) {
                return id;
            }

            final PoolSubpage<T>[] subpages = this.subpages;
            final int pageSize = this.pageSize;
            freeBytes -= pageSize;
           
            //1.定位一个SubPage对象：确定PoolChunk中第几个Page会以SubPage方式存在
            int subpageIdx = subpageIdx(id);
            PoolSubpage<T> subpage = subpages[subpageIdx];
            //2.初始化SubPage对象
            if (subpage == null) {
                subpage = new PoolSubpage<T>(head, this, id, runOffset(id), pageSize, normCapacity);
                subpages[subpageIdx] = subpage;
            } else {
                subpage.init(head, normCapacity);
            }
            //3.调用SubPage的allocate()方法进行分配
            return subpage.allocate();
        }
    }
    ...
}

abstract class PoolArena<T> implements PoolArenaMetric {
    //有32个元素：16B、32B、48B、...、480、496B
    private final PoolSubpage<T>[] tinySubpagePools;
    //有4个元素：512B、1K、2K、4K
    private final PoolSubpage<T>[] smallSubpagePools;
    ...
    //找到在PoolArena的tinySubpagePools中用于存放16B的PoolSubpage结点
    PoolSubpage<T> findSubpagePoolHead(int elemSize) {
        int tableIdx;
        PoolSubpage<T>[] table;
        if (isTiny(elemSize)) {//< 512
            tableIdx = elemSize >>> 4;
            table = tinySubpagePools;
        } else {
            tableIdx = 0;
            elemSize >>>= 10;
            while (elemSize != 0) {
                elemSize >>>= 1;
                tableIdx ++;
            }
            table = smallSubpagePools;
        }
        return table[tableIdx];
    }
    ...
}

final class PoolChunk<T> implements PoolChunkMetric {
    ...
    //Algorithm to allocate an index in memoryMap when we query for a free node at depth d
    //@param d depth
    //@return index in memoryMap
    private int allocateNode(int d) {
        int id = 1;
        int initial = - (1 << d);//has last d bits = 0 and rest all = 1
        //取出memoryMap[id]的值
        byte val = value(id);
        if (val > d) {//val = unusable = 12
            return -1;
        }
        //val < d表示当前结点可用
        while (val < d || (id & initial) == 0) {//id & initial == 1 << d for all ids at depth d, for < d it is 0
            id <<= 1;//每循环一次乘以2
            val = value(id);
            if (val > d) {//如果当前id对应的结点不可用
                id ^= 1;//通过异或运算实现对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 = 12
        updateParentsAlloc(id);//逐层往上标记结点不可用
        return id;
    }
    
    private int subpageIdx(int memoryMapIdx) {
        return memoryMapIdx ^ maxSubpageAllocs;//remove highest set bit, to get offset
    }
    ...
}

final class PoolSubpage<T> implements PoolSubpageMetric {
    final PoolChunk<T> chunk;
    private final int memoryMapIdx;
    private final int runOffset;
    private final int pageSize;
    private final long[] bitmap;
    ...
    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) {
            maxNumElems = numAvail = pageSize / elemSize;
            nextAvail = 0;
            bitmapLength = maxNumElems >>> 6;
            if ((maxNumElems & 63) != 0) {
                bitmapLength ++;
            }
            for (int i = 0; i < bitmapLength; i ++) {
                bitmap[i] = 0;
            }
        }
        //将当前PoolSubpage对象添加到PoolArena的tinySubpagePools数组中用于存放
        //可以分配16B内存的PoolSubpage链表中
        addToPool(head);
    }
    
    private void addToPool(PoolSubpage<T> head) {
        assert prev == null && next == null;
        prev = head;
        next = head.next;
        next.prev = this;
        head.next = this;
    }
    ...
}

final class PoolSubpage<T> implements PoolSubpageMetric {
    int elemSize;
    private final long[] bitmap;
    private int numAvail;
    private final int memoryMapIdx;
    ...
    //Returns the bitmap index of the subpage allocation.
    long allocate() {
        if (elemSize == 0) {
            return toHandle(0);
        }
        if (numAvail == 0 || !doNotDestroy) {
            return -1;
        }
        //1.先从位图bitmap中寻找一个未被使用的SubPage
        final int bitmapIdx = getNextAvail();
        int q = bitmapIdx >>> 6;
        int r = bitmapIdx & 63;
        assert (bitmap[q] >>> r & 1) == 0;
        bitmap[q] |= 1L << r;
      
        //如果可用的SubPage为0
        if (-- numAvail == 0) {
            //把该PoolSubpage对象从PoolArena的tinySubpagePools数组的某种规格的结点中移除
            removeFromPool();
        }
        //2.将bitmapIdx转换成handle
        return toHandle(bitmapIdx);
    }
    
    private long toHandle(int bitmapIdx) {
        //拼接成64位，bitmapIdx变成高32位，memoryMapIdx变成低32位
        return 0x4000000000000000L | (long) bitmapIdx << 32 | memoryMapIdx;
    }
    ...
}

public abstract class AbstractReferenceCountedByteBuf extends AbstractByteBuf {
    ...
    //byteBuf执行release()方法释放内存
    @Override
    public boolean release() {
        return release0(1);
    }
    
    private boolean release0(int decrement) {
        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;
            }
        }
    }

    protected abstract void deallocate();
    ...
}

abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
    ...
    protected PoolChunk<T> chunk;
    protected long handle;
    protected T memory;
    ...
    @Override
    protected final void deallocate() {
        if (handle >= 0) {
            //1.清空PooledByteBuf对象的handle、chunk、memory变量值
            final long handle = this.handle;
            this.handle = -1;
            memory = null;
            //2.调用PoolArena的free()方法去释放内存
            //也就是将连续内存的区段添加到缓存 + 标记连续内存的区段为未使用；
            chunk.arena.free(chunk, handle, maxLength, cache);
            //3.调用PooledByteBuf的recycle()方法，将PooledByteBuf对象添加到对象池
            recycle();
        }
    }
}

abstract class PoolArena<T> implements PoolArenaMetric {
    private final PoolSubpage<T>[] tinySubpagePools;
    private final PoolSubpage<T>[] smallSubpagePools;
    
    private final PoolChunkList<T> qInit;
    private final PoolChunkList<T> q000;
    private final PoolChunkList<T> q025;
    private final PoolChunkList<T> q050;
    private final PoolChunkList<T> q075;
    private final PoolChunkList<T> q100;
    ...
    void free(PoolChunk<T> chunk, long handle, int normCapacity, PoolThreadCache cache) {
        if (chunk.unpooled) {
            int size = chunk.chunkSize();
            destroyChunk(chunk);
            activeBytesHuge.add(-size);
            deallocationsHuge.increment();
        } else {
            //判断要释放的内存大小属于哪一种规格
            SizeClass sizeClass = sizeClass(normCapacity);
            //1.调用PoolThreadCache的add()方法将释放的连续内存区段添加到缓存
            if (cache != null && cache.add(this, chunk, handle, normCapacity, sizeClass)) {
                //cached so not free it.
                return;
            }
            //2.调用PoolArena的freeChunk()方法释放内存，也就是标记连续内存的区段为未使用
            freeChunk(chunk, handle, sizeClass);
        }
    }
    
    private SizeClass sizeClass(int normCapacity) {
        if (!isTinyOrSmall(normCapacity)) {
            return SizeClass.Normal;
        }
        return isTiny(normCapacity) ? SizeClass.Tiny : SizeClass.Small;
    }
    ...
}

final class PoolThreadCache {
    private final MemoryRegionCache<byte[]>[] tinySubPageHeapCaches;
    private final MemoryRegionCache<byte[]>[] smallSubPageHeapCaches;
    private final MemoryRegionCache<byte[]>[] normalHeapCaches;
    private final MemoryRegionCache<ByteBuffer>[] tinySubPageDirectCaches;
    private final MemoryRegionCache<ByteBuffer>[] smallSubPageDirectCaches;
    private final MemoryRegionCache<ByteBuffer>[] normalDirectCaches;
    ...
    //Add PoolChunk and handle to the cache if there is enough room.
    //Returns true if it fit into the cache false otherwise.
    @SuppressWarnings({ "unchecked", "rawtypes" })
    boolean add(PoolArena<?> area, PoolChunk chunk, long handle, int normCapacity, SizeClass sizeClass) {
        MemoryRegionCache<?> cache = cache(area, normCapacity, sizeClass);
        if (cache == null) {
            return false;
        }
        //将PoolChunk的连续内存区段添加到缓存
        return cache.add(chunk, handle);
    }

    private MemoryRegionCache<?> cache(PoolArena<?> area, int normCapacity, SizeClass sizeClass) {
        switch (sizeClass) {
        case Normal:
            return cacheForNormal(area, normCapacity);
        case Small:
            return cacheForSmall(area, normCapacity);
        case Tiny:
            return cacheForTiny(area, normCapacity);
        default:
            throw new Error();
        }
    }
    
    private MemoryRegionCache<?> cacheForTiny(PoolArena<?> area, int normCapacity) {
        int idx = PoolArena.tinyIdx(normCapacity);
        if (area.isDirect()) {
            return cache(tinySubPageDirectCaches, idx);
        }
        return cache(tinySubPageHeapCaches, idx);
    }

    private MemoryRegionCache<?> cacheForSmall(PoolArena<?> area, int normCapacity) {
        int idx = PoolArena.smallIdx(normCapacity);
        if (area.isDirect()) {
            return cache(smallSubPageDirectCaches, idx);
        }
        return cache(smallSubPageHeapCaches, idx);
    }

    private MemoryRegionCache<?> cacheForNormal(PoolArena<?> area, int normCapacity) {
        if (area.isDirect()) {
            int idx = log2(normCapacity >> numShiftsNormalDirect);
            return cache(normalDirectCaches, idx);
        }
        int idx = log2(normCapacity >> numShiftsNormalHeap);
        return cache(normalHeapCaches, idx);
    }

    private static <T> MemoryRegionCache<T> cache(MemoryRegionCache<T>[] cache, int idx) {
        if (cache == null || idx > cache.length - 1) {
            return null;
        }
        return cache[idx];
    }
    ...
    private abstract static class MemoryRegionCache<T> {
        private final Queue<Entry<T>> queue;
        ...
        //Add to cache if not already full.
        //将PoolChunk的连续内存区段添加到缓存
        @SuppressWarnings("unchecked")
        public final boolean add(PoolChunk<T> chunk, long handle) {
            Entry<T> entry = newEntry(chunk, handle);
            boolean queued = queue.offer(entry);
            if (!queued) {
                //If it was not possible to cache the chunk, immediately recycle the entry
                entry.recycle();
            }
            return queued;
        }
        ...
    }
}

abstract class PoolArena<T> implements PoolArenaMetric {
    ...
    void freeChunk(PoolChunk<T> chunk, long handle, SizeClass sizeClass) {
        final boolean destroyChunk;
        synchronized (this) {
            switch (sizeClass) {
            case Normal:
                ++deallocationsNormal;
                break;
            case Small:
                ++deallocationsSmall;
                break;
            case Tiny:
                ++deallocationsTiny;
                break;
            default:
                throw new Error();
            }
            //调用PoolChunk的parent也就是PoolChunkList的free()方法释放PoolChunk
            destroyChunk = !chunk.parent.free(chunk, handle);
        }
        if (destroyChunk) {
            //destroyChunk not need to be called while holding the synchronized lock.
            destroyChunk(chunk);
        }
    }
    ...
}

final class PoolChunkList<T> implements PoolChunkListMetric {
    private PoolChunk<T> head;
    private PoolChunkList<T> prevList;
    ...
    boolean free(PoolChunk<T> chunk, long handle) {
        //标记PoolChunk中连续内存区段为未使用
        chunk.free(handle);
        //如果要释放的PoolChunk的使用率小于当前PoolChunkList的最小使用率
        if (chunk.usage() < minUsage) {
            //从当前PoolChunkList中移除PoolChunk
            remove(chunk);
            //将PoolChunk添加到当前PoolChunkList的下一个PoolChunkList中
            return move0(chunk);
        }
        return true;
    }
    
    private void remove(PoolChunk<T> cur) {
        if (cur == head) {
            head = cur.next;
            if (head != null) {
                head.prev = null;
            }
        } else {
            PoolChunk<T> next = cur.next;
            cur.prev.next = next;
            if (next != null) {
                next.prev = cur.prev;
            }
        }
    }
    
    //Moves the PoolChunk down the PoolChunkList linked-list so it will end up in the right PoolChunkList 
    //that has the correct minUsage / maxUsage in respect to PoolChunk#usage().
    private boolean move0(PoolChunk<T> chunk) {
        if (prevList == null) {
            //There is no previous PoolChunkList so return false which result in having the PoolChunk destroyed and
            //all memory associated with the PoolChunk will be released.
            assert chunk.usage() == 0;
            return false;
        }
        return prevList.move(chunk);
    }
    ...
}

final class PoolChunk<T> implements PoolChunkMetric {
    final PoolArena<T> arena;
    private final PoolSubpage<T>[] subpages;
    ...
    //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 = memoryMapIdx(handle);
        int bitmapIdx = bitmapIdx(handle);

        if (bitmapIdx != 0) { // free a subpage
            PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
            assert subpage != null && subpage.doNotDestroy;

            //Obtain the head of the PoolSubPage pool that is owned by the PoolArena and synchronize on it.
            //This is need as we may add it back and so alter the linked-list structure.
            PoolSubpage<T> head = arena.findSubpagePoolHead(subpage.elemSize);
            synchronized (head) {
                //2.SubPage级别通过位图进行标记
                if (subpage.free(head, bitmapIdx & 0x3FFFFFFF)) {
                    return;
                }
            }
        }
        //1.Page级别根据二叉树来进行标记
        freeBytes += runLength(memoryMapIdx);
        setValue(memoryMapIdx, depth(memoryMapIdx));
        updateParentsFree(memoryMapIdx);
    }
    ...
}

abstract class PooledByteBuf<T> extends AbstractReferenceCountedByteBuf {
    private final Recycler.Handle<PooledByteBuf<T>> recyclerHandle;
    ...
    private void recycle() {
        recyclerHandle.recycle(this);
    }
    ...
}