Fix false-positives when using ResourceLeakDetector.

Motivation:

We need to ensure the tracked object can not be GC'ed before ResourceLeak.close() is called as otherwise we may get false-positives reported by the ResourceLeakDetector. This can happen as the JIT / GC may be able to figure out that we do not need the tracked object anymore and so already enqueue it for collection before we actually get a chance to close the enclosing ResourceLeak.

Modifications:

- Add ResourceLeakTracker and deprecate the old ResourceLeak
- Fix some javadocs to correctly release buffers.
- Add a unit test for ResourceLeakDetector that shows that ResourceLeakTracker has not the problems.

Result:

No more false-positives reported by ResourceLeakDetector when ResourceLeakDetector.track(...) is used.

buffer/src/main/java/io/netty/buffer/AbstractByteBufAllocator.java

@@ -16,8 +16,8 @@
 
 package io.netty.buffer;
 
-import io.netty.util.ResourceLeak;
 import io.netty.util.ResourceLeakDetector;
+import io.netty.util.ResourceLeakTracker;
 import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.StringUtil;
 
@@ -29,17 +29,17 @@ public abstract class AbstractByteBufAllocator implements ByteBufAllocator {
     static final int DEFAULT_MAX_COMPONENTS = 16;
 
     protected static ByteBuf toLeakAwareBuffer(ByteBuf buf) {
-        ResourceLeak leak;
+        ResourceLeakTracker<ByteBuf> leak;
         switch (ResourceLeakDetector.getLevel()) {
             case SIMPLE:
-                leak = AbstractByteBuf.leakDetector.open(buf);
+                leak = AbstractByteBuf.leakDetector.track(buf);
                 if (leak != null) {
                     buf = new SimpleLeakAwareByteBuf(buf, leak);
                 }
                 break;
             case ADVANCED:
             case PARANOID:
-                leak = AbstractByteBuf.leakDetector.open(buf);
+                leak = AbstractByteBuf.leakDetector.track(buf);
                 if (leak != null) {
                     buf = new AdvancedLeakAwareByteBuf(buf, leak);
                 }
@@ -49,17 +49,17 @@ public abstract class AbstractByteBufAllocator implements ByteBufAllocator {
     }
 
     protected static CompositeByteBuf toLeakAwareBuffer(CompositeByteBuf buf) {
-        ResourceLeak leak;
+        ResourceLeakTracker<ByteBuf> leak;
         switch (ResourceLeakDetector.getLevel()) {
             case SIMPLE:
-                leak = AbstractByteBuf.leakDetector.open(buf);
+                leak = AbstractByteBuf.leakDetector.track(buf);
                 if (leak != null) {
                     buf = new SimpleLeakAwareCompositeByteBuf(buf, leak);
                 }
                 break;
             case ADVANCED:
             case PARANOID:
-                leak = AbstractByteBuf.leakDetector.open(buf);
+                leak = AbstractByteBuf.leakDetector.track(buf);
                 if (leak != null) {
                     buf = new AdvancedLeakAwareCompositeByteBuf(buf, leak);
                 }

buffer/src/main/java/io/netty/buffer/AdvancedLeakAwareByteBuf.java

@@ -16,7 +16,7 @@
 
 package io.netty.buffer;
 
-import io.netty.util.ResourceLeak;
+import io.netty.util.ResourceLeakTracker;
 import io.netty.util.internal.SystemPropertyUtil;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
@@ -30,7 +30,7 @@ import java.nio.channels.GatheringByteChannel;
 import java.nio.channels.ScatteringByteChannel;
 import java.nio.charset.Charset;
 
-final class AdvancedLeakAwareByteBuf extends WrappedByteBuf {
+final class AdvancedLeakAwareByteBuf extends SimpleLeakAwareByteBuf {
 
     private static final String PROP_ACQUIRE_AND_RELEASE_ONLY = "io.netty.leakDetection.acquireAndReleaseOnly";
     private static final boolean ACQUIRE_AND_RELEASE_ONLY;
@@ -45,36 +45,15 @@ final class AdvancedLeakAwareByteBuf extends WrappedByteBuf {
         }
     }
 
-    private final ResourceLeak leak;
+    AdvancedLeakAwareByteBuf(ByteBuf buf, ResourceLeakTracker<ByteBuf> leak) {
-
+        super(buf, leak);
-    AdvancedLeakAwareByteBuf(ByteBuf buf, ResourceLeak leak) {
-        super(buf);
-        this.leak = leak;
     }
 
-    @Override
+    AdvancedLeakAwareByteBuf(ByteBuf wrapped, ByteBuf trackedByteBuf, ResourceLeakTracker<ByteBuf> leak) {
-    public boolean release() {
+        super(wrapped, trackedByteBuf, leak);
-        boolean deallocated =  super.release();
-        if (deallocated) {
-            leak.close();
-        } else {
-            leak.record();
-        }
-        return deallocated;
     }
 
-    @Override
+    static void recordLeakNonRefCountingOperation(ResourceLeakTracker<ByteBuf> leak) {
-    public boolean release(int decrement) {
-        boolean deallocated = super.release(decrement);
-        if (deallocated) {
-            leak.close();
-        } else {
-            leak.record();
-        }
-        return deallocated;
-    }
-
-    static void recordLeakNonRefCountingOperation(ResourceLeak leak) {
         if (!ACQUIRE_AND_RELEASE_ONLY) {
             leak.record();
         }
@@ -83,35 +62,31 @@ final class AdvancedLeakAwareByteBuf extends WrappedByteBuf {
     @Override
     public ByteBuf order(ByteOrder endianness) {
         recordLeakNonRefCountingOperation(leak);
-        if (order() == endianness) {
+        return super.order(endianness);
-            return this;
-        } else {
-            return new AdvancedLeakAwareByteBuf(super.order(endianness), leak);
-        }
     }
 
     @Override
     public ByteBuf slice() {
         recordLeakNonRefCountingOperation(leak);
-        return new AdvancedLeakAwareByteBuf(super.slice(), leak);
+        return super.slice();
     }
 
     @Override
     public ByteBuf slice(int index, int length) {
         recordLeakNonRefCountingOperation(leak);
-        return new AdvancedLeakAwareByteBuf(super.slice(index, length), leak);
+        return super.slice(index, length);
     }
 
     @Override
     public ByteBuf duplicate() {
         recordLeakNonRefCountingOperation(leak);
-        return new AdvancedLeakAwareByteBuf(super.duplicate(), leak);
+        return super.duplicate();
     }
 
     @Override
     public ByteBuf readSlice(int length) {
         recordLeakNonRefCountingOperation(leak);
-        return new AdvancedLeakAwareByteBuf(super.readSlice(length), leak);
+        return super.readSlice(length);
     }
 
     @Override
@@ -726,6 +701,12 @@ final class AdvancedLeakAwareByteBuf extends WrappedByteBuf {
         return super.toString(index, length, charset);
     }
 
+    @Override
+    public ByteBuf capacity(int newCapacity) {
+        recordLeakNonRefCountingOperation(leak);
+        return super.capacity(newCapacity);
+    }
+
     @Override
     public ByteBuf retain() {
         leak.record();
@@ -739,8 +720,8 @@ final class AdvancedLeakAwareByteBuf extends WrappedByteBuf {
     }
 
     @Override
-    public ByteBuf capacity(int newCapacity) {
+    protected AdvancedLeakAwareByteBuf newLeakAwareByteBuf(
-        recordLeakNonRefCountingOperation(leak);
+            ByteBuf buf, ByteBuf trackedByteBuf, ResourceLeakTracker<ByteBuf> leakTracker) {
-        return super.capacity(newCapacity);
+        return new AdvancedLeakAwareByteBuf(buf, trackedByteBuf, leakTracker);
     }
 }

buffer/src/main/java/io/netty/buffer/AdvancedLeakAwareCompositeByteBuf.java

@@ -15,8 +15,7 @@
  */
 package io.netty.buffer;
 
-
+import io.netty.util.ResourceLeakTracker;
-import io.netty.util.ResourceLeak;
 
 import java.io.IOException;
 import java.io.InputStream;
@@ -31,47 +30,40 @@ import java.util.List;
 
 import static io.netty.buffer.AdvancedLeakAwareByteBuf.recordLeakNonRefCountingOperation;
 
-final class AdvancedLeakAwareCompositeByteBuf extends WrappedCompositeByteBuf {
+final class AdvancedLeakAwareCompositeByteBuf extends SimpleLeakAwareCompositeByteBuf {
 
-    private final ResourceLeak leak;
+    AdvancedLeakAwareCompositeByteBuf(CompositeByteBuf wrapped, ResourceLeakTracker<ByteBuf> leak) {
-
+        super(wrapped, leak);
-    AdvancedLeakAwareCompositeByteBuf(CompositeByteBuf wrapped, ResourceLeak leak) {
-        super(wrapped);
-        this.leak = leak;
     }
 
     @Override
     public ByteBuf order(ByteOrder endianness) {
         recordLeakNonRefCountingOperation(leak);
-        if (order() == endianness) {
+        return super.order(endianness);
-            return this;
-        } else {
-            return new AdvancedLeakAwareByteBuf(super.order(endianness), leak);
-        }
     }
 
     @Override
     public ByteBuf slice() {
         recordLeakNonRefCountingOperation(leak);
-        return new AdvancedLeakAwareByteBuf(super.slice(), leak);
+        return super.slice();
     }
 
     @Override
     public ByteBuf slice(int index, int length) {
         recordLeakNonRefCountingOperation(leak);
-        return new AdvancedLeakAwareByteBuf(super.slice(index, length), leak);
+        return super.slice(index, length);
     }
 
     @Override
     public ByteBuf duplicate() {
         recordLeakNonRefCountingOperation(leak);
-        return new AdvancedLeakAwareByteBuf(super.duplicate(), leak);
+        return super.duplicate();
     }
 
     @Override
     public ByteBuf readSlice(int length) {
         recordLeakNonRefCountingOperation(leak);
-        return new AdvancedLeakAwareByteBuf(super.readSlice(length), leak);
+        return super.readSlice(length);
     }
 
     @Override
@@ -807,24 +799,8 @@ final class AdvancedLeakAwareCompositeByteBuf extends WrappedCompositeByteBuf {
     }
 
     @Override
-    public boolean release() {
+    protected AdvancedLeakAwareByteBuf newLeakAwareByteBuf(
-        boolean deallocated = super.release();
+            ByteBuf wrapped, ByteBuf trackedByteBuf, ResourceLeakTracker<ByteBuf> leakTracker) {
-        if (deallocated) {
+        return new AdvancedLeakAwareByteBuf(wrapped, trackedByteBuf, leakTracker);
-            leak.close();
-        } else {
-            leak.record();
-        }
-        return deallocated;
-    }
-
-    @Override
-    public boolean release(int decrement) {
-        boolean deallocated = super.release(decrement);
-        if (deallocated) {
-            leak.close();
-        } else {
-            leak.record();
-        }
-        return deallocated;
     }
 }

buffer/src/main/java/io/netty/buffer/SimpleLeakAwareByteBuf.java

@@ -16,64 +16,93 @@
 
 package io.netty.buffer;
 
-import io.netty.util.ResourceLeak;
+import io.netty.util.ResourceLeakDetector;
+import io.netty.util.ResourceLeakTracker;
+import io.netty.util.internal.ObjectUtil;
 
 import java.nio.ByteOrder;
 
-final class SimpleLeakAwareByteBuf extends WrappedByteBuf {
+class SimpleLeakAwareByteBuf extends WrappedByteBuf {
 
-    private final ResourceLeak leak;
+    /**
+     * This object's is associated with the {@link ResourceLeakTracker}. When {@link ResourceLeakTracker#close(Object)}
+     * is called this object will be used as the argument. It is also assumed that this object is used when
+     * {@link ResourceLeakDetector#track(Object)} is called to create {@link #leak}.
+     */
+    private final ByteBuf trackedByteBuf;
+    final ResourceLeakTracker<ByteBuf> leak;
 
-    SimpleLeakAwareByteBuf(ByteBuf buf, ResourceLeak leak) {
+    SimpleLeakAwareByteBuf(ByteBuf wrapped, ByteBuf trackedByteBuf, ResourceLeakTracker<ByteBuf> leak) {
-        super(buf);
+        super(wrapped);
-        this.leak = leak;
+        this.trackedByteBuf = ObjectUtil.checkNotNull(trackedByteBuf, "trackedByteBuf");
+        this.leak = ObjectUtil.checkNotNull(leak, "leak");
     }
 
-    @Override
+    SimpleLeakAwareByteBuf(ByteBuf wrapped, ResourceLeakTracker<ByteBuf> leak) {
-    public boolean release() {
+        this(wrapped, wrapped, leak);
-        boolean deallocated =  super.release();
-        if (deallocated) {
-            leak.close();
-        }
-        return deallocated;
-    }
-
-    @Override
-    public boolean release(int decrement) {
-        boolean deallocated = super.release(decrement);
-        if (deallocated) {
-            leak.close();
-        }
-        return deallocated;
-    }
-
-    @Override
-    public ByteBuf order(ByteOrder endianness) {
-        leak.record();
-        if (order() == endianness) {
-            return this;
-        } else {
-            return new SimpleLeakAwareByteBuf(super.order(endianness), leak);
-        }
     }
 
     @Override
     public ByteBuf slice() {
-        return new SimpleLeakAwareByteBuf(super.slice(), leak);
+        return newSharedLeakAwareByteBuf(super.slice());
     }
 
     @Override
     public ByteBuf slice(int index, int length) {
-        return new SimpleLeakAwareByteBuf(super.slice(index, length), leak);
+        return newSharedLeakAwareByteBuf(super.slice(index, length));
     }
 
     @Override
     public ByteBuf duplicate() {
-        return new SimpleLeakAwareByteBuf(super.duplicate(), leak);
+        return newSharedLeakAwareByteBuf(super.duplicate());
     }
 
     @Override
     public ByteBuf readSlice(int length) {
-        return new SimpleLeakAwareByteBuf(super.readSlice(length), leak);
+        return newSharedLeakAwareByteBuf(super.readSlice(length));
+    }
+
+    @Override
+    public final boolean release() {
+        if (super.release()) {
+            closeLeak();
+            return true;
+        }
+        return false;
+    }
+
+    @Override
+    public final boolean release(int decrement) {
+        if (super.release(decrement)) {
+            closeLeak();
+            return true;
+        }
+        return false;
+    }
+
+    private void closeLeak() {
+        // Close the ResourceLeakTracker with the tracked ByteBuf as argument. This must be the same that was used when
+        // calling DefaultResourceLeak.track(...).
+        boolean closed = leak.close(trackedByteBuf);
+        assert closed;
+    }
+
+    @Override
+    public ByteBuf order(ByteOrder endianness) {
+        if (order() == endianness) {
+            return this;
+        } else {
+            return newSharedLeakAwareByteBuf(super.order(endianness));
+        }
+    }
+
+    private SimpleLeakAwareByteBuf newSharedLeakAwareByteBuf(
+            ByteBuf wrapped) {
+        return newLeakAwareByteBuf(wrapped, trackedByteBuf, leak);
+    }
+
+    protected SimpleLeakAwareByteBuf newLeakAwareByteBuf(
+            ByteBuf buf, ByteBuf trackedByteBuf, ResourceLeakTracker<ByteBuf> leakTracker) {
+        return new SimpleLeakAwareByteBuf(buf, trackedByteBuf, leakTracker);
     }
 }

buffer/src/main/java/io/netty/buffer/SimpleLeakAwareCompositeByteBuf.java

@@ -16,64 +16,86 @@
 package io.netty.buffer;
 
 
-import io.netty.util.ResourceLeak;
+import io.netty.util.ResourceLeakTracker;
+import io.netty.util.internal.ObjectUtil;
 
 import java.nio.ByteOrder;
 
-final class SimpleLeakAwareCompositeByteBuf extends WrappedCompositeByteBuf {
+class SimpleLeakAwareCompositeByteBuf extends WrappedCompositeByteBuf {
 
-    private final ResourceLeak leak;
+    final ResourceLeakTracker<ByteBuf> leak;
 
-    SimpleLeakAwareCompositeByteBuf(CompositeByteBuf wrapped, ResourceLeak leak) {
+    SimpleLeakAwareCompositeByteBuf(CompositeByteBuf wrapped, ResourceLeakTracker<ByteBuf> leak) {
         super(wrapped);
-        this.leak = leak;
+        this.leak = ObjectUtil.checkNotNull(leak, "leak");
     }
 
     @Override
-    public boolean release() {
+    public final boolean release() {
-        boolean deallocated = super.release();
+        // Call unwrap() before just in case that super.release() will change the ByteBuf instance that is returned
-        if (deallocated) {
+        // by unwrap().
-            leak.close();
+        ByteBuf unwrapped = unwrap();
+        if (super.release()) {
+            closeLeak(unwrapped);
+            return true;
         }
-        return deallocated;
+        return false;
     }
 
     @Override
-    public boolean release(int decrement) {
+    public final boolean release(int decrement) {
-        boolean deallocated = super.release(decrement);
+        // Call unwrap() before just in case that super.release() will change the ByteBuf instance that is returned
-        if (deallocated) {
+        // by unwrap().
-            leak.close();
+        ByteBuf unwrapped = unwrap();
+        if (super.release(decrement)) {
+            closeLeak(unwrapped);
+            return true;
         }
-        return deallocated;
+        return false;
+    }
+
+    private void closeLeak(ByteBuf trackedByteBuf) {
+        // Close the ResourceLeakTracker with the tracked ByteBuf as argument. This must be the same that was used when
+        // calling DefaultResourceLeak.track(...).
+        boolean closed = leak.close(trackedByteBuf);
+        assert closed;
     }
 
     @Override
     public ByteBuf order(ByteOrder endianness) {
-        leak.record();
         if (order() == endianness) {
             return this;
         } else {
-            return new SimpleLeakAwareByteBuf(super.order(endianness), leak);
+            return newLeakAwareByteBuf(super.order(endianness));
         }
     }
 
     @Override
     public ByteBuf slice() {
-        return new SimpleLeakAwareByteBuf(super.slice(), leak);
+        return newLeakAwareByteBuf(super.slice());
     }
 
     @Override
     public ByteBuf slice(int index, int length) {
-        return new SimpleLeakAwareByteBuf(super.slice(index, length), leak);
+        return newLeakAwareByteBuf(super.slice(index, length));
     }
 
     @Override
     public ByteBuf duplicate() {
-        return new SimpleLeakAwareByteBuf(super.duplicate(), leak);
+        return newLeakAwareByteBuf(super.duplicate());
     }
 
     @Override
     public ByteBuf readSlice(int length) {
-        return new SimpleLeakAwareByteBuf(super.readSlice(length), leak);
+        return newLeakAwareByteBuf(super.readSlice(length));
+    }
+
+    private SimpleLeakAwareByteBuf newLeakAwareByteBuf(ByteBuf wrapped) {
+        return newLeakAwareByteBuf(wrapped, unwrap(), leak);
+    }
+
+    protected SimpleLeakAwareByteBuf newLeakAwareByteBuf(
+            ByteBuf wrapped, ByteBuf trackedByteBuf, ResourceLeakTracker<ByteBuf> leakTracker) {
+        return new SimpleLeakAwareByteBuf(wrapped, trackedByteBuf, leakTracker);
     }
 }

buffer/src/test/java/io/netty/buffer/AbstractByteBufTest.java

@@ -1525,9 +1525,9 @@ public abstract class AbstractByteBufTest {
         random.nextBytes(value);
         // Prevent overflow / underflow
         if (value[0] == 0) {
-            value[0] ++;
+            value[0]++;
         } else if (value[0] == -1) {
-            value[0] --;
+            value[0]--;
         }
 
         buffer.setIndex(0, value.length);
@@ -1536,13 +1536,13 @@ public abstract class AbstractByteBufTest {
         assertEquals(0, buffer.compareTo(wrappedBuffer(value)));
         assertEquals(0, buffer.compareTo(wrappedBuffer(value).order(LITTLE_ENDIAN)));
 
-        value[0] ++;
+        value[0]++;
         assertTrue(buffer.compareTo(wrappedBuffer(value)) < 0);
         assertTrue(buffer.compareTo(wrappedBuffer(value).order(LITTLE_ENDIAN)) < 0);
         value[0] -= 2;
         assertTrue(buffer.compareTo(wrappedBuffer(value)) > 0);
         assertTrue(buffer.compareTo(wrappedBuffer(value).order(LITTLE_ENDIAN)) > 0);
-        value[0] ++;
+        value[0]++;
 
         assertTrue(buffer.compareTo(wrappedBuffer(value, 0, 31)) > 0);
         assertTrue(buffer.compareTo(wrappedBuffer(value, 0, 31).order(LITTLE_ENDIAN)) > 0);

buffer/src/test/java/io/netty/buffer/AdvancedLeakAwareByteBufTest.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer;
+
+public class AdvancedLeakAwareByteBufTest extends SimpleLeakAwareByteBufTest {
+
+    @Override
+    protected Class<? extends ByteBuf> leakClass() {
+        return AdvancedLeakAwareByteBuf.class;
+    }
+
+    @Override
+    protected ByteBuf wrap(ByteBuf buffer) {
+        return new AdvancedLeakAwareByteBuf(buffer, new NoopResourceLeakTracker<ByteBuf>());
+    }
+}

buffer/src/test/java/io/netty/buffer/AdvancedLeakAwareCompositeByteBufTest.java

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer;
+
+public class AdvancedLeakAwareCompositeByteBufTest extends SimpleLeakAwareCompositeByteBufTest {
+
+    @Override
+    protected WrappedCompositeByteBuf wrap(CompositeByteBuf buffer) {
+        return new AdvancedLeakAwareCompositeByteBuf(buffer, new NoopResourceLeakTracker<ByteBuf>());
+    }
+
+    @Override
+    protected Class<? extends ByteBuf> leakClass() {
+        return AdvancedLeakAwareByteBuf.class;
+    }
+}

buffer/src/test/java/io/netty/buffer/NoopResourceLeakTracker.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer;
+
+import io.netty.util.ResourceLeakTracker;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+
+final class NoopResourceLeakTracker<T> extends AtomicBoolean implements ResourceLeakTracker<T> {
+
+    private static final long serialVersionUID = 7874092436796083851L;
+
+    @Override
+    public void record() {
+        // NOOP
+    }
+
+    @Override
+    public void record(Object hint) {
+        // NOOP
+    }
+
+    @Override
+    public boolean close(T trackedObject) {
+        return compareAndSet(false, true);
+    }
+}

buffer/src/test/java/io/netty/buffer/SimpleLeakAwareByteBufTest.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class SimpleLeakAwareByteBufTest extends BigEndianHeapByteBufTest {
+    private final Class<? extends ByteBuf> clazz = leakClass();
+
+    @Override
+    protected final ByteBuf newBuffer(int capacity) {
+        return wrap(super.newBuffer(capacity));
+    }
+
+    protected ByteBuf wrap(ByteBuf buffer) {
+        return new SimpleLeakAwareByteBuf(buffer, new NoopResourceLeakTracker<ByteBuf>());
+    }
+
+    protected Class<? extends ByteBuf> leakClass() {
+        return SimpleLeakAwareByteBuf.class;
+    }
+
+    @Test
+    public void testWrapSlice() {
+        assertWrapped(newBuffer(8).slice());
+    }
+
+    @Test
+    public void testWrapSlice2() {
+        assertWrapped(newBuffer(8).slice(0, 1));
+    }
+
+    @Test
+    public void testWrapReadSlice() {
+        assertWrapped(newBuffer(8).readSlice(1));
+    }
+    @Test
+    public void testWrapDuplicate() {
+        assertWrapped(newBuffer(8).duplicate());
+    }
+
+    protected final void assertWrapped(ByteBuf buf) {
+        try {
+            Assert.assertSame(clazz, buf.getClass());
+        } finally {
+            buf.release();
+        }
+    }
+}

buffer/src/test/java/io/netty/buffer/SimpleLeakAwareCompositeByteBufTest.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+public class SimpleLeakAwareCompositeByteBufTest extends WrappedCompositeByteBufTest {
+
+    private final Class<? extends ByteBuf> clazz = leakClass();
+
+    @Override
+    protected WrappedCompositeByteBuf wrap(CompositeByteBuf buffer) {
+        return new SimpleLeakAwareCompositeByteBuf(buffer, new NoopResourceLeakTracker<ByteBuf>());
+    }
+
+    protected Class<? extends ByteBuf> leakClass() {
+        return SimpleLeakAwareByteBuf.class;
+    }
+
+   @Test
+    public void testWrapSlice() {
+        assertWrapped(newBuffer(8).slice());
+    }
+
+    @Test
+    public void testWrapSlice2() {
+        assertWrapped(newBuffer(8).slice(0, 1));
+    }
+
+    @Test
+    public void testWrapReadSlice() {
+        assertWrapped(newBuffer(8).readSlice(1));
+    }
+
+    @Test
+    public void testWrapDuplicate() {
+        assertWrapped(newBuffer(8).duplicate());
+    }
+
+    protected final void assertWrapped(ByteBuf buf) {
+        try {
+            Assert.assertSame(clazz, buf.getClass());
+        } finally {
+            buf.release();
+        }
+    }
+}

buffer/src/test/java/io/netty/buffer/WrappedCompositeByteBufTest.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.buffer;
+
+public class WrappedCompositeByteBufTest extends BigEndianCompositeByteBufTest {
+
+    @Override
+    protected final ByteBuf newBuffer(int length) {
+        return wrap((CompositeByteBuf) super.newBuffer(length));
+    }
+
+    protected WrappedCompositeByteBuf wrap(CompositeByteBuf buffer) {
+        return new WrappedCompositeByteBuf(buffer);
+    }
+}

common/src/main/java/io/netty/util/HashedWheelTimer.java

@@ -91,7 +91,7 @@ public class HashedWheelTimer implements Timer {
         WORKER_STATE_UPDATER = workerStateUpdater;
     }
 
-    private final ResourceLeak leak;
+    private final ResourceLeakTracker<HashedWheelTimer> leak;
     private final Worker worker = new Worker();
     private final Thread workerThread;
 
@@ -270,7 +270,7 @@ public class HashedWheelTimer implements Timer {
         }
         workerThread = threadFactory.newThread(worker);
 
-        leak = leakDetection || !workerThread.isDaemon() ? leakDetector.open(this) : null;
+        leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;
 
         this.maxPendingTimeouts = maxPendingTimeouts;
     }
@@ -347,7 +347,8 @@ public class HashedWheelTimer implements Timer {
             WORKER_STATE_UPDATER.set(this, WORKER_STATE_SHUTDOWN);
 
             if (leak != null) {
-                leak.close();
+                boolean closed = leak.close(this);
+                assert closed;
             }
 
             return Collections.emptySet();
@@ -368,7 +369,8 @@ public class HashedWheelTimer implements Timer {
         }
 
         if (leak != null) {
-            leak.close();
+            boolean closed = leak.close(this);
+            assert closed;
         }
         return worker.unprocessedTimeouts();
     }

common/src/main/java/io/netty/util/ResourceLeak.java

@@ -16,6 +16,10 @@
 
 package io.netty.util;
 
+/**
+ * @deprecated please use {@link ResourceLeakTracker} as it may lead to false-positives.
+ */
+@Deprecated
 public interface ResourceLeak {
     /**
      * Records the caller's current stack trace so that the {@link ResourceLeakDetector} can tell where the leaked

common/src/main/java/io/netty/util/ResourceLeakDetector.java

@@ -202,8 +202,24 @@ public class ResourceLeakDetector<T> {
      * related resource is deallocated.
      *
      * @return the {@link ResourceLeak} or {@code null}
+     * @deprecated use {@link #track(Object)}
      */
+    @Deprecated
     public final ResourceLeak open(T obj) {
+        return track0(obj);
+    }
+
+    /**
+     * Creates a new {@link ResourceLeakTracker} which is expected to be closed via
+     * {@link ResourceLeakTracker#close(Object)} when the related resource is deallocated.
+     *
+     * @return the {@link ResourceLeakTracker} or {@code null}
+     */
+    public final ResourceLeakTracker<T> track(T obj) {
+        return track0(obj);
+    }
+
+    private DefaultResourceLeak track0(T obj) {
         Level level = ResourceLeakDetector.level;
         if (level == Level.DISABLED) {
             return null;
@@ -300,9 +316,13 @@ public class ResourceLeakDetector<T> {
                 "so that only a few instances are created.");
     }
 
-    private final class DefaultResourceLeak extends PhantomReference<Object> implements ResourceLeak {
+    @SuppressWarnings("deprecation")
+    private final class DefaultResourceLeak extends PhantomReference<Object> implements ResourceLeakTracker<T>,
+            ResourceLeak {
         private final String creationRecord;
         private final Deque<String> lastRecords = new ArrayDeque<String>();
+        private final int trackedHash;
+
         private int removedRecords;
 
         DefaultResourceLeak(Object referent) {
@@ -310,20 +330,33 @@ public class ResourceLeakDetector<T> {
 
             assert referent != null;
 
+            // Store the hash of the tracked object to later assert it in the close(...) method.
+            // It's important that we not store a reference to the referent as this would disallow it from
+            // be collected via the PhantomReference.
+            trackedHash = System.identityHashCode(referent);
+
             Level level = getLevel();
             if (level.ordinal() >= Level.ADVANCED.ordinal()) {
-                creationRecord = newRecord(3);
+                creationRecord = newRecord(null, 3);
             } else {
                 creationRecord = null;
             }
-
             allLeaks.put(this, LeakEntry.INSTANCE);
         }
 
         @Override
         public void record() {
+            record0(null, 3);
+        }
+
+        @Override
+        public void record(Object hint) {
+            record0(hint, 3);
+        }
+
+        private void record0(Object hint, int recordsToSkip) {
             if (creationRecord != null) {
-                String value = newRecord(2);
+                String value = newRecord(hint, recordsToSkip);
 
                 synchronized (lastRecords) {
                     int size = lastRecords.size();
@@ -344,6 +377,18 @@ public class ResourceLeakDetector<T> {
             return allLeaks.remove(this, LeakEntry.INSTANCE);
         }
 
+        @Override
+        public boolean close(T trackedObject) {
+            // Ensure that the object that was tracked is the same as the one that was passed to close(...).
+            assert trackedHash == System.identityHashCode(trackedObject);
+
+            // We need to actually do the null check of the trackedObject after we close the leak because otherwise
+            // we may get false-positives reported by the ResourceLeakDetector. This can happen as the JIT / GC may
+            // be able to figure out that we do not need the trackedObject anymore and so already enqueue it for
+            // collection before we actually get a chance to close the enclosing ResourceLeak.
+            return close() && trackedObject != null;
+        }
+
         @Override
         public String toString() {
             if (creationRecord == null) {
@@ -396,8 +441,22 @@ public class ResourceLeakDetector<T> {
             "io.netty.buffer.AdvancedLeakAwareByteBuf.recordLeakNonRefCountingOperation("
     };
 
-    static String newRecord(int recordsToSkip) {
+    static String newRecord(Object hint, int recordsToSkip) {
         StringBuilder buf = new StringBuilder(4096);
+
+        // Append the hint first if available.
+        if (hint != null) {
+            buf.append("\tHint: ");
+            // Prefer a hint string to a simple string form.
+            if (hint instanceof ResourceLeakHint) {
+                buf.append(((ResourceLeakHint) hint).toHintString());
+            } else {
+                buf.append(hint);
+            }
+            buf.append(NEWLINE);
+        }
+
+        // Append the stack trace.
         StackTraceElement[] array = new Throwable().getStackTrace();
         for (StackTraceElement e: array) {
             if (recordsToSkip > 0) {
@@ -429,7 +488,8 @@ public class ResourceLeakDetector<T> {
         static final LeakEntry INSTANCE = new LeakEntry();
         private static final int HASH = System.identityHashCode(INSTANCE);
 
-        private LeakEntry() { }
+        private LeakEntry() {
+        }
 
         @Override
         public int hashCode() {

common/src/main/java/io/netty/util/ResourceLeakHint.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2014 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.netty.util;
+
+/**
+ * A hint object that provides human-readable message for easier resource leak tracking.
+ */
+public interface ResourceLeakHint {
+    /**
+     * Returns a human-readable message that potentially enables easier resource leak tracking.
+     */
+    String toHintString();
+}

common/src/main/java/io/netty/util/ResourceLeakTracker.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.util;
+
+public interface ResourceLeakTracker<T>  {
+
+    /**
+     * Records the caller's current stack trace so that the {@link ResourceLeakDetector} can tell where the leaked
+     * resource was accessed lastly. This method is a shortcut to {@link #record(Object) record(null)}.
+     */
+    void record();
+
+    /**
+     * Records the caller's current stack trace and the specified additional arbitrary information
+     * so that the {@link ResourceLeakDetector} can tell where the leaked resource was accessed lastly.
+     */
+    void record(Object hint);
+
+    /**
+     * Close the leak so that {@link ResourceLeakTracker} does not warn about leaked resources.
+     * After this method is called a leak associated with this ResourceLeakTracker should not be reported.
+     *
+     * @return {@code true} if called first time, {@code false} if called already
+     */
+    boolean close(T trackedObject);
+}

common/src/test/java/io/netty/util/ResourceLeakDetectorTest.java

@@ -0,0 +1,172 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.util;
+
+import org.junit.Test;
+
+import java.util.ArrayDeque;
+import java.util.Queue;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class ResourceLeakDetectorTest {
+
+    @Test(timeout = 30000)
+    public void testConcurentUsage() throws Throwable {
+        final AtomicBoolean finished = new AtomicBoolean();
+        final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
+        // With 50 threads issue #6087 is reproducible on every run.
+        Thread[] threads = new Thread[50];
+        final CyclicBarrier barrier = new CyclicBarrier(threads.length);
+        for (int i = 0; i < threads.length; i++) {
+            Thread t = new Thread(new Runnable() {
+                Queue<LeakAwareResource> resources = new ArrayDeque<LeakAwareResource>(100);
+
+                @Override
+                public void run() {
+                    try {
+                        barrier.await();
+
+                        // Run 10000 times or until the test is marked as finished.
+                        for (int b = 0; b < 1000 && !finished.get(); b++) {
+
+                            // Allocate 100 LeakAwareResource per run and close them after it.
+                            for (int a = 0; a < 100; a++) {
+                                DefaultResource resource = new DefaultResource();
+                                ResourceLeakTracker<Resource> leak = DefaultResource.detector.track(resource);
+                                LeakAwareResource leakAwareResource = new LeakAwareResource(resource, leak);
+                                resources.add(leakAwareResource);
+                            }
+                            if (closeResources(true)) {
+                                finished.set(true);
+                            }
+                        }
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    } catch (Throwable e) {
+                        error.compareAndSet(null, e);
+                    } finally {
+                        // Just close all resource now without assert it to eliminate more reports.
+                        closeResources(false);
+                    }
+                }
+
+                private boolean closeResources(boolean checkClosed) {
+                    for (;;) {
+                        LeakAwareResource r = resources.poll();
+                        if (r == null) {
+                            return false;
+                        }
+                        boolean closed = r.close();
+                        if (checkClosed && !closed) {
+                            error.compareAndSet(null,
+                                    new AssertionError("ResourceLeak.close() returned 'false' but expected 'true'"));
+                            return true;
+                        }
+                    }
+                }
+            });
+            threads[i] = t;
+            t.start();
+        }
+
+        // Just wait until all threads are done.
+        for (Thread t: threads) {
+            t.join();
+        }
+
+        // Check if we had any leak reports in the ResourceLeakDetector itself
+        DefaultResource.detector.assertNoErrors();
+
+        assertNoErrors(error);
+    }
+
+    // Mimic the way how we implement our classes that should help with leak detection
+    private static final  class LeakAwareResource implements Resource {
+        private final Resource resource;
+        private final ResourceLeakTracker<Resource> leak;
+
+        LeakAwareResource(Resource resource, ResourceLeakTracker<Resource> leak) {
+            this.resource = resource;
+            this.leak = leak;
+        }
+
+        @Override
+        public boolean close() {
+            // Using ResourceLeakDetector.close(...) to prove this fixes the leak problem reported
+            // in https://github.com/netty/netty/issues/6034 .
+            //
+            // The following implementation would produce a leak:
+            //     return leak.close();
+            return leak.close(resource);
+        }
+    }
+
+    private static final class DefaultResource implements Resource {
+        // Sample every allocation
+        static final TestResourceLeakDetector<Resource> detector = new TestResourceLeakDetector<Resource>(
+                Resource.class, 1, Integer.MAX_VALUE);
+
+        @Override
+        public boolean close() {
+            return true;
+        }
+    }
+
+    private interface Resource {
+        boolean close();
+    }
+
+    private static void assertNoErrors(AtomicReference<Throwable> ref) throws Throwable {
+        Throwable error = ref.get();
+        if (error != null) {
+            throw error;
+        }
+    }
+
+    private static final class TestResourceLeakDetector<T> extends ResourceLeakDetector<T> {
+
+        private final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
+
+        TestResourceLeakDetector(Class<?> resourceType, int samplingInterval, long maxActive) {
+            super(resourceType, samplingInterval, maxActive);
+        }
+
+        @Override
+        protected void reportTracedLeak(String resourceType, String records) {
+            reportError(new AssertionError("Leak reported for '" + resourceType + "':\n" + records));
+        }
+
+        @Override
+        protected void reportUntracedLeak(String resourceType) {
+            reportError(new AssertionError("Leak reported for '" + resourceType + '\''));
+        }
+
+        @Override
+        protected void reportInstancesLeak(String resourceType) {
+            reportError(new AssertionError("Leak reported for '" + resourceType + '\''));
+        }
+
+        private void reportError(AssertionError cause) {
+            error.compareAndSet(null, cause);
+        }
+
+        void assertNoErrors() throws Throwable {
+            ResourceLeakDetectorTest.assertNoErrors(error);
+        }
+    }
+}

handler/src/main/java/io/netty/handler/ssl/ReferenceCountedOpenSslContext.java

@@ -19,9 +19,9 @@ import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.util.AbstractReferenceCounted;
 import io.netty.util.ReferenceCounted;
-import io.netty.util.ResourceLeak;
 import io.netty.util.ResourceLeakDetector;
 import io.netty.util.ResourceLeakDetectorFactory;
+import io.netty.util.ResourceLeakTracker;
 import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.StringUtil;
 import io.netty.util.internal.SystemPropertyUtil;
@@ -108,13 +108,14 @@ public abstract class ReferenceCountedOpenSslContext extends SslContext implemen
     private final int mode;
 
     // Reference Counting
-    private final ResourceLeak leak;
+    private final ResourceLeakTracker<ReferenceCountedOpenSslContext> leak;
     private final AbstractReferenceCounted refCnt = new AbstractReferenceCounted() {
         @Override
         protected void deallocate() {
             destroy();
             if (leak != null) {
-                leak.close();
+                boolean closed = leak.close(ReferenceCountedOpenSslContext.this);
+                assert closed;
             }
         }
     };
@@ -208,7 +209,7 @@ public abstract class ReferenceCountedOpenSslContext extends SslContext implemen
         if (mode != SSL.SSL_MODE_SERVER && mode != SSL.SSL_MODE_CLIENT) {
             throw new IllegalArgumentException("mode most be either SSL.SSL_MODE_SERVER or SSL.SSL_MODE_CLIENT");
         }
-        leak = leakDetection ? leakDetector.open(this) : null;
+        leak = leakDetection ? leakDetector.track(this) : null;
         this.mode = mode;
         this.clientAuth = isServer() ? checkNotNull(clientAuth, "clientAuth") : ClientAuth.NONE;
 

handler/src/main/java/io/netty/handler/ssl/ReferenceCountedOpenSslEngine.java

@@ -20,9 +20,9 @@ import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
 import io.netty.util.AbstractReferenceCounted;
 import io.netty.util.ReferenceCounted;
-import io.netty.util.ResourceLeak;
 import io.netty.util.ResourceLeakDetector;
 import io.netty.util.ResourceLeakDetectorFactory;
+import io.netty.util.ResourceLeakTracker;
 import io.netty.util.internal.EmptyArrays;
 import io.netty.util.internal.PlatformDependent;
 import io.netty.util.internal.StringUtil;
@@ -208,13 +208,14 @@ public class ReferenceCountedOpenSslEngine extends SSLEngine implements Referenc
     private volatile int destroyed;
 
     // Reference Counting
-    private final ResourceLeak leak;
+    private final ResourceLeakTracker<ReferenceCountedOpenSslEngine> leak;
     private final AbstractReferenceCounted refCnt = new AbstractReferenceCounted() {
         @Override
         protected void deallocate() {
             shutdown();
             if (leak != null) {
-                leak.close();
+                boolean closed = leak.close(ReferenceCountedOpenSslEngine.this);
+                assert closed;
             }
         }
     };
@@ -260,7 +261,7 @@ public class ReferenceCountedOpenSslEngine extends SSLEngine implements Referenc
                                   int peerPort, boolean leakDetection) {
         super(peerHost, peerPort);
         OpenSsl.ensureAvailability();
-        leak = leakDetection ? leakDetector.open(this) : null;
+        leak = leakDetection ? leakDetector.track(this) : null;
         this.alloc = checkNotNull(alloc, "alloc");
         apn = (OpenSslApplicationProtocolNegotiator) context.applicationProtocolNegotiator();
         ssl = SSL.newSSL(context.ctx, !context.isClient());