netty源码阅读之ByteBuf之内存分配器PooledByteBufAllocator

一时失言乱红尘 2022-05-15 03:42 211阅读 0赞

那么照例我们从以下两点进行分析吧：

1、拿到线程局部缓存PoolThreadCache

2、在线程局部缓存的Arena上进行分配

我们从PooledByteBufAllocator的newHeapBuffer方法入手：

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

首先看 PoolThreadCache cache = threadCache.get();这一句的意思是把PoolThreadCache取出来。

(newHeapBuffer是在多线程环境下执行的，threadCache.get()可以拿到当前线程的cache)

PoolArena<byte\[\]> heapArena = cache.heapArena;的意思是取出cache里面的heapArena。

PoolThreadLocalCache

首先看看threadCache的定义PoolThreadLocalCache：

final class PoolThreadLocalCache extends FastThreadLocal<PoolThreadCache> {
    
            @Override
            protected synchronized PoolThreadCache initialValue() {
                final PoolArena<byte[]> heapArena = leastUsedArena(heapArenas);
                final PoolArena<ByteBuffer> directArena = leastUsedArena(directArenas);
    
                return new PoolThreadCache(
                        heapArena, directArena, tinyCacheSize, smallCacheSize, normalCacheSize,
                        DEFAULT_MAX_CACHED_BUFFER_CAPACITY, DEFAULT_CACHE_TRIM_INTERVAL);
            }
         ...
    }

继承自FastThreadLocal，FastThreadLocal我们可以简单的理解为ThreadLocal的快速版。在调用initialValue方法的时候，会创建heapArena和directArena，并且通过这两个对象返回一个PoolThreadCache。(在PoolThreadLocalCache第一次调用get的时候，没有获取到对象，就会调用initialValue方法产生PoolThreadCache)

回到PoolArena<byte\[\]> heapArena = cache.heapArena;，heapArena为何物？是PoolThreadCache的heapArena，我们从上一段代码返回PooleThreadCache进入，也就是这里进去：

return new PoolThreadCache(
                        heapArena, directArena, tinyCacheSize, smallCacheSize, normalCacheSize,
                        DEFAULT_MAX_CACHED_BUFFER_CAPACITY, DEFAULT_CACHE_TRIM_INTERVAL);

可以看到：

PoolThreadCache(PoolArena<byte[]> heapArena, PoolArena<ByteBuffer> directArena,
                        int tinyCacheSize, int smallCacheSize, int normalCacheSize,
                        int maxCachedBufferCapacity, int freeSweepAllocationThreshold) {
            if (maxCachedBufferCapacity < 0) {
                throw new IllegalArgumentException("maxCachedBufferCapacity: "
                        + maxCachedBufferCapacity + " (expected: >= 0)");
            }
            if (freeSweepAllocationThreshold < 1) {
                throw new IllegalArgumentException("freeSweepAllocationThreshold: "
                        + freeSweepAllocationThreshold + " (expected: > 0)");
            }
            this.freeSweepAllocationThreshold = freeSweepAllocationThreshold;
            this.heapArena = heapArena;
            this.directArena = directArena;
            ...
        }

PoolThreadCache的heapArena是PoolThreadLocalCache创建它的时候传入进去的的heapArena。

那我们再看看PoolThreadLocalCache的heapArena是从哪里来，有一行代码（从字面意思我们可以理解这是从heapArenas里面拿到最小使用的heapArenas）：

final PoolArena<byte[]> heapArena = leastUsedArena(heapArenas);

找找这个heapArenas的定义：

private final PoolArena<byte[]>[] heapArenas;

它是PooledByteBufAllocator类的一个成员变量。查找heapArenas在哪里引用，（idea的话在变量定义的地方按住ctrl点击变量就能看到在哪里引用）：

public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder,
                                      int tinyCacheSize, int smallCacheSize, int normalCacheSize) {
            ...
            if (nHeapArena > 0) {
                heapArenas = newArenaArray(nHeapArena);
                List<PoolArenaMetric> metrics = new ArrayList<PoolArenaMetric>(heapArenas.length);
                for (int i = 0; i < heapArenas.length; i ++) {
                    PoolArena.HeapArena arena = new PoolArena.HeapArena(this, pageSize, maxOrder, pageShifts, chunkSize);
                    heapArenas[i] = arena;
                    metrics.add(arena);
                }
                heapArenaMetrics = Collections.unmodifiableList(metrics);
            } else {
                heapArenas = null;
                heapArenaMetrics = Collections.emptyList();
            }
            ...  
        }

这是在PooledByteBufAllocator的构造函数里面完成的，也就是在我创建PooledByteBufAllocator的时候，我就创建这个heapArenas，

heapArenas = newArenaArray(nHeapArena);

这是一个数组，然后往里面填充内容：

for (int i = 0; i < heapArenas.length; i ++) {
                    PoolArena.HeapArena arena = new PoolArena.HeapArena(this, pageSize, maxOrder, pageShifts, chunkSize);
                    heapArenas[i] = arena;
                    metrics.add(arena);
                }

上一段代码的nHeapArena一看就是定义这个数组的大小，它从哪里来呢，我们找到调用它的方法：

public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder) {
            this(preferDirect, nHeapArena, nDirectArena, pageSize, maxOrder,
                    DEFAULT_TINY_CACHE_SIZE, DEFAULT_SMALL_CACHE_SIZE, DEFAULT_NORMAL_CACHE_SIZE);
        }

然后我们继续向上找：

public PooledByteBufAllocator(boolean preferDirect) {
            this(preferDirect, DEFAULT_NUM_HEAP_ARENA, DEFAULT_NUM_DIRECT_ARENA, DEFAULT_PAGE_SIZE, DEFAULT_MAX_ORDER);
        }

这个DEFAULT\_NUM\_HEAP\_ARENA就是我们要找的东西，找到定义：

private static final int DEFAULT_NUM_HEAP_ARENA;

然后找到赋值的地方，在一个static块里面：

// Use 2 * cores by default to reduce condition as we use 2 * cores for the number of EventLoops
            // in NIO and EPOLL as well. If we choose a smaller number we will run into hotspots as allocation and
            // deallocation needs to be synchronized on the PoolArena.
            // See https://github.com/netty/netty/issues/3888
            final int defaultMinNumArena = runtime.availableProcessors() * 2;
            final int defaultChunkSize = DEFAULT_PAGE_SIZE << DEFAULT_MAX_ORDER;
            DEFAULT_NUM_HEAP_ARENA = Math.max(0,
                    SystemPropertyUtil.getInt(
                            "io.netty.allocator.numHeapArenas",
                            (int) Math.min(
                                    defaultMinNumArena,
                                    runtime.maxMemory() / defaultChunkSize / 2 / 3)));

如果没有定义io.netty.allocator.numHeapArenas这个属性，那么就取defaultMinNumArena和runtime.maxMemory() / defaultChunkSize / 2 / 3的最小值，一般是defaultMinNumArena，也就是核心线程的2倍。

这个和我们[这篇文章][Link 1]呼应了。也就是默认情况下，NioEventloop的线程数量，和arena的数量是相同的，可以保证每一个线程都有一个arena，也为保证线程安全奠定了基础。

然后我们总结一下，一个PooledByteBufAllocator创建的时候，会创建两个Arean（一个heap和一个direct）数组，这个数组里的arena会分配给每个线程（意思就是每个线程有两个arena）。一个PooledByteBufAllocator里面会有一个PoolThreadLocalCache，这个PoolThreadLocalCache，调用get的时候，会得到当前线程的PoolThreadCache，通过当前线程的这个PoolThreadCache 就能读取到directArena，然后就在这个directArena这里开始分配内存了。

分配内存，除了在poolThreadCache的arena区分配，还可以在poolThreadCache的buffer里面（缓存）分配。buffer里面，根据分配内容的大小，分为tinyCacheSize、smallCacheSize、 normalCacheSize。我们从PooledByteBufAllocator的成员变量开始看起：

private final int tinyCacheSize;
        private final int smallCacheSize;
        private final int normalCacheSize;

这写成员变量，通过PoolThreadLocalCache的initalValue方法传入，PoolThreadCache类的构造器，到达PoolThreadCache：

PoolThreadCache(PoolArena<byte[]> heapArena, PoolArena<ByteBuffer> directArena,
                        int tinyCacheSize, int smallCacheSize, int normalCacheSize,
                        int maxCachedBufferCapacity, int freeSweepAllocationThreshold) {
            ...
            if (heapArena != null) {
                // Create the caches for the heap allocations
                tinySubPageHeapCaches = createSubPageCaches(
                        tinyCacheSize, PoolArena.numTinySubpagePools, SizeClass.Tiny);
                smallSubPageHeapCaches = createSubPageCaches(
                        smallCacheSize, heapArena.numSmallSubpagePools, SizeClass.Small);
    
                numShiftsNormalHeap = log2(heapArena.pageSize);
                normalHeapCaches = createNormalCaches(
                        normalCacheSize, maxCachedBufferCapacity, heapArena);
    
                heapArena.numThreadCaches.getAndIncrement();
            } else {
                // No heapArea is configured so just null out all caches
                tinySubPageHeapCaches = null;
                smallSubPageHeapCaches = null;
                normalHeapCaches = null;
                numShiftsNormalHeap = -1;
            }
            ...
        }

通过createSubPageCaches等这些方法，创建buffer：

private static <T> MemoryRegionCache<T>[] createSubPageCaches(
                int cacheSize, int numCaches, SizeClass sizeClass) {
            if (cacheSize > 0) {
                @SuppressWarnings("unchecked")
                MemoryRegionCache<T>[] cache = new MemoryRegionCache[numCaches];
                for (int i = 0; i < cache.length; i++) {
                    // TODO: maybe use cacheSize / cache.length
                    cache[i] = new SubPageMemoryRegionCache<T>(cacheSize, sizeClass);
                }
                return cache;
            } else {
                return null;
            }
        }

然后是SubPageMemoryRegionCache：

SubPageMemoryRegionCache(int size, SizeClass sizeClass) {
                super(size, sizeClass);
            }

父类构造方法：

MemoryRegionCache(int size, SizeClass sizeClass) {
                this.size = MathUtil.safeFindNextPositivePowerOfTwo(size);
                queue = PlatformDependent.newFixedMpscQueue(this.size);
                this.sizeClass = sizeClass;
            }

也就是在这里创建一个size，为2的n次方，并且创建一个queue，组合成一个MemoryRegionCache。这里我们先点到这里，后门的文章会深入讲解这个MemoryRegionCache。

[Link 1]: https://blog.csdn.net/fst438060684/article/details/81534359