Buffers always have a cleaner attached

Motivation:
Although having a cleaner attached adds a bit of overhead when allocating or closing buffers,
it is more important to make our systems, libraries and frameworks misuse resistant and safe by default.

Modification:
Remove the ability to allocate a buffer that does not have a cleaner attached.
Reference counting and the ability to explicitly release memory remains.
This just makes sure that we always have a safety net to fall back on.

Result:
This will make systems less prone to crashes through running out of memory, native or otherwise, even in the face of true memory leaks.
(Leaks through retained strong references cannot be fixed in any way)

src/main/java/io/netty/buffer/api/Allocator.java

@@ -150,14 +150,10 @@ public interface Allocator extends AutoCloseable {
     }
 
     static Allocator heap() {
-        return new ManagedAllocator(MemoryManager.getHeapMemoryManager(), null);
+        return new ManagedAllocator(MemoryManager.getHeapMemoryManager(), Statics.CLEANER);
     }
 
     static Allocator direct() {
-        return new ManagedAllocator(MemoryManager.getNativeMemoryManager(), null);
-    }
-
-    static Allocator directWithCleaner() {
         return new ManagedAllocator(MemoryManager.getNativeMemoryManager(), Statics.CLEANER);
     }
 
@@ -168,13 +164,4 @@ public interface Allocator extends AutoCloseable {
     static Allocator pooledDirect() {
         return new SizeClassedMemoryPool(MemoryManager.getNativeMemoryManager());
     }
-
-    static Allocator pooledDirectWithCleaner() {
-        return new SizeClassedMemoryPool(MemoryManager.getNativeMemoryManager()) {
-            @Override
-            protected Drop<Buf> getDrop() {
-                return new NativeMemoryCleanerDrop(this, getMemoryManager(), super.getDrop());
-            }
-        };
-    }
 }

src/main/java/io/netty/buffer/api/CleanerPooledDrop.java

@@ -24,17 +24,17 @@ import static io.netty.buffer.api.Statics.CLEANER;
 import static io.netty.buffer.api.Statics.findVarHandle;
 import static java.lang.invoke.MethodHandles.lookup;
 
-class NativeMemoryCleanerDrop implements Drop<Buf> {
+class CleanerPooledDrop implements Drop<Buf> {
     private static final VarHandle CLEANABLE =
-            findVarHandle(lookup(), NativeMemoryCleanerDrop.class, "cleanable", GatedCleanable.class);
+            findVarHandle(lookup(), CleanerPooledDrop.class, "cleanable", GatedCleanable.class);
     private final SizeClassedMemoryPool pool;
     private final MemoryManager manager;
     private final Drop<Buf> delegate;
     @SuppressWarnings("unused")
     private volatile GatedCleanable cleanable;
 
-    NativeMemoryCleanerDrop(SizeClassedMemoryPool pool, MemoryManager manager,
-                            Drop<Buf> delegate) {
+    CleanerPooledDrop(SizeClassedMemoryPool pool, MemoryManager manager,
+                      Drop<Buf> delegate) {
         this.pool = pool;
         this.manager = manager;
         this.delegate = delegate;

src/main/java/io/netty/buffer/api/SizeClassedMemoryPool.java

@@ -59,7 +59,7 @@ class SizeClassedMemoryPool implements Allocator, AllocatorControl, Drop<Buf> {
     }
 
     protected Drop<Buf> getDrop() {
-        return this;
+        return new CleanerPooledDrop(this, getMemoryManager(), this);
     }
 
     @Override

src/test/java/io/netty/buffer/api/BufTest.java

@@ -68,11 +68,9 @@ public class BufTest {
     static List<Fixture> initialAllocators() {
         return List.of(
                 new Fixture("heap", Allocator::heap, HEAP),
-                new Fixture("direct", Allocator::direct, DIRECT),
-                new Fixture("directWithCleaner", Allocator::directWithCleaner, DIRECT, CLEANER),
+                new Fixture("direct", Allocator::direct, DIRECT, CLEANER),
                 new Fixture("pooledHeap", Allocator::pooledHeap, POOLED, HEAP),
-                new Fixture("pooledDirect", Allocator::pooledDirect, POOLED, DIRECT),
-                new Fixture("pooledDirectWithCleaner", Allocator::pooledDirectWithCleaner, POOLED, DIRECT, CLEANER));
+                new Fixture("pooledDirect", Allocator::pooledDirect, POOLED, DIRECT, CLEANER));
     }
 
     static Stream<Fixture> nonSliceAllocators() {
@@ -87,18 +85,10 @@ public class BufTest {
         return fixtureCombinations().filter(Fixture::isDirect);
     }
 
-    static Stream<Fixture> directWithCleanerAllocators() {
-        return fixtureCombinations().filter(f -> f.isDirect() && f.isCleaner());
-    }
-
-    static Stream<Fixture> directPooledWithCleanerAllocators() {
+    static Stream<Fixture> directPooledAllocators() {
         return fixtureCombinations().filter(f -> f.isDirect() && f.isCleaner() && f.isPooled());
     }
 
-    static Stream<Fixture> poolingAllocators() {
-        return fixtureCombinations().filter(f -> f.isPooled());
-    }
-
     private static Stream<Fixture> fixtureCombinations() {
         Fixture[] fxs = fixtures;
         if (fxs != null) {
@@ -1496,7 +1486,7 @@ public class BufTest {
     class CleanerTests {
         @Disabled("Precise native memory accounting does not work since recent panama-foreign changes.")
         @ParameterizedTest
-        @MethodSource("io.netty.buffer.api.BufTest#directWithCleanerAllocators")
+        @MethodSource("io.netty.buffer.api.BufTest#directAllocators")
         public void bufferMustBeClosedByCleaner(Fixture fixture) throws InterruptedException {
             var allocator = fixture.createAllocator();
             allocator.close();
@@ -1518,7 +1508,7 @@ public class BufTest {
 
         @Disabled("Precise native memory accounting does not work since recent panama-foreign changes.")
         @ParameterizedTest
-        @MethodSource("io.netty.buffer.api.BufTest#directPooledWithCleanerAllocators")
+        @MethodSource("io.netty.buffer.api.BufTest#directPooledAllocators")
         public void buffersMustBeReusedByPoolingAllocatorEvenWhenDroppedByCleanerInsteadOfExplicitly(Fixture fixture)
                 throws InterruptedException {
             try (var allocator = fixture.createAllocator()) {

src/test/java/io/netty/buffer/api/benchmarks/MemorySegmentClosedByCleanerBenchmark.java

@@ -42,9 +42,7 @@ import static java.util.concurrent.CompletableFuture.completedFuture;
 @State(Scope.Benchmark)
 public class MemorySegmentClosedByCleanerBenchmark {
     private static final Allocator direct = Allocator.direct();
-    private static final Allocator withCleaner = Allocator.directWithCleaner();
     private static final Allocator directPooled = Allocator.pooledDirect();
-    private static final Allocator pooledWithCleaner = Allocator.pooledDirectWithCleaner();
 
     @Param({"heavy", "light"})
     public String workload;
@@ -77,19 +75,12 @@ public class MemorySegmentClosedByCleanerBenchmark {
 
     @Benchmark
     public Buf cleanerClose() throws Exception {
-        return process(withCleaner.allocate(256));
+        return process(direct.allocate(256));
     }
 
     @Benchmark
     public Buf cleanerClosePooled() throws Exception {
-        return process(pooledWithCleaner.allocate(256));
-    }
-
-    @Benchmark
-    public Buf pooledWithCleanerExplicitClose() throws Exception {
-        try (Buf buf = pooledWithCleaner.allocate(256)) {
-            return process(buf);
-        }
+        return process(directPooled.allocate(256));
     }
 
     private Buf process(Buf buffer) throws Exception {

src/test/java/io/netty/buffer/api/examples/AsyncExample.java

@@ -22,7 +22,7 @@ import static java.util.concurrent.CompletableFuture.completedFuture;
 
 public final class AsyncExample {
     public static void main(String[] args) throws Exception {
-        try (Allocator allocator = Allocator.pooledDirectWithCleaner();
+        try (Allocator allocator = Allocator.pooledDirect();
              Buf startBuf = allocator.allocate(16)) {
             startBuf.writeLong(threadId());