andreacavalli
/
netty-incubator-buffer-api
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix more tests 

Fundamental design issues remain, though.
Drops can end up being shared across instances with different memory allocations, and this means we can't currently attach the de-allocation information to the drop instance.
We also cannot use the AllocationControl instance for this because it has the same problem.
This commit is contained in:
Chris Vest 2021-05-12 16:05:09 +02:00
parent 6b62b3a6c7
commit 670cca2d43
6 changed files with 79 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
src/main/java/io/netty/buffer/api/pool/PoolArena.java
View File

@ -220,20 +220,12 @@ class PoolArena extends SizeClasses implements PoolArenaMetric, AllocatorControl
}
void free(PoolChunk chunk, long handle, int normCapacity, PoolThreadCache cache) {
if (chunk.unpooled) {
int size = chunk.chunkSize();
chunk.destroy();
activeBytesHuge.add(-size);
deallocationsHuge.increment();
} else {
SizeClass sizeClass = sizeClass(handle);
if (cache != null && cache.add(this, chunk, handle, normCapacity, sizeClass)) {
// cached so not free it.
return;
}
freeChunk(chunk, handle, normCapacity, sizeClass);
SizeClass sizeClass = sizeClass(handle);
if (cache != null && cache.add(this, chunk, handle, normCapacity, sizeClass)) {
// cached so not free it.
return;
}
freeChunk(chunk, handle, normCapacity, sizeClass);
}
private static SizeClass sizeClass(long handle) {

29
src/main/java/io/netty/buffer/api/pool/PoolChunk.java
View File

@ -16,7 +16,6 @@
package io.netty.buffer.api.pool;
import io.netty.buffer.api.Buffer;
import io.netty.buffer.api.Drop;
import java.util.PriorityQueue;
@ -143,7 +142,6 @@ final class PoolChunk implements PoolChunkMetric {
final PoolArena arena;
final Buffer base; // The buffer that is the source of the memory. Closing it will free the memory.
final Object memory;
final boolean unpooled;
/**
* store the first page and last page of each avail run
@ -171,7 +169,6 @@ final class PoolChunk implements PoolChunkMetric {
PoolChunk next;
PoolChunk(PoolArena arena, Buffer base, Object memory, int pageSize, int pageShifts, int chunkSize, int maxPageIdx) {
unpooled = false;
this.arena = arena;
this.base = base;
this.memory = memory;
@ -520,13 +517,9 @@ final class PoolChunk implements PoolChunkMetric {
int offset = runOffset(handle) << pageShifts;
int maxLength = runSize(pageShifts, handle);
PoolThreadCache poolThreadCache = arena.parent.threadCache();
initAllocatorControl(control, poolThreadCache, handle, maxLength, offset, size);
return arena.manager.recoverMemory(control, memory, offset, size, new Drop<Buffer>() {
@Override
public void drop(Buffer obj) {
arena.free(PoolChunk.this, handle, maxLength, poolThreadCache);
}
});
initAllocatorControl(control, poolThreadCache, handle, maxLength);
return arena.manager.recoverMemory(control, memory, offset, size,
new PooledDrop(control, arena, this, poolThreadCache, handle, maxLength));
} else {
return allocateBufferWithSubpage(handle, size, threadCache, control);
}
@ -542,25 +535,19 @@ final class PoolChunk implements PoolChunkMetric {
assert size <= s.elemSize;
int offset = (runOffset << pageShifts) + bitmapIdx * s.elemSize;
initAllocatorControl(control, threadCache, handle, s.elemSize, offset, size);
return arena.manager.recoverMemory(control, memory, offset, size, new Drop<Buffer>() {
@Override
public void drop(Buffer obj) {
arena.free(PoolChunk.this, handle, s.elemSize, threadCache);
}
});
initAllocatorControl(control, threadCache, handle, s.elemSize);
return arena.manager.recoverMemory(control, memory, offset, size,
new PooledDrop(control, arena, this, threadCache, handle, s.elemSize));
}
private void initAllocatorControl(PooledAllocatorControl control, PoolThreadCache threadCache, long handle,
int normSize, int offset, int size) {
int normSize) {
control.arena = arena;
control.chunk = this;
control.threadCache = threadCache;
control.handle = handle;
control.normSize = normSize;
control.memory = memory;
control.offset = offset;
control.size = size;
control.updates++;
}
@Override

2
src/main/java/io/netty/buffer/api/pool/PoolSubpage.java
View File

@ -108,7 +108,7 @@ final class PoolSubpage implements PoolSubpageMetric {
setNextAvail(bitmapIdx);
if (numAvail ++ == 0) {
if (numAvail++ == 0) {
addToPool(head);
// When maxNumElems == 1, the maximum numAvail is also 1.
// Each of these PoolSubpages will go in here when they do free operation.

4
src/main/java/io/netty/buffer/api/pool/PooledAllocatorControl.java
View File

@ -24,9 +24,7 @@ class PooledAllocatorControl implements AllocatorControl {
public PoolThreadCache threadCache;
public long handle;
public int normSize;
public Object memory;
public int offset;
public int size;
public int updates;
@Override
public Object allocateUntethered(Buffer originator, int size) {

13
src/main/java/io/netty/buffer/api/pool/PooledBufferAllocator.java
View File

@ -45,7 +45,7 @@ public class PooledBufferAllocator implements BufferAllocator, BufferAllocatorMe
private static final int DEFAULT_NUM_DIRECT_ARENA;
private static final int DEFAULT_PAGE_SIZE;
private static final int DEFAULT_MAX_ORDER; // 8192 << 11 = 16 MiB per chunk
private static final int DEFAULT_MAX_ORDER; // 8192 << 9 = 4 MiB per chunk
private static final int DEFAULT_SMALL_CACHE_SIZE;
private static final int DEFAULT_NORMAL_CACHE_SIZE;
static final int DEFAULT_MAX_CACHED_BUFFER_CAPACITY;
@ -75,7 +75,7 @@ public class PooledBufferAllocator implements BufferAllocator, BufferAllocatorMe
DEFAULT_PAGE_SIZE = defaultPageSize;
DEFAULT_DIRECT_MEMORY_CACHE_ALIGNMENT = defaultAlignment;
int defaultMaxOrder = SystemPropertyUtil.getInt("io.netty.allocator.maxOrder", 11);
int defaultMaxOrder = SystemPropertyUtil.getInt("io.netty.allocator.maxOrder", 9);
Throwable maxOrderFallbackCause = null;
try {
validateAndCalculateChunkSize(DEFAULT_PAGE_SIZE, defaultMaxOrder);
@ -284,7 +284,10 @@ public class PooledBufferAllocator implements BufferAllocator, BufferAllocatorMe
@Override
public Buffer allocate(int size) {
return allocate(new PooledAllocatorControl(), size).order(ByteOrder.nativeOrder());
if (size < 1) {
throw new IllegalArgumentException("Allocation size must be positive, but was " + size + '.');
}
return allocate(new PooledAllocatorControl(), size);
}
Buffer allocate(PooledAllocatorControl control, int size) {
@ -292,7 +295,7 @@ public class PooledBufferAllocator implements BufferAllocator, BufferAllocatorMe
PoolArena arena = cache.arena;
if (arena != null) {
return arena.allocate(control, cache, size);
return arena.allocate(control, cache, size).fill((byte) 0).order(ByteOrder.nativeOrder());
}
BufferAllocator unpooled = manager.isNative()? BufferAllocator.direct() : BufferAllocator.heap();
return unpooled.allocate(size);
@ -300,9 +303,11 @@ public class PooledBufferAllocator implements BufferAllocator, BufferAllocatorMe
@Override
public void close() {
trimCurrentThreadCache();
for (PoolArena arena : arenas) {
arena.close();
}
threadCache.remove();
}
/**

55
src/main/java/io/netty/buffer/api/pool/PooledDrop.java Normal file
View File

@ -0,0 +1,55 @@
/*
* Copyright 2021 The Netty Project
*
* The Netty Project licenses this file to you under the Apache License,
* version 2.0 (the "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at:
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations
* under the License.
*/
package io.netty.buffer.api.pool;
import io.netty.buffer.api.Buffer;
import io.netty.buffer.api.Drop;
class PooledDrop implements Drop<Buffer> {
private final PooledAllocatorControl control;
private PoolArena arena;
private PoolChunk chunk;
private PoolThreadCache threadCache;
private long handle;
private int normSize;
PooledDrop(PooledAllocatorControl control, PoolArena arena, PoolChunk chunk, PoolThreadCache threadCache,
long handle, int normSize) {
this.control = control;
this.arena = arena;
this.chunk = chunk;
this.threadCache = threadCache;
this.handle = handle;
this.normSize = normSize;
}
@Override
public void drop(Buffer obj) {
arena.free(chunk, handle, normSize, threadCache);
}
@Override
public void attach(Buffer obj) {
if (control.updates > 0) {
arena = control.arena;
chunk = control.chunk;
threadCache = control.threadCache;
handle = control.handle;
normSize = control.normSize;
control.updates = 0;
}
}
}