Repository: netty/netty
Issue: PooledByteBufAllocator.pinnedDirectMemory is sometimes returning 0 even if some direct buffers are used

I'm trying to monitor the estimate of memory that is used by in-use direct buffers.
To do so, I'm trying to use the PooledByteBufAllocator.pinnedDirectMemory() method but sometimes, it seems that this method is returning 0 even if some buffers are currently allocated and used (not yet released).

### Expected behavior

PooledByteBufAllocator.pinnedDirectMemory() should not return 0 in case some buffers are currently allocated.

### Actual behavior

Sometimes, not always, it seems that the PooledByteBufAllocator.pinnedDirectMemory() is returning 0 even if some buffers are allocated.

### Steps to reproduce

Please check attached sample project: 

[netty.pinnedMemory.test.tar.gz](https://github.com/netty/netty/files/7835795/netty.pinnedMemory.test.tar.gz)

The Test class is allocating some buffers, and before releasing them, it displays the pinned direct memory. It also displays an estimate of used memory by traversing arena of direct buffers, and for each arena, it accumulates the diff between the  chunk size and the bytes that are not yet allocated by in-use buffers:

```
    static long usedMem(List<PoolArenaMetric> arenas) {
        long totalUsed = 0;
        for (PoolArenaMetric arenaMetrics : arenas) {
            for (PoolChunkListMetric arenaMetric : arenaMetrics.chunkLists()) {
                for (PoolChunkMetric chunkMetric : arenaMetric) {
                    // chunkMetric.chunkSize() returns maximum of bytes that can be served out of the chunk
                    // and chunkMetric.freeBytes() returns the bytes that are not yet allocated by in-use buffers
                    totalUsed += (chunkMetric.chunkSize() - chunkMetric.freeBytes());
                }
            }
        }
        return totalUsed;
    }
```
To run the program:

./gradlew jar
./gradlew runTest

The program displays many logs, like this:

```
used=1810432, pinned=1372971008
used=2097152, pinned=1373257728
used=1974272, pinned=1371643904
```

_used_ corresponds to the result of the _usedMem_ method shown above, and _pinned_ corresponds to the result of the call to alloc.pinnedDirectMemory().

Now, sometimes, you will notice that the pinned memory is zero:

```
used=2031616, pinned=1835008
used=1892352, pinned=188416
used=1916928, pinned=212992
used=1892352, pinned=0
used=2023424, pinned=0
used=2072576, pinned=0
...
used=2220032, pinned=2146779136
used=2310144, pinned=2146869248
used=1843200, pinned=2144354304
used=1974272, pinned=2144485376
used=2007040, pinned=2144518144
used=1998848, pinned=2143215616
used=2138112, pinned=2143354880
```

To stop, just press CTRL-C

### Minimal yet complete reproducer code (or URL to code)

Please see attached sample project above

### Netty version

4.1.72.Final

### JVM version (e.g. `java -version`)

openjdk version "1.8.0_302"
OpenJDK Runtime Environment (Zulu 8.56.0.23-CA-macos-aarch64) (build 1.8.0_302-b08)
OpenJDK 64-Bit Server VM (Zulu 8.56.0.23-CA-macos-aarch64) (build 25.302-b08, mixed mode)

### OS version (e.g. `uname -a`)

MacOS M1:

Darwin xxx 21.1.0 Darwin Kernel Version 21.1.0: Wed Oct 13 17:33:24 PDT 2021; root:xnu-8019.41.5~1/RELEASE_ARM64_T8101 arm64

The issue is also observed on linux Ubuntu:

Linux xxx 5.13.0-22-generic #22-Ubuntu SMP Fri Nov 5 13:21:36 UTC 2021 x86_64 x86_64 x86_64 GNU/Linux


Issue comments:
@chrisvest can you have a look at this one ?
I got the test running and showing the expected failure. It disappears if I add a `alloc.trimCurrentThreadCache();` after the `release()` loop. The accounting is done in a bad place, where we might either miss out on activity happening at the cache layer, or we'll have the same memory activity accounted twice. I need to find a better place to do the accounting of the pinned memory.

Pull request review comments:
```suggestion
        int result = PINNED_UPDATER.addAndGet(this, -delta);
        assert result >= 0;
```
nit: Not strong about this but we could also just use `AtomicInteger` directly here as I wouldn't expect a massive amount of these instances. That said I am not strong on it at all.
nit: we could size it correctly from the start 
I wonder if we want to run the test one time with cache enabled and one time without. 
I was mostly thinking about the number of memory indirections, tbh.
I guess it not matters too much... That said I am not strong here like I said. So just keep it.
Added a second test.
Changed it anyway.

Patch that resolved the issue:
diff --git a/buffer/src/main/java/io/netty/buffer/PoolChunk.java b/buffer/src/main/java/io/netty/buffer/PoolChunk.java
index 70bc6419275..96df62ffed9 100644
--- a/buffer/src/main/java/io/netty/buffer/PoolChunk.java
+++ b/buffer/src/main/java/io/netty/buffer/PoolChunk.java
@@ -19,6 +19,7 @@
 import java.util.ArrayDeque;
 import java.util.Deque;
 import java.util.PriorityQueue;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Description of algorithm for PageRun/PoolSubpage allocation from PoolChunk
@@ -160,6 +161,11 @@ final class PoolChunk<T> implements PoolChunkMetric {
      */
     private final PoolSubpage<T>[] subpages;
 
+    /**
+     * Accounting of pinned memory – memory that is currently in use by ByteBuf instances.
+     */
+    private final AtomicInteger pinnedBytes;
+
     private final int pageSize;
     private final int pageShifts;
     private final int chunkSize;
@@ -172,7 +178,6 @@ final class PoolChunk<T> implements PoolChunkMetric {
     private final Deque<ByteBuffer> cachedNioBuffers;
 
     int freeBytes;
-    int pinnedBytes;
 
     PoolChunkList<T> parent;
     PoolChunk<T> prev;
@@ -202,6 +207,7 @@ final class PoolChunk<T> implements PoolChunkMetric {
         insertAvailRun(0, pages, initHandle);
 
         cachedNioBuffers = new ArrayDeque<ByteBuffer>(8);
+        pinnedBytes = new AtomicInteger();
     }
 
     /** Creates a special chunk that is not pooled. */
@@ -217,6 +223,7 @@ final class PoolChunk<T> implements PoolChunkMetric {
         subpages = null;
         chunkSize = size;
         cachedNioBuffers = null;
+        pinnedBytes = new AtomicInteger();
     }
 
     private static LongPriorityQueue[] newRunsAvailqueueArray(int size) {
@@ -343,7 +350,6 @@ private long allocateRun(int runSize) {
 
             int pinnedSize = runSize(pageShifts, handle);
             freeBytes -= pinnedSize;
-            pinnedBytes += pinnedSize;
             return handle;
         }
     }
@@ -453,7 +459,6 @@ private long allocateSubpage(int sizeIdx) {
      */
     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);
@@ -557,8 +562,9 @@ private static long toRunHandle(int runOffset, int runPages, int inUsed) {
     void initBuf(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle, int reqCapacity,
                  PoolThreadCache threadCache) {
         if (isRun(handle)) {
+            int maxLength = runSize(pageShifts, handle);
             buf.init(this, nioBuffer, handle, runOffset(handle) << pageShifts,
-                     reqCapacity, runSize(pageShifts, handle), arena.parent.threadCache());
+                     reqCapacity, maxLength, arena.parent.threadCache());
         } else {
             initBufWithSubpage(buf, nioBuffer, handle, reqCapacity, threadCache);
         }
@@ -577,6 +583,18 @@ void initBufWithSubpage(PooledByteBuf<T> buf, ByteBuffer nioBuffer, long handle,
         buf.init(this, nioBuffer, handle, offset, reqCapacity, s.elemSize, threadCache);
     }
 
+    void incrementPinnedMemory(int delta) {
+        assert delta > 0;
+        int result = pinnedBytes.addAndGet(delta);
+        assert result > 0;
+    }
+
+    void decrementPinnedMemory(int delta) {
+        assert delta > 0;
+        int result = pinnedBytes.addAndGet(-delta);
+        assert result >= 0;
+    }
+
     @Override
     public int chunkSize() {
         return chunkSize;
@@ -590,9 +608,7 @@ public int freeBytes() {
     }
 
     public int pinnedBytes() {
-        synchronized (arena) {
-            return pinnedBytes;
-        }
+        return pinnedBytes.get();
     }
 
     @Override
diff --git a/buffer/src/main/java/io/netty/buffer/PooledByteBuf.java b/buffer/src/main/java/io/netty/buffer/PooledByteBuf.java
index bd57f5acfa8..a7fbf48e563 100644
--- a/buffer/src/main/java/io/netty/buffer/PooledByteBuf.java
+++ b/buffer/src/main/java/io/netty/buffer/PooledByteBuf.java
@@ -62,6 +62,7 @@ private void init0(PoolChunk<T> chunk, ByteBuffer nioBuffer,
         assert !PoolChunk.isSubpage(handle) || chunk.arena.size2SizeIdx(maxLength) <= chunk.arena.smallMaxSizeIdx:
                 "Allocated small sub-page handle for a buffer size that isn't \"small.\"";
 
+        chunk.incrementPinnedMemory(maxLength);
         this.chunk = chunk;
         memory = chunk.memory;
         tmpNioBuf = nioBuffer;
@@ -117,6 +118,7 @@ public final ByteBuf capacity(int newCapacity) {
         }
 
         // Reallocation required.
+        chunk.decrementPinnedMemory(maxLength);
         chunk.arena.reallocate(this, newCapacity, true);
         return this;
     }
@@ -170,6 +172,7 @@ protected final void deallocate() {
             final long handle = this.handle;
             this.handle = -1;
             memory = null;
+            chunk.decrementPinnedMemory(maxLength);
             chunk.arena.free(chunk, tmpNioBuf, handle, maxLength, cache);
             tmpNioBuf = null;
             chunk = null;


Test patch:
diff --git a/buffer/src/test/java/io/netty/buffer/PooledByteBufAllocatorTest.java b/buffer/src/test/java/io/netty/buffer/PooledByteBufAllocatorTest.java
index e2bf4672361..0a977ff0b71 100644
--- a/buffer/src/test/java/io/netty/buffer/PooledByteBufAllocatorTest.java
+++ b/buffer/src/test/java/io/netty/buffer/PooledByteBufAllocatorTest.java
@@ -30,7 +30,9 @@
 import java.util.Random;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.LockSupport;
 import org.junit.jupiter.api.Timeout;
@@ -821,4 +823,92 @@ private void testUsedHeapMemory(int initialCapacity) {
         trimCaches(allocator);
         assertEquals(0, allocator.pinnedDirectMemory());
     }
+
+    @Test
+    public void pinnedMemoryMustReflectBuffersInUseWithThreadLocalCaching() {
+        pinnedMemoryMustReflectBuffersInUse(true);
+    }
+
+    @Test
+    public void pinnedMemoryMustReflectBuffersInUseWithoutThreadLocalCaching() {
+        pinnedMemoryMustReflectBuffersInUse(false);
+    }
+
+    private static void pinnedMemoryMustReflectBuffersInUse(boolean useThreadLocalCaching) {
+        int smallCacheSize;
+        int normalCacheSize;
+        if (useThreadLocalCaching) {
+            smallCacheSize = PooledByteBufAllocator.defaultSmallCacheSize();
+            normalCacheSize = PooledByteBufAllocator.defaultNormalCacheSize();
+        } else {
+            smallCacheSize = 0;
+            normalCacheSize = 0;
+        }
+        int directMemoryCacheAlignment = 0;
+        PooledByteBufAllocator alloc = new PooledByteBufAllocator(
+                PooledByteBufAllocator.defaultPreferDirect(),
+                PooledByteBufAllocator.defaultNumHeapArena(),
+                PooledByteBufAllocator.defaultNumDirectArena(),
+                PooledByteBufAllocator.defaultPageSize(),
+                PooledByteBufAllocator.defaultMaxOrder(),
+                smallCacheSize,
+                normalCacheSize,
+                useThreadLocalCaching,
+                directMemoryCacheAlignment);
+        PooledByteBufAllocatorMetric metric = alloc.metric();
+        AtomicLong capSum = new AtomicLong();
+
+        for (long index = 0; index < 10000; index++) {
+            ThreadLocalRandom rnd = ThreadLocalRandom.current();
+            int bufCount = rnd.nextInt(1, 100);
+            List<ByteBuf> buffers = new ArrayList<ByteBuf>(bufCount);
+
+            if (index % 2 == 0) {
+                // ensure that we allocate a small buffer
+                for (int i = 0; i < bufCount; i++) {
+                    ByteBuf buf = alloc.directBuffer(rnd.nextInt(8, 128));
+                    buffers.add(buf);
+                    capSum.addAndGet(buf.capacity());
+                }
+            } else {
+                // allocate a larger buffer
+                for (int i = 0; i < bufCount; i++) {
+                    ByteBuf buf = alloc.directBuffer(rnd.nextInt(1024, 1024 * 100));
+                    buffers.add(buf);
+                    capSum.addAndGet(buf.capacity());
+                }
+            }
+
+            if (index % 100 == 0) {
+                long used = usedMemory(metric.directArenas());
+                long pinned = alloc.pinnedDirectMemory();
+                assertThat(capSum.get()).isLessThanOrEqualTo(pinned);
+                assertThat(pinned).isLessThanOrEqualTo(used);
+            }
+
+            for (ByteBuf buffer : buffers) {
+                buffer.release();
+            }
+            capSum.set(0);
+            // After releasing all buffers, pinned memory must be zero
+            assertThat(alloc.pinnedDirectMemory()).isZero();
+        }
+    }
+
+    /**
+     * Returns an estimate of bytes used by currently in-use buffers
+     */
+    private static long usedMemory(List<PoolArenaMetric> arenas) {
+        long totalUsed = 0;
+        for (PoolArenaMetric arenaMetrics : arenas) {
+            for (PoolChunkListMetric arenaMetric : arenaMetrics.chunkLists()) {
+                for (PoolChunkMetric chunkMetric : arenaMetric) {
+                    // chunkMetric.chunkSize() returns maximum of bytes that can be served out of the chunk
+                    // and chunkMetric.freeBytes() returns the bytes that are not yet allocated by in-use buffers
+                    totalUsed += chunkMetric.chunkSize() - chunkMetric.freeBytes();
+                }
+            }
+        }
+        return totalUsed;
+    }
 }