netty源码解析(4.0)-28 ByteBuf内存池:PooledByteBufAllocator-把一切组装起来
PooledByteBufAllocator负责初始化PoolArena(PA)和PoolThreadCache(PTC)。它提供了一系列的接口,用来创建使用堆内存或直接内存的PooledByteBuf对象,这些接口只是一张皮,内部完全使用了PA和PTC的能力。初始化过程分两个步骤,首先初始化一系列的默认参数,然后初始化PTC对象和PA数组。
默认参数和它们的值
DEFAULT_PAGE_SIZE: PoolChunk中的page的大小-pageSize, 使用-Dio.netty.allocator.pageSize设置, 默认值:8192。
DEFAULT_MAX_ORDER: PoolChunk中二叉树的高度: maxOrder, 使用-Dio.netty.allocator.maxOrder设置,默认值:11。
DEFAULT_NUM_HEAP_ARENA: 使用堆内存的PA数组的长度,使用-Dio.netty.allocator.numHeapArenas设置,默认值: CPU核心数 * 2。
DEFAULT_NUM_DIRECT_ARENA: 使用直接内存的PA数组的长度,使用-Dio.netty.allocator.numHeapArenas设置,默认值: CPU核心数 * 2。
DEFAULT_TINY_CACHE_SIZE: PTC对象中每个用来缓存Tiny内存的MemoryRegionCache对象中queue的长度,使用-Dio.netty.allocator.tinyCacheSize设置,默认值:512。
DEFAULT_SMALL_CACHE_SIZE: PTC对象中每个用来缓存Small内存的MemoryRegionCache对象中queue的长度,使用-Dio.netty.allocator.smallCacheSize设置,默认值:256。
DEFAULT_NORMAL_CACHE_SIZE: PTC对象中每个用来缓存Normal内存的MemoryRegionCache对象中queue的长度,使用-Dio.netty.allocator.normalCacheSize设置,默认值:64。
DEFAULT_MAX_CACHED_BUFFER_CAPACITY: PTC对象中缓存Normal内存的大小上限。使用-Dio.netty.allocator.maxCachedBufferCapacity设置,默认值32 * 1024。
DEFAULT_CACHE_TRIM_INTERVAL: PTC对象中释放缓存的内存阈值。当PTC分配内存次数大于这个值时会释放缓存的内存。使用-Dio.netty.allocator.cacheTrimInterval设置,默认值:8192。
DEFAULT_USE_CACHE_FOR_ALL_THREADS: 是否对所有的线程使用缓存。使用-Dio.netty.allocator.useCacheForAllThreads设置,默认值:true。
DEFAULT_DIRECT_MEMORY_CACHE_ALIGNMENT: 直接内存的对齐参数,分配直接内存的大小必须是它的整数倍。使用-Dio.netty.allocator.directMemoryCacheAlignment设置,默认值:0, 表示不对齐。
初始化PoolArena数组
PooledByteBufAllocator维护了两个数组:
PoolArena<byte[]>[] heapArenas; PoolArena<ByteBuffer>[] directArenas;
heapArenas用来管理堆内存,directArenas用来管理直接内存。这两个数组在构造方法中初始化,构造方法的定义是:
public PooledByteBufAllocator(boolean preferDirect, int nHeapArena, int nDirectArena, int pageSize, int maxOrder, int tinyCacheSize, int smallCacheSize, int normalCacheSize, boolean useCacheForAllThreads, int directMemoryCacheAlignment)
prefreDirect: 创建PooledByteBuf时,是否优先使用直接内存。
nHeapArena: 默认使用DEFAULT_NUM_HEAP_ARENA。
nDirectArena: 默认使用DEFAULT_NUM_DIRECT_ARENA。
pageSize: 默认使用的DEFAULT_PAGE_SIZE。
maxOrder: 默认使用DEFAULT_MAX_ORDER。
tinyCacheSize: 默认使用DEFAULT_TINY_CACHE_SIZE。
smallCacheSize: 默认使用DEFAULT_SMALL_CACHE_SIZE。
normalCacheSize: 默认使用DEFAULT_NORMAL_CACHE_SIZE。
useCacheForAllThreads: 默认使用DEFAULT_USE_CACHE_FOR_ALL_THREADS。
directMemoryCacheAlignment: 默认使用DEFAULT_DIRECT_MEMORY_CACHE_ALIGNMENT。
这两数组的初始化代码如下:
int pageShifts = validateAndCalculatePageShifts(pageSize); 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, directMemoryCacheAlignment); heapArenas[i] = arena; metrics.add(arena); } heapArenaMetrics = Collections.unmodifiableList(metrics); } else { heapArenas = null; heapArenaMetrics = Collections.emptyList(); } if (nDirectArena > 0) { directArenas = newArenaArray(nDirectArena); List<PoolArenaMetric> metrics = new ArrayList<PoolArenaMetric>(directArenas.length); for (int i = 0; i < directArenas.length; i ++) { PoolArena.DirectArena arena = new PoolArena.DirectArena( this, pageSize, maxOrder, pageShifts, chunkSize, directMemoryCacheAlignment); directArenas[i] = arena; metrics.add(arena); } directArenaMetrics = Collections.unmodifiableList(metrics); } else { directArenas = null; directArenaMetrics = Collections.emptyList(); }
1行,计算pageShifts,算法是pageShifts = Integer.SIZE - 1 - Integer.numberOfLeadingZeros(pageSize) = 31 - Integer.numberOfLeadingZeros(pageSize)。 Integer.numberOfLeadingZeros(pageSize)是pageSize(32位整数)从最高位起连续是0的位数,因此pageShifts可以简化为pageShifts = log2(pageSize)。
4,20行,创建数组,new PoolArena[size]。
6-12,22-17行, 初始化数组中的PoolArena对象,分别使用PooArena的两个内部类: HeapArena, DirectArena。
初始化PoolThreadCache
PoolThreadCache使用PoolThreadLocalCache(PTLC)间接初始化,PTLC是PooledByteBufAllocator的内部内,它的定义如下:
final class PoolThreadLocalCache extends FastThreadLocal<PoolThreadCache>
这个类派生自io.netty.util.concurrent.FastThreadLocal<T>, 和java.lang.ThreadLocal<T>功能一样,实现了线程本地存储(TLS)的功能,不同的是FastThreadLocal<T>优化了访问性能。PTLC覆盖了父类的initialValue方法,这个方法负责初始化线程本地的PoolThreadCache对象。当第一次调用PTLC对象的get方法时,这个方法会被调用。
@Override protected synchronized PoolThreadCache initialValue() { final PoolArena<byte[]> heapArena = leastUsedArena(heapArenas); final PoolArena<ByteBuffer> directArena = leastUsedArena(directArenas); if (useCacheForAllThreads || Thread.currentThread() instanceof FastThreadLocalThread) { return new PoolThreadCache( heapArena, directArena, tinyCacheSize, smallCacheSize, normalCacheSize, DEFAULT_MAX_CACHED_BUFFER_CAPACITY, DEFAULT_CACHE_TRIM_INTERVAL); } // No caching for non FastThreadLocalThreads. return new PoolThreadCache(heapArena, directArena, 0, 0, 0, 0, 0); }
3,4行,分别从headArenas,directArenas中取出一个使用次数最少的PoolArena对象。PoolArena有一个numThreadCaches属性,这个属性是AtomicInteger类型的原子变量。它的作用是在用来记录被PoolThreadCache对象使用的次数。PoolThreadCache对象创建时会在构造方法中会调用它的getAndIncrement方法,释放时在free0方法中调用他的getAndDecrement方法。
6行, 如果运行每个线程都使用缓存(userCacheForAllThreads==true),或者当成线程对象是FastThreadLocalThread时, 在第8行创建一个线程专用的PTC对象。
PoolChunkList(PCKL)
关键属性
PoolChunkList<T> nextList
PoolChunkList<T> prevList
这两个属性表明PCKL对象是一个双向链表的节点。
PoolChunk<T> head
这个属性表明PCKL对象还维护的一个PCK类型的链表,head指向这个链表的头。
int minUsage;
int maxUsage;
int maxCapacity;
minUsage是PCK链表中每个PCK对象内存的最小使用率,maxUseage是PCK的最大使用率。这两个值是百分比,例如:minUsage=10, maxUse=50,表示PCK链表中只能保存使用率在[10%,50%)的PCK对象。 maxCapacity表示PCK最大可分配的内存数,算法是: maxCapacity = (int)(chunkSize * (100L - minUseage) / 100L)。
初始化PCKL链表
PCKL链表有PoolArena负责维护,在PoolArena的构造方法中初始化:
// io.netty.buffer.PoolArena#PoolArena(PooledByteBufAllocator parent, int pageSize, // int maxOrder, int pageShifts, int chunkSize, int cacheAlignment) q100 = new PoolChunkList<T>(this, null, 100, Integer.MAX_VALUE, chunkSize); q075 = new PoolChunkList<T>(this, q100, 75, 100, chunkSize); q050 = new PoolChunkList<T>(this, q075, 50, 100, chunkSize); q025 = new PoolChunkList<T>(this, q050, 25, 75, chunkSize); q000 = new PoolChunkList<T>(this, q025, 1, 50, chunkSize); qInit = new PoolChunkList<T>(this, q000, Integer.MIN_VALUE, 25, chunkSize); q100.prevList(q075); q075.prevList(q050); q050.prevList(q025); q025.prevList(q000); q000.prevList(null); qInit.prevList(qInit);
4-9行,初始化PCKL节点。每个节点的名字q{num},其中num表示这个节点的最小使用率minUsage,如q075节点的minUsage=%75。
11-16行,把PCKL节点组装成一个链表。
使用q(minUsage, maxUsage)表示一个节点,那么:
qInit = q(Integer.MIN_VALUE, 25%)
q000 = q(1%, 50%)
q025 = q(25%, 75%)
q075 = q(75%, 100%)
q100 = q(100%, Integer.MAX_VALUE)
这个链表的结构如下图所示:
PoolChunk(PCK)在PoolChunkList(PCKL)中移动
一个新创建的PCK对象,它的内存使用率是usage=%0,被放进qInit节节点。每次从这个PCK对象中分配内存,都会导致它的使用率增加,当usage>=25%,即大于等于qInit的maxUsage时,会把它移动到q000中。继续从PCK对象中分配内存,它的usage继续增加,当usage大于等于它所属PCKL的maxUsage时,把它移动到PKCL链表中的下一个节点,直到q100为止。下面是内存分配导致PCK移动的代码:
//io.netty.buffer.PoolChunkList#allocate boolean allocate(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) { if (head == null || normCapacity > maxCapacity) { // Either this PoolChunkList is empty or the requested capacity is larger then the capacity which can // be handled by the PoolChunks that are contained in this PoolChunkList. return false; } for (PoolChunk<T> cur = head;;) { long handle = cur.allocate(normCapacity); if (handle < 0) { cur = cur.next; if (cur == null) { return false; } } else { cur.initBuf(buf, handle, reqCapacity); if (cur.usage() >= maxUsage) { remove(cur); nextList.add(cur); } return true; } } }
9-12行,尝试从PCK链表中的所有PCK节点分配所需的内存。
14行,没有找到能分配内存的PCK节点。
17行,从cur节点分配到所需的内存,并初始化PooledByteBuf对象。
18-21行,如cur节点的使用率大于等于当前PCKL节点maxUsage,调用remove方法把cur从head链表中删除,然后调用PCKL链表中的下一个节点的add方法,把cur移动到下一个节点中。
如果持续地释放内存,把内存还给PCK对象,会导致usage持续减小,当usage小于它所属的PCKL的minUsage时,把它移动到PCKL链表中的前一个节点,直到q000位为止。当释放内存导致PCK对象的usage等于%0,会销毁这个PCK对象,释放整个chunk的内存。下面是释放内存导致PCK对象移动的代码:
//io.netty.buffer.PoolChunkList#free boolean free(PoolChunk<T> chunk, long handle) { chunk.free(handle); if (chunk.usage() < minUsage) { remove(chunk); // Move the PoolChunk down the PoolChunkList linked-list. return move0(chunk); } return true; } //io.netty.buffer.PoolChunkList#move0 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); }
第3行,释放内存,把内存返还给PCK对象。
4-7行,如PCK的使用率小于当前PCKL的minUsage,调用remove方法把PCK对象从当前PCKL对象中删除,然后调用move0方法把它移动到前一个PCKL节点。
13-31行,移动PCK到前一个PCKL。
完整的内存分配释放流程
内存分配
入口方法:
io.netty.buffer.AbstractByteBufAllocator#heapBuffer(int, int),创建使用堆内存的ByteBuf, 调用newHeapBuffer方法。
io.netty.buffer.AbstractByteBufAllocator#directBuffer(int, int), 创建使用直接内存的ByteBuf, 调用newDirectBuffer方法。
具体实现:
io.netty.buffer.PooledByteBufAllocator#newHeapBuffer(int initialCapacity, int maxCapacity)。
io.netty.buffer.PooledByteBufAllocator#newDirectBuffer(int initialCapacity, int maxCapacity)。
这两个方法都是从PoolThreadCache对象中得到线程专用的PoolArena对象,然后调用PoolArena的allocate方法创建PoolByteBuf对象。
PoolArena入口方法:
io.netty.buffer.PoolArena#allocate(io.netty.buffer.PoolThreadCache, int, int),这个方法是PoolArena分配内存,创建PoolByteBuf对象的入口方法。它先调用子类实现的newByteBuf创建一个PoolByteBuf对象,这个方法有两个实现:
io.netty.buffer.PoolArena.HeapArena#newByteBuf(int maxCapacity),创建使用堆内存的PooledByteBuf对象。
io.netty.buffer.PoolArena.DirectArena#newByteBuf(int maxCapacity),创建使用直接内存PooledByteBuf对象。
然后调用io.netty.buffer.PoolArena#allocate(io.netty.buffer.PoolThreadCache, io.netty.buffer.PooledByteBuf<T>, int)方法为PoolByteBuf对象分配内存,这个方法是分配内存的核心方法,下面来重点分析一下它的代码:
private void allocate(PoolThreadCache cache, PooledByteBuf<T> buf, final int reqCapacity) { final int normCapacity = normalizeCapacity(reqCapacity); if (isTinyOrSmall(normCapacity)) { // capacity < pageSize int tableIdx; PoolSubpage<T>[] table; boolean tiny = isTiny(normCapacity); if (tiny) { // < 512 if (cache.allocateTiny(this, buf, reqCapacity, normCapacity)) { // was able to allocate out of the cache so move on return; } 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; } tableIdx = smallIdx(normCapacity); table = smallSubpagePools; } final PoolSubpage<T> head = table[tableIdx]; /** * Synchronize on the head. This is needed as {@link PoolChunk#allocateSubpage(int)} and * {@link PoolChunk#free(long)} may modify the doubly linked list as well. */ synchronized (head) { final PoolSubpage<T> s = head.next; if (s != head) { assert s.doNotDestroy && s.elemSize == normCapacity; long handle = s.allocate(); assert handle >= 0; s.chunk.initBufWithSubpage(buf, handle, reqCapacity); incTinySmallAllocation(tiny); return; } } synchronized (this) { allocateNormal(buf, reqCapacity, normCapacity); } incTinySmallAllocation(tiny); return; } if (normCapacity <= chunkSize) { if (cache.allocateNormal(this, buf, reqCapacity, normCapacity)) { // was able to allocate out of the cache so move on return; } synchronized (this) { allocateNormal(buf, reqCapacity, normCapacity); ++allocationsNormal; } } else { // Huge allocations are never served via the cache so just call allocateHuge allocateHuge(buf, reqCapacity); } }
第2行,根据需要的内存大小reqCapacity,计算可以分配的标准内存大小normCapacity。必须满足(1)normCapacity>=reqCapacity, (2)normCapacity是directMemoryCacheAlignment的整数倍,此外,还要根据reqCapacity的大小分3中情况:
reqCapacity>=chunkSize:normCapacity取同时满足(1),(2)的最小值。
reqCapacity>=512且reqCapacity<chunkSize: (3)normCapacity>=512*2k, (4)normCapacity<=chunkSize,normCapacit取同时满足(1),(2),(3),(4)的最小值。
reqCapacity<412: (5)normCapacity<512, (6)normCapacity是16的整数倍,normCapacity取同时满足(1),(2),(5),(6)的最小值。
8-13行,分配Tiny类型的内存(<512)。 8-10行,如果PoolThreadCache缓存对象中分配到内存,分配内流程结束。12-13行,如果缓存中没有,就从Tiny内存池中分配一块内存。
15-20行,分配Small类型的内存(>=512且<pageSize)。和分配Tiny内存的逻辑相同。
29-27行, 使用从前两个步骤中得到的Tiny或Small内存的索引,从子页面池中分配一块内存。33行,从子页面中分配内存。35行,使用分配到的内存初始化PoolByteBuf对象,如果能到这里,分配内存流程结束。
41行,如果子页面池中还没有内存可用,调用allocateNormal方法从PoolChunk对象中分配一个子页面,再从子页面中分配所需的内存。
47-55行,分配Normal类型的内存(>=pageSize且<chunkSize)。48,49行,从缓存中分配内存,如果成功,分配内存流程结束。53行,缓存中没有可用的内存,调用allocateNormal方法从PoolChunk中分配内存。
58行,如果分配的是>chunkSize的内存。这块内存不会进入PCKL链表中。
上面代码中的allocateNormal方法封装了创建PCK对象,从PCK对象中分配内存,再把PCK对象放入到PCKL链表中的逻辑,也是十分重要的代码。
private void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) { 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)) { return; } // Add a new chunk. PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize); long handle = c.allocate(normCapacity); assert handle > 0; c.initBuf(buf, handle, reqCapacity); qInit.add(c); }
2-5行,依次尝试从每个PCKL节点中分配内存,如果成功,分配内存流程结束。
9-13行,先创建一个新的PCK对象,然后从中分配内存,使用内存初始化PooledByteBuf对象,最后把PCK对象添加PCKL链表头节点qInit中。PKCL对象的add方法会和allocate一样,根据PCK对象的内存使用率,把它移动到链表中合适的位置。
内存释放
io.netty.buffer.PooledByteBuf#deallocate方法调用io.netty.buffer.PoolArena#free方法,这个free方法负责整个内存释放过程。
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); if (cache != null && cache.add(this, chunk, handle, normCapacity, sizeClass)) { // cached so not free it. return; } freeChunk(chunk, handle, sizeClass); } }
这段代码重点在8-14行。第8,9行,优先把内存放到缓存中,这样下次就能快速地从缓存中直接取用。第14行,在不能放进缓存的情况下把内存返回给PCK对象。
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(); } destroyChunk = !chunk.parent.free(chunk, handle); } if (destroyChunk) { // destroyChunk not need to be called while holding the synchronized lock. destroyChunk(chunk); } }
第17行,掉用PCKL对象的free方法把内存还给PCK对象,移动PCK对象在PCKL链表中位置。如果此时这个PCK对象的使用率变成0,destroyChunk=true。
第21行,调用destroyChunk方法销毁掉PCK对象。
- netty源码分析系列——PooledByteBuf&PooledByteBufAllocator
- netty源码解解析(4.0)-24 ByteBuf基于内存池的内存管理
- Netty源码分析:PooledByteBufAllocator
- netty源码解解析(4.0)-22 ByteBuf的I/O
- netty源码分析系列——ByteBuf&UnpooledByteBuf
- netty源码解解析(4.0)-21 ByteBuf的设计原理
- Netty bytebuf 源码解析
- netty源码阅读之UnpooledByteBufAllocator
- Netty学习之旅------源码分析Netty线程本地分配机制与PooledByteBuf线程级对象池原理分析
- (五) netty内存池:分配器PooledByteBufAllocator和内存池字节缓冲的表示
- netty源码解解析(4.0)-12 Channel NIO实现:channel初始化
- AbstractReferenceCountedByteBuf源码分析
- Netty学习之旅----ByteBuf源码解读之初探UnpooledHeapByteBuf、UnpooledDirectByteBuf
- netty(十一)源码分析之ByteBuf 二
- netty源码解解析(4.0)-4 线程模型-概览
- netty源码解解析(4.0)-14 Channel NIO实现:读取数据
- netty源码解解析(4.0)-10 ChannelPipleline的默认实现--事件传递及处理
- netty源码解解析(4.0)-19 ChannelHandler: codec--常用编解码实现
- netty源码解解析(4.0)-15 Channel NIO实现:写数据
- netty源码解解析(4.0)-7 线程模型-IO线程EventLoopGroup和NIO实现(二)