Repository: netty/netty (Java)

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



Hints:
@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.

Patch:
diff --git a/buffer/src/main/java/io/netty/buffer/PoolChunk.java b/buffer/src/main/java/io/netty/buffer/PoolChunk.java
--- 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
--- 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
--- 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;
+    }
 }