Add ReferenceCountUtil.touch(...) calls before we store messages into… (#13838)

normanmaurer · web-flow · commit 8d7c11357cf5 · 2024-02-12T19:04:11.000+01:00
… a queue to make debugging leaks easier

Motivation:

We should touch our messages when we take over ownership, this allows
easier debugging of buffer leaks.

Modifications:

Let's add touch call before enqueue into internal datastructure

Result:

Easier to debug buffer leaks
diff --git a/transport/src/main/java/io/netty/channel/AbstractCoalescingBufferQueue.java b/transport/src/main/java/io/netty/channel/AbstractCoalescingBufferQueue.java
@@ -58,6 +58,9 @@ public final void addFirst(ByteBuf buf, ChannelPromise promise) {
     }
 
     private void addFirst(ByteBuf buf, ChannelFutureListener listener) {
+        // Touch the message to make it easier to debug buffer leaks.
+        buf.touch();
+
         if (listener != null) {
             bufAndListenerPairs.addFirst(listener);
         }
@@ -91,6 +94,9 @@ public final void add(ByteBuf buf, ChannelPromise promise) {
      * @param listener to notify when all the bytes have been consumed and written, can be {@code null}.
      */
     public final void add(ByteBuf buf, ChannelFutureListener listener) {
+        // Touch the message to make it easier to debug buffer leaks.
+        buf.touch();
+
         // buffers are added before promises so that we naturally 'consume' the entire buffer during removal
         // before we complete it's promise.
         bufAndListenerPairs.add(buf);
diff --git a/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java b/transport/src/main/java/io/netty/channel/ChannelOutboundBuffer.java
@@ -125,6 +125,16 @@ public void addMessage(Object msg, int size, ChannelPromise promise) {
             unflushedEntry = entry;
         }
 
+        // Touch the message to make it easier to debug buffer leaks.
+
+        // this save both checking against the ReferenceCounted interface
+        // and makes better use of virtual calls vs interface ones
+        if (msg instanceof AbstractReferenceCountedByteBuf) {
+            ((AbstractReferenceCountedByteBuf) msg).touch();
+        } else {
+            ReferenceCountUtil.touch(msg);
+        }
+
         // increment pending bytes after adding message to the unflushed arrays.
         // See https://github.com/netty/netty/issues/1619
         incrementPendingOutboundBytes(entry.pendingSize, false);
diff --git a/transport/src/main/java/io/netty/channel/PendingWriteQueue.java b/transport/src/main/java/io/netty/channel/PendingWriteQueue.java
@@ -15,6 +15,7 @@
  */
 package io.netty.channel;
 
+import io.netty.buffer.AbstractReferenceCountedByteBuf;
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.concurrent.EventExecutor;
 import io.netty.util.concurrent.PromiseCombiner;
@@ -119,6 +120,15 @@ public void add(Object msg, ChannelPromise promise) {
         size ++;
         bytes += messageSize;
         tracker.incrementPendingOutboundBytes(write.size);
+        // Touch the message to make it easier to debug buffer leaks.
+
+        // this save both checking against the ReferenceCounted interface
+        // and makes better use of virtual calls vs interface ones
+        if (msg instanceof AbstractReferenceCountedByteBuf) {
+            ((AbstractReferenceCountedByteBuf) msg).touch();
+        } else {
+            ReferenceCountUtil.touch(msg);
+        }
     }
 
     /**
