Repository: netty/netty
Issue: Add support getting the pooled memory size for each PooledByteBufAllocator instance

### Motivation
Our server has its own implementation to decide whether to provide service for users according to the real-time total free memory of the heap memory and the direct memory (mainly the memory used by `io.netty.buffer.PooledByteBufAllocator` for our server). If our server detects it's running out of physical memory, it will just respond with the client requests with something like `Service Unavailable`.

But the problem is: We cannot know the free memory of PooledByteBufAllocator efficiently. For example, we have allocated and cached 8GB chunks in a `PooledByteBufAllocator`, but there is no property to know: what exactly the free memory is, 1GB or 8GB?

### Solution
The [PR](https://github.com/netty/netty/pull/5133/files) in 2016 is similar to what we want and it provides a new property `pooledBytes` to record the free memory of the arena but we want the free memory in the scope of each allocator instance.

And `io.netty.maxDirectMemory` is similar but isn't what we want because it applies the limit to all buffers while we just want to know the free memory size of a PooledByteBufAllocator instance

Issue comments:
For the heap _in general_, the free memory on the heap is not the same as the amount of memory that could be freed if the GC was to do a run, so we can only realistically do this for direct memory where we have precise accounting of what memory is in use and what is not. But then there's also fragmentation; having X free bytes doesn't necessarily mean there's room for an X-sized allocation.

But to count the free bytes, you can iterate the arenas (`pool.arenaMetrics()`), and then iterate the chunk lists for each (`arena.chunkLists`) and then for each chunk in those you can get the `freeBytes()`. Not super convenient, but if you are up for it you could make a PR to improve that.

Pull request review comments:

Patch that resolved the issue:
diff --git a/buffer/src/main/java/io/netty/buffer/PoolArena.java b/buffer/src/main/java/io/netty/buffer/PoolArena.java
index fba1127b5d7..beff6e115bd 100644
--- a/buffer/src/main/java/io/netty/buffer/PoolArena.java
+++ b/buffer/src/main/java/io/netty/buffer/PoolArena.java
@@ -455,6 +455,22 @@ public long numActiveBytes() {
         return max(0, val);
     }
 
+    /**
+     * Return the number of bytes that are currently pinned to buffer instances, by the arena. The pinned memory is not
+     * accessible for use by any other allocation, until the buffers using have all been released.
+     */
+    public long numPinnedBytes() {
+        long val = activeBytesHuge.value(); // Huge chunks are exact-sized for the buffers they were allocated to.
+        synchronized (this) {
+            for (int i = 0; i < chunkListMetrics.size(); i++) {
+                for (PoolChunkMetric m: chunkListMetrics.get(i)) {
+                    val += ((PoolChunk<?>) m).pinnedBytes();
+                }
+            }
+        }
+        return max(0, val);
+    }
+
     protected abstract PoolChunk<T> newChunk(int pageSize, int maxPageIdx, int pageShifts, int chunkSize);
     protected abstract PoolChunk<T> newUnpooledChunk(int capacity);
     protected abstract PooledByteBuf<T> newByteBuf(int maxCapacity);
diff --git a/buffer/src/main/java/io/netty/buffer/PoolChunk.java b/buffer/src/main/java/io/netty/buffer/PoolChunk.java
index b9a1b4cb49b..5cf495b974b 100644
--- a/buffer/src/main/java/io/netty/buffer/PoolChunk.java
+++ b/buffer/src/main/java/io/netty/buffer/PoolChunk.java
@@ -172,6 +172,7 @@ final class PoolChunk<T> implements PoolChunkMetric {
     private final Deque<ByteBuffer> cachedNioBuffers;
 
     int freeBytes;
+    int pinnedBytes;
 
     PoolChunkList<T> parent;
     PoolChunk<T> prev;
@@ -339,7 +340,9 @@ private long allocateRun(int runSize) {
                 handle = splitLargeRun(handle, pages);
             }
 
-            freeBytes -= runSize(pageShifts, handle);
+            int pinnedSize = runSize(pageShifts, handle);
+            freeBytes -= pinnedSize;
+            pinnedBytes += pinnedSize;
             return handle;
         }
     }
@@ -448,6 +451,8 @@ private long allocateSubpage(int sizeIdx) {
      * @param handle handle to free
      */
     void free(long handle, int normCapacity, ByteBuffer nioBuffer) {
+        int runSize = runSize(pageShifts, handle);
+        pinnedBytes -= runSize;
         if (isSubpage(handle)) {
             int sizeIdx = arena.size2SizeIdx(normCapacity);
             PoolSubpage<T> head = arena.findSubpagePoolHead(sizeIdx);
@@ -470,8 +475,6 @@ void free(long handle, int normCapacity, ByteBuffer nioBuffer) {
         }
 
         //start free run
-        int pages = runPages(handle);
-
         synchronized (runsAvail) {
             // collapse continuous runs, successfully collapsed runs
             // will be removed from runsAvail and runsAvailMap
@@ -483,7 +486,7 @@ void free(long handle, int normCapacity, ByteBuffer nioBuffer) {
             finalRun &= ~(1L << IS_SUBPAGE_SHIFT);
 
             insertAvailRun(runOffset(finalRun), runPages(finalRun), finalRun);
-            freeBytes += pages << pageShifts;
+            freeBytes += runSize;
         }
 
         if (nioBuffer != null && cachedNioBuffers != null &&
@@ -585,6 +588,12 @@ public int freeBytes() {
         }
     }
 
+    public int pinnedBytes() {
+        synchronized (arena) {
+            return pinnedBytes;
+        }
+    }
+
     @Override
     public String toString() {
         final int freeBytes;
diff --git a/buffer/src/main/java/io/netty/buffer/PooledByteBufAllocator.java b/buffer/src/main/java/io/netty/buffer/PooledByteBufAllocator.java
index dd9e36f55b7..8fd7b1fd0a7 100644
--- a/buffer/src/main/java/io/netty/buffer/PooledByteBufAllocator.java
+++ b/buffer/src/main/java/io/netty/buffer/PooledByteBufAllocator.java
@@ -677,6 +677,40 @@ private static long usedMemory(PoolArena<?>[] arenas) {
         return used;
     }
 
+    /**
+     * Returns the number of bytes of heap memory that is currently pinned to heap buffers allocated by a
+     * {@link ByteBufAllocator}, or {@code -1} if unknown.
+     * A buffer can pin more memory than its {@linkplain ByteBuf#capacity() capacity} might indicate,
+     * due to implementation details of the allocator.
+     */
+    public final long pinnedHeapMemory() {
+        return pinnedMemory(heapArenas);
+    }
+
+    /**
+     * Returns the number of bytes of direct memory that is currently pinned to direct buffers allocated by a
+     * {@link ByteBufAllocator}, or {@code -1} if unknown.
+     * A buffer can pin more memory than its {@linkplain ByteBuf#capacity() capacity} might indicate,
+     * due to implementation details of the allocator.
+     */
+    public final long pinnedDirectMemory() {
+        return pinnedMemory(directArenas);
+    }
+
+    private static long pinnedMemory(PoolArena<?>[] arenas) {
+        if (arenas == null) {
+            return -1;
+        }
+        long used = 0;
+        for (PoolArena<?> arena : arenas) {
+            used += arena.numPinnedBytes();
+            if (used < 0) {
+                return Long.MAX_VALUE;
+            }
+        }
+        return used;
+    }
+
     final PoolThreadCache threadCache() {
         PoolThreadCache cache =  threadCache.get();
         assert cache != null;


Test patch:
diff --git a/buffer/src/test/java/io/netty/buffer/AbstractByteBufAllocatorTest.java b/buffer/src/test/java/io/netty/buffer/AbstractByteBufAllocatorTest.java
index e2f3ab72ef3..e6e064b40cc 100644
--- a/buffer/src/test/java/io/netty/buffer/AbstractByteBufAllocatorTest.java
+++ b/buffer/src/test/java/io/netty/buffer/AbstractByteBufAllocatorTest.java
@@ -95,7 +95,6 @@ protected static void assertInstanceOf(ByteBuf buffer, Class<? extends ByteBuf>
         assertTrue(clazz.isInstance(buffer instanceof SimpleLeakAwareByteBuf ? buffer.unwrap() : buffer));
     }
 
-    @SuppressWarnings("unchecked")
     @Test
     public void testUsedDirectMemory() {
         T allocator =  newAllocator(true);
@@ -114,7 +113,6 @@ public void testUsedDirectMemory() {
         assertEquals(expectedUsedMemoryAfterRelease(allocator, capacity), metric.usedDirectMemory());
     }
 
-    @SuppressWarnings("unchecked")
     @Test
     public void testUsedHeapMemory() {
         T allocator =  newAllocator(true);
@@ -141,4 +139,7 @@ protected long expectedUsedMemory(T allocator, int capacity) {
     protected long expectedUsedMemoryAfterRelease(T allocator, int capacity) {
         return 0;
     }
+
+    protected void trimCaches(T allocator) {
+    }
 }
diff --git a/buffer/src/test/java/io/netty/buffer/PooledByteBufAllocatorTest.java b/buffer/src/test/java/io/netty/buffer/PooledByteBufAllocatorTest.java
index ef3a73d7789..16cb9927b9e 100644
--- a/buffer/src/test/java/io/netty/buffer/PooledByteBufAllocatorTest.java
+++ b/buffer/src/test/java/io/netty/buffer/PooledByteBufAllocatorTest.java
@@ -27,6 +27,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Queue;
+import java.util.Random;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -37,6 +38,7 @@
 import static io.netty.buffer.PoolChunk.runOffset;
 import static io.netty.buffer.PoolChunk.runPages;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -67,6 +69,11 @@ protected long expectedUsedMemoryAfterRelease(PooledByteBufAllocator allocator,
         return allocator.metric().chunkSize();
     }
 
+    @Override
+    protected void trimCaches(PooledByteBufAllocator allocator) {
+        allocator.trimCurrentThreadCache();
+    }
+
     @Test
     public void testTrim() {
         PooledByteBufAllocator allocator = newAllocator(true);
@@ -689,4 +696,129 @@ public void testNormalPoolSubpageRelease() {
 
         assertTrue(beforeFreeBytes < afterFreeBytes);
     }
+
+    @Override
+    @Test
+    public void testUsedDirectMemory() {
+        for (int power = 0; power < 8; power++) {
+            int initialCapacity = 1024 << power;
+            testUsedDirectMemory(initialCapacity);
+        }
+    }
+
+    private void testUsedDirectMemory(int initialCapacity) {
+        PooledByteBufAllocator allocator = newAllocator(true);
+        ByteBufAllocatorMetric metric = allocator.metric();
+        assertEquals(0, metric.usedDirectMemory());
+        assertEquals(0, allocator.pinnedDirectMemory());
+        ByteBuf buffer = allocator.directBuffer(initialCapacity, 4 * initialCapacity);
+        int capacity = buffer.capacity();
+        assertEquals(expectedUsedMemory(allocator, capacity), metric.usedDirectMemory());
+        assertThat(allocator.pinnedDirectMemory())
+                .isGreaterThanOrEqualTo(capacity)
+                .isLessThanOrEqualTo(metric.usedDirectMemory());
+
+        // Double the size of the buffer
+        buffer.capacity(capacity << 1);
+        capacity = buffer.capacity();
+        assertEquals(expectedUsedMemory(allocator, capacity), metric.usedDirectMemory(), buffer.toString());
+        assertThat(allocator.pinnedDirectMemory())
+                .isGreaterThanOrEqualTo(capacity)
+                .isLessThanOrEqualTo(metric.usedDirectMemory());
+
+        buffer.release();
+        assertEquals(expectedUsedMemoryAfterRelease(allocator, capacity), metric.usedDirectMemory());
+        assertThat(allocator.pinnedDirectMemory())
+                .isGreaterThanOrEqualTo(0)
+                .isLessThanOrEqualTo(metric.usedDirectMemory());
+        trimCaches(allocator);
+        assertEquals(0, allocator.pinnedDirectMemory());
+
+        int[] capacities = new int[30];
+        Random rng = new Random();
+        for (int i = 0; i < capacities.length; i++) {
+            capacities[i] = initialCapacity / 4 + rng.nextInt(8 * initialCapacity);
+        }
+        ByteBuf[] bufs = new ByteBuf[capacities.length];
+        for (int i = 0; i < 20; i++) {
+            bufs[i] = allocator.directBuffer(capacities[i], 2 * capacities[i]);
+        }
+        for (int i = 0; i < 10; i++) {
+            bufs[i].release();
+        }
+        for (int i = 20; i < 30; i++) {
+            bufs[i] = allocator.directBuffer(capacities[i], 2 * capacities[i]);
+        }
+        for (int i = 0; i < 10; i++) {
+            bufs[i] = allocator.directBuffer(capacities[i], 2 * capacities[i]);
+        }
+        for (int i = 0; i < 30; i++) {
+            bufs[i].release();
+        }
+        trimCaches(allocator);
+        assertEquals(0, allocator.pinnedDirectMemory());
+    }
+
+    @Override
+    @Test
+    public void testUsedHeapMemory() {
+        for (int power = 0; power < 8; power++) {
+            int initialCapacity = 1024 << power;
+            testUsedHeapMemory(initialCapacity);
+        }
+    }
+
+    private void testUsedHeapMemory(int initialCapacity) {
+        PooledByteBufAllocator allocator = newAllocator(true);
+        ByteBufAllocatorMetric metric = allocator.metric();
+
+        assertEquals(0, metric.usedHeapMemory());
+        assertEquals(0, allocator.pinnedDirectMemory());
+        ByteBuf buffer = allocator.heapBuffer(initialCapacity, 4 * initialCapacity);
+        int capacity = buffer.capacity();
+        assertEquals(expectedUsedMemory(allocator, capacity), metric.usedHeapMemory());
+        assertThat(allocator.pinnedHeapMemory())
+                .isGreaterThanOrEqualTo(capacity)
+                .isLessThanOrEqualTo(metric.usedHeapMemory());
+
+        // Double the size of the buffer
+        buffer.capacity(capacity << 1);
+        capacity = buffer.capacity();
+        assertEquals(expectedUsedMemory(allocator, capacity), metric.usedHeapMemory());
+        assertThat(allocator.pinnedHeapMemory())
+                .isGreaterThanOrEqualTo(capacity)
+                .isLessThanOrEqualTo(metric.usedHeapMemory());
+
+        buffer.release();
+        assertEquals(expectedUsedMemoryAfterRelease(allocator, capacity), metric.usedHeapMemory());
+        assertThat(allocator.pinnedHeapMemory())
+                .isGreaterThanOrEqualTo(0)
+                .isLessThanOrEqualTo(metric.usedHeapMemory());
+        trimCaches(allocator);
+        assertEquals(0, allocator.pinnedHeapMemory());
+
+        int[] capacities = new int[30];
+        Random rng = new Random();
+        for (int i = 0; i < capacities.length; i++) {
+            capacities[i] = initialCapacity / 4 + rng.nextInt(8 * initialCapacity);
+        }
+        ByteBuf[] bufs = new ByteBuf[capacities.length];
+        for (int i = 0; i < 20; i++) {
+            bufs[i] = allocator.heapBuffer(capacities[i], 2 * capacities[i]);
+        }
+        for (int i = 0; i < 10; i++) {
+            bufs[i].release();
+        }
+        for (int i = 20; i < 30; i++) {
+            bufs[i] = allocator.heapBuffer(capacities[i], 2 * capacities[i]);
+        }
+        for (int i = 0; i < 10; i++) {
+            bufs[i] = allocator.heapBuffer(capacities[i], 2 * capacities[i]);
+        }
+        for (int i = 0; i < 30; i++) {
+            bufs[i].release();
+        }
+        trimCaches(allocator);
+        assertEquals(0, allocator.pinnedDirectMemory());
+    }
 }