ByteBuf分类
Pooled: 每次申请内存都是从曾经调配好的内存池中取
Unpooled: 每次申请内存都重新分配
Heap: 应用jvm的堆内存
Direct: 操作系统间接内存
Unsafe: 调用native办法间接操作内存
非Unsafe: 通过jdk的api间接操作内存
Unsafe形式: 调用jdk的unsafe实例，依据根底偏移量+index算出总偏移量
static byte getByte(byte[] data, int index) {
return UNSAFE.getByte(data, BYTE_ARRAY_BASE_OFFSET + index);
}非Unsafe形式: 就是通过数组下标获取
static byte getByte(byte[] memory, int index) {
return memory[index];
}Unsafe与下面两种分类形式不同，用户无奈决定用哪种形式；是否选用unsafe是由Netty判断能不能拿到底层的unsafe决定的。
if (PlatformDependent.hasUnsafe()) {
buf = UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
} else {
buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
}UnpooledByteBufAllocator调配形式比较简单，就是每次申请内存的时候都重新分配一块新的内存返回。
重点看PooledByteBufAllocator内存调配流程
PooledByteBufAllocator分配内存流程
以堆外内存非配为例
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
// 拿到线程独有的PoolThreadCache
PoolThreadCache cache = threadCache.get();
PoolArena<ByteBuffer> directArena = cache.directArena;
ByteBuf buf;
if (directArena != null) {
buf = directArena.allocate(cache, initialCapacity, maxCapacity);
} else {
if (PlatformDependent.hasUnsafe()) {
buf = UnsafeByteBufUtil.newUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
} else {
buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
}
}
return toLeakAwareBuffer(buf);
}先通过threadCache拿到一个PoolThreadCache类型的属性，threadCache是PoolThreadLocalCache类型的，PoolThreadLocalCache继承自netty自定义的FastThreadLocal，相似jdk的ThreadLocal，每个线程都有本人的PoolThreadLocalCache，这样每个线程也就有本人的PoolThreadCache。
PoolThreadCache次要有两块区域，arena和cache
通过PoolThreadCache分配内存时，会先从cache获取，cache是之前被应用过被回收的内存；
cache获取不到，就从arena开拓一块内存，用完后再回收到cache，不便下次间接从cache获取。
cache局部是由多个MemoryRegionCache类型的数组组织起来的。不同数组下标代表不同的MemoryRegionCache内存规格。
MemoryRegionCache里有一个队列，队列元素Entry蕴含一个chunk和一个handle，chunk是一块内存区域，而后通过handle能够定位到这个Entry它所援用的内存地位。
所以整个PoolThreadCache的cache局部的构造如下:
队列元素Entry代表一块对应大小的可分配内存。
ByteBufAllocator内存调配
在PooledByteBufAllocator构造函数里，会创立两个PoolArena类型的数组，别离用于堆内和堆外
heapArenas = newArenaArray(nHeapArena);
directArenas = newArenaArray(nDirectArena);nHeapArena默认状况下是 2*CPU核数
final int defaultMinNumArena = runtime.availableProcessors() * 2;
这个应该是跟NioEventLoopGroup无关，默认workGroup会创立 2*CPU核数 个NioEventLoop，这样就能保障每个NioEventLoop线程都有一个本人独享的Arena防止线程争用，不必加锁。
接着下面的directArena.allocate()办法
PooledByteBuf<T> allocate(PoolThreadCache cache, int reqCapacity, int maxCapacity) {
// 获取一个PooledByteBuf对象，如果对象池里有间接拿来用，否则就创立
PooledByteBuf<T> buf = newByteBuf(maxCapacity);
// 在PoolThreadCache里分配内存
allocate(cache, buf, reqCapacity);
return buf;
}allocate()办法先尝试从缓存中调配，也就是命中缓存；
如果失败，再从arena里开拓一块内存调配
命中缓存调配流程
- 找到对应size的MemoryRegionCache
- 从队列中弹出一个entry给ByteBuf初始化
- 将弹出的entry扔到对象池复用
在PoolThreadCache里分配内存时，都是先尝试从缓存上进行调配，如果缓存上没有适合的，就在arena开拓一块内存。
内存规格
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];
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);
if (tiny) {
allocationsTiny.increment();
} else {
allocationsSmall.increment();
}
return;
}
}
allocateNormal(buf, reqCapacity, normCapacity);
return;
}
if (normCapacity <= chunkSize) {
if (cache.allocateNormal(this, buf, reqCapacity, normCapacity)) {
// was able to allocate out of the cache so move on
return;
}
allocateNormal(buf, reqCapacity, normCapacity);
} else {
// Huge allocations are never served via the cache so just call allocateHuge
allocateHuge(buf, reqCapacity);
}
}以tiny类型的内存调配为例
boolean allocateTiny(PoolArena<?> area, PooledByteBuf<?> buf, int reqCapacity, int normCapacity) {
return allocate(cacheForTiny(area, normCapacity), buf, reqCapacity);
}
private MemoryRegionCache<?> cacheForTiny(PoolArena<?> area, int normCapacity) {
int idx = PoolArena.tinyIdx(normCapacity);
if (area.isDirect()) {
return cache(tinySubPageDirectCaches, idx);
}
return cache(tinySubPageHeapCaches, idx);
}
// tinyIdx办法间接把 normCapacity/16
static int tinyIdx(int normCapacity) {
return normCapacity >>> 4;
}
private static <T> MemoryRegionCache<T> cache(MemoryRegionCache<T>[] cache, int idx) {
if (cache == null || idx > cache.length - 1) {
return null;
}
// 拿到MemoryRegionCache节点
return cache[idx];
}
private boolean allocate(MemoryRegionCache<?> cache, PooledByteBuf buf, int reqCapacity) {
if (cache == null) {
// no cache found so just return false here
return false;
}
boolean allocated = cache.allocate(buf, reqCapacity);
if (++ allocations >= freeSweepAllocationThreshold) {
allocations = 0;
trim();
}
return allocated;
}
public final boolean allocate(PooledByteBuf<T> buf, int reqCapacity) {
// 找到对应size的MemoryRegionCache后，用它的queue弹出一个entry
Entry<T> entry = queue.poll();
if (entry == null) {
return false;
}
// 初始化
initBuf(entry.chunk, entry.handle, buf, reqCapacity);
// 回收entry对象
entry.recycle();
// allocations is not thread-safe which is fine as this is only called from the same thread all time.
++ allocations;
return true;
}
protected void initBuf(PoolChunk<T> chunk, long handle, PooledByteBuf<T> buf, int reqCapacity) {
chunk.initBufWithSubpage(buf, handle, reqCapacity);
}
private void initBufWithSubpage(PooledByteBuf<T> buf, long handle, int bitmapIdx, int reqCapacity) {
assert bitmapIdx != 0;
int memoryMapIdx = memoryMapIdx(handle);
PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
assert subpage.doNotDestroy;
assert reqCapacity <= subpage.elemSize;
buf.init(
this, handle,
runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize, reqCapacity, subpage.elemSize,
arena.parent.threadCache());
}
void init(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
assert handle >= 0;
assert chunk != null;
this.chunk = chunk;
this.handle = handle;
memory = chunk.memory;
this.offset = offset;
this.length = length;
this.maxLength = maxLength;
tmpNioBuf = null;
this.cache = cache;
}
void recycle() {
chunk = null;
handle = -1;
recyclerHandle.recycle(this);
}
public void recycle(Object object) {
if (object != value) {
throw new IllegalArgumentException("object does not belong to handle");
}
stack.push(this);
}未命中缓存
arena数据结构
依照内存使用率，把雷同内存使用率的chunk归为一个chunkList，而后chunkList之间双向连贯。
每个chunk的大小是16M，分配内存不可能每次申请都调配16M，所以chunk又能够划分为不同大小的subPage。
page级别内存调配
private synchronized void allocateNormal(PooledByteBuf<T> buf, int reqCapacity, int normCapacity) {
// 尝试在chunkList链表上进行调配
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;
}
// 以后没有适合的chunk，就创立一个chunk
PoolChunk<T> c = newChunk(pageSize, maxOrder, pageShifts, chunkSize);
// 通过chunk调配指定大小的一块内存，返回一个handle
// handle指向chunk内调配的一块间断内存
long handle = c.allocate(normCapacity);
++allocationsNormal;
assert handle > 0;
// 拿到内存后，初始化PoolByteBuf
c.initBuf(buf, handle, reqCapacity);
qInit.add(c);
}- 现有chunk上调配
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;
}
// 遍历chunkList，尝试调配
for (PoolChunk<T> cur = head;;) {
long handle = cur.allocate(normCapacity);
// handle<0阐明调配失败，持续遍历
if (handle < 0) {
cur = cur.next;
if (cur == null) {
return false;
}
} else {
// 调配胜利，初始化byteBuf
cur.initBuf(buf, handle, reqCapacity);
// 查看以后chunk调配后的使用率是否还满足原来的使用率范畴
// 不满足就将它转移到别的chunkList
if (cur.usage() >= maxUsage) {
remove(cur);
nextList.add(cur);
}
return true;
}
}
}- 创立一个chunk调配
protected PoolChunk<ByteBuffer> newChunk(int pageSize, int maxOrder, int pageShifts, int chunkSize) {
return new PoolChunk<ByteBuffer>(
this, allocateDirect(chunkSize),
pageSize, maxOrder, pageShifts, chunkSize);
}
// 调用JDK API，分配内存
private static ByteBuffer allocateDirect(int capacity) {
return PlatformDependent.useDirectBufferNoCleaner() ?
PlatformDependent.allocateDirectNoCleaner(capacity) : ByteBuffer.allocateDirect(capacity);
}netty用一个齐全二叉树示意一个chunk被调配的状况。
示意形式有点像线段树，树的每个节点代表一段区间，通过这个节点来示意这段区间有没有被应用。
叶子节点(一个page)示意的区间大小为8k，一个chunk也就是根节点的大小是16M，所以树的高度为11。
节点示意相应的区间是否被调配
PoolChunk(PoolArena<T> arena, T memory, int pageSize, int maxOrder, int pageShifts, int chunkSize) {
unpooled = false;
this.arena = arena;
this.memory = memory;
this.pageSize = pageSize;
this.pageShifts = pageShifts;
this.maxOrder = maxOrder;
this.chunkSize = chunkSize;
unusable = (byte) (maxOrder + 1);
log2ChunkSize = log2(chunkSize);
subpageOverflowMask = ~(pageSize - 1);
freeBytes = chunkSize;
assert maxOrder < 30 : "maxOrder should be < 30, but is: " + maxOrder;
maxSubpageAllocs = 1 << maxOrder;
// Generate the memory map.
// memoryMap、depthMap示意第几个节点在树的第几层
memoryMap = new byte[maxSubpageAllocs << 1];
depthMap = new byte[memoryMap.length];
int memoryMapIndex = 1;
// 遍历树的每一层
for (int d = 0; d <= maxOrder; ++ d) { // move down the tree one level at a time
int depth = 1 << d;
// 遍历这一层的每个节点
for (int p = 0; p < depth; ++ p) {
// in each level traverse left to right and set value to the depth of subtree
memoryMap[memoryMapIndex] = (byte) d;
depthMap[memoryMapIndex] = (byte) d;
memoryMapIndex ++;
}
}
subpages = newSubpageArray(maxSubpageAllocs);
}long allocate(int normCapacity) {
if ((normCapacity & subpageOverflowMask) != 0) { // >= pageSize
return allocateRun(normCapacity);
} else {
return allocateSubpage(normCapacity);
}
}
private long allocateRun(int normCapacity) {
// 算出要调配的内存在树的第几层
int d = maxOrder - (log2(normCapacity) - pageShifts);
int id = allocateNode(d);
if (id < 0) {
return id;
}
freeBytes -= runLength(id);
return id;
}
// 在算出的层数上进行内存调配
private int allocateNode(int d) {
int id = 1;
int initial = - (1 << d); // has last d bits = 0 and rest all = 1
byte val = value(id);
if (val > d) { // unusable
return -1;
}
while (val < d || (id & initial) == 0) { // id & initial == 1 << d for all ids at depth d, for < d it is 0
id <<= 1;
val = value(id);
if (val > d) {
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
// 逐层向上标记它的父节点已被应用
updateParentsAlloc(id);
return id;
}初始化byteBuf
void initBuf(PooledByteBuf<T> buf, long handle, int reqCapacity) {
int memoryMapIdx = memoryMapIdx(handle);
int bitmapIdx = bitmapIdx(handle);
if (bitmapIdx == 0) {
byte val = value(memoryMapIdx);
assert val == unusable : String.valueOf(val);
buf.init(this, handle, runOffset(memoryMapIdx), reqCapacity, runLength(memoryMapIdx),
arena.parent.threadCache());
} else {
initBufWithSubpage(buf, handle, bitmapIdx, reqCapacity);
}
}void init(PoolChunk<T> chunk, long handle, int offset, int length, int maxLength, PoolThreadCache cache) {
assert handle >= 0;
assert chunk != null;
this.chunk = chunk;
this.handle = handle;
memory = chunk.memory;
this.offset = offset;
this.length = length;
this.maxLength = maxLength;
tmpNioBuf = null;
this.cache = cache;
}subPage级别的内存调配
- 定位到一个subPage
- 初始化subPage(在chunk中找到一个page，对这个page依照自定义的subPage大小进行划分)
- 初始化ByteBuf
// tableIdx是间接将申请的内存大小除以16
tableIdx = tinyIdx(normCapacity);
// 数组，不同下标示意不同大小的subPage
table = tinySubpagePools;
// 拿到对应大小的的subPage的链表头结点
final PoolSubpage<T> head = table[tableIdx];tinySubpagePools构造和MemoryRegionCache的tiny数组相似，不同下标代表不同大小的内存。
先通过allocateNormal调配
// 与page级别不同的是，这里会走到上面的allocateSubpage
long allocate(int normCapacity) {
if ((normCapacity & subpageOverflowMask) != 0) { // >= pageSize
return allocateRun(normCapacity);
} else {
return allocateSubpage(normCapacity);
}
}
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.
PoolSubpage<T> head = arena.findSubpagePoolHead(normCapacity);
synchronized (head) {
// d 间接赋值为11，在树的最初一层调配(因为最初一次层的大小是一个page，大于要调配的大小)
int d = maxOrder; // subpages are only be allocated from pages i.e., leaves
int id = allocateNode(d);
if (id < 0) {
return id;
}
final PoolSubpage<T>[] subpages = this.subpages;
final int pageSize = this.pageSize;
freeBytes -= pageSize;
// subpageIdx: 这个subPage在page中的地位
int subpageIdx = subpageIdx(id);
PoolSubpage<T> subpage = subpages[subpageIdx];
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();
}
}
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 ++;
}
// bitmap: 示意page中那个subPage已被调配
for (int i = 0; i < bitmapLength; i ++) {
bitmap[i] = 0;
}
}
addToPool(head);
}
// 将创立好的subPage加到tinySubpagePools的链表中
private void addToPool(PoolSubpage<T> head) {
assert prev == null && next == null;
prev = head;
next = head.next;
next.prev = this;
head.next = this;
}long allocate() {
if (elemSize == 0) {
return toHandle(0);
}
if (numAvail == 0 || !doNotDestroy) {
return -1;
}
final int bitmapIdx = getNextAvail();
int q = bitmapIdx >>> 6;
int r = bitmapIdx & 63;
assert (bitmap[q] >>> r & 1) == 0;
bitmap[q] |= 1L << r;
if (-- numAvail == 0) {
removeFromPool();
}
return toHandle(bitmapIdx);
}
private long toHandle(int bitmapIdx) {
return 0x4000000000000000L | (long) bitmapIdx << 32 | memoryMapIdx;
}private void initBufWithSubpage(PooledByteBuf<T> buf, long handle, int bitmapIdx, int reqCapacity) {
assert bitmapIdx != 0;
int memoryMapIdx = memoryMapIdx(handle);
PoolSubpage<T> subpage = subpages[subpageIdx(memoryMapIdx)];
assert subpage.doNotDestroy;
assert reqCapacity <= subpage.elemSize;
buf.init(
this, handle,
runOffset(memoryMapIdx) + (bitmapIdx & 0x3FFFFFFF) * subpage.elemSize, reqCapacity, subpage.elemSize,
arena.parent.threadCache());
}ByteBuf的开释
- 将要开释的内存加到缓存
- 如果退出缓存失败(缓存已满)，就标记该内存未应用
- 将ByteBuf放到对象池
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)) {
// 援用计数缩小到decrement
if (refCnt == decrement) {
deallocate();
return true;
}
return false;
}
}
}
protected final void deallocate() {
if (handle >= 0) {
final long handle = this.handle;
this.handle = -1;
memory = null;
chunk.arena.free(chunk, handle, maxLength, cache);
recycle();
}
}
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);
}
}
boolean add(PoolArena<?> area, PoolChunk chunk, long handle, int normCapacity, SizeClass sizeClass) {
MemoryRegionCache<?> cache = cache(area, normCapacity, sizeClass);
if (cache == null) {
return false;
}
return cache.add(chunk, handle);
}
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;
}增加缓存失败
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);
}
}
// 标记内存为未应用
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) {
if (subpage.free(head, bitmapIdx & 0x3FFFFFFF)) {
return;
}
}
}
freeBytes += runLength(memoryMapIdx);
setValue(memoryMapIdx, depth(memoryMapIdx));
updateParentsFree(memoryMapIdx);
}将ByteBuf退出到对象池
private void recycle() {
recyclerHandle.recycle(this);
}
发表回复