聊聊netty的ResourceLeakDetector
序
本文主要研究一下netty的ResourceLeakDetector
LEAK异常
2019-04-02 15:23:17.026 ERROR 1 --- [reactor-http-epoll-2] io.netty.util.ResourceLeakDetector : LEAK: ByteBuf.release() was not called before it's garbage-collected. See http://netty.io/wiki/reference-counted-objects.html for more information.
Recent access records:
#1:
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:286)
io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:253)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:427)
io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:328)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
java.base/java.lang.Thread.run(Thread.java:835)
#2:
io.netty.buffer.AdvancedLeakAwareByteBuf.forEachByte(AdvancedLeakAwareByteBuf.java:670)
io.netty.handler.codec.http.HttpObjectDecoder$HeaderParser.parse(HttpObjectDecoder.java:801)
io.netty.handler.codec.http.HttpObjectDecoder.readHeaders(HttpObjectDecoder.java:601)
io.netty.handler.codec.http.HttpObjectDecoder.decode(HttpObjectDecoder.java:227)
io.netty.handler.codec.http.HttpClientCodec$Decoder.decode(HttpClientCodec.java:202)
io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:502)
io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:441)
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:278)
io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:253)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:427)
io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:328)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
java.base/java.lang.Thread.run(Thread.java:835)
#3:
io.netty.buffer.AdvancedLeakAwareByteBuf.forEachByte(AdvancedLeakAwareByteBuf.java:670)
io.netty.handler.codec.http.HttpObjectDecoder$HeaderParser.parse(HttpObjectDecoder.java:801)
io.netty.handler.codec.http.HttpObjectDecoder.readHeaders(HttpObjectDecoder.java:581)
io.netty.handler.codec.http.HttpObjectDecoder.decode(HttpObjectDecoder.java:227)
io.netty.handler.codec.http.HttpClientCodec$Decoder.decode(HttpClientCodec.java:202)
io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:502)
io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:441)
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:278)
io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:253)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:427)
io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:328)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
java.base/java.lang.Thread.run(Thread.java:835)
#4:
io.netty.buffer.AdvancedLeakAwareByteBuf.forEachByte(AdvancedLeakAwareByteBuf.java:670)
io.netty.handler.codec.http.HttpObjectDecoder$HeaderParser.parse(HttpObjectDecoder.java:801)
io.netty.handler.codec.http.HttpObjectDecoder$LineParser.parse(HttpObjectDecoder.java:850)
io.netty.handler.codec.http.HttpObjectDecoder.decode(HttpObjectDecoder.java:208)
io.netty.handler.codec.http.HttpClientCodec$Decoder.decode(HttpClientCodec.java:202)
io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:502)
io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:441)
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:278)
io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:253)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:427)
io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:328)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
java.base/java.lang.Thread.run(Thread.java:835)
#5:
io.netty.buffer.AdvancedLeakAwareByteBuf.getUnsignedByte(AdvancedLeakAwareByteBuf.java:160)
io.netty.handler.codec.http.HttpObjectDecoder.skipControlCharacters(HttpObjectDecoder.java:566)
io.netty.handler.codec.http.HttpObjectDecoder.decode(HttpObjectDecoder.java:202)
io.netty.handler.codec.http.HttpClientCodec$Decoder.decode(HttpClientCodec.java:202)
io.netty.handler.codec.ByteToMessageDecoder.decodeRemovalReentryProtection(ByteToMessageDecoder.java:502)
io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:441)
io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:278)
io.netty.channel.CombinedChannelDuplexHandler.channelRead(CombinedChannelDuplexHandler.java:253)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:427)
io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:328)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
java.base/java.lang.Thread.run(Thread.java:835)
#6:
Hint: 'reactor.left.httpCodec' will handle the message from this point.
io.netty.channel.DefaultChannelPipeline.touch(DefaultChannelPipeline.java:116)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:345)
io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:340)
io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:362)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:427)
io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:328)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
java.base/java.lang.Thread.run(Thread.java:835)
#7:
Hint: 'DefaultChannelPipeline$HeadContext#0' will handle the message from this point.
io.netty.channel.DefaultChannelPipeline.touch(DefaultChannelPipeline.java:116)
io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:345)
io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930)
io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:799)
io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:427)
io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:328)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
java.base/java.lang.Thread.run(Thread.java:835)
Created at:
io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:339)
io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:185)
io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:176)
io.netty.channel.unix.PreferredDirectByteBufAllocator.ioBuffer(PreferredDirectByteBufAllocator.java:53)
io.netty.channel.DefaultMaxMessagesRecvByteBufAllocator$MaxMessageHandle.allocate(DefaultMaxMessagesRecvByteBufAllocator.java:114)
io.netty.channel.epoll.EpollRecvByteAllocatorHandle.allocate(EpollRecvByteAllocatorHandle.java:77)
io.netty.channel.epoll.AbstractEpollStreamChannel$EpollStreamUnsafe.epollInReady(AbstractEpollStreamChannel.java:784)
io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:427)
io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:328)
io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905)
java.base/java.lang.Thread.run(Thread.java:835)
: 9 leak records were discarded because the leak record count is targeted to 4. Use system property io.netty.leakDetection.targetRecords to increase the limit.
ResourceLeakDetector
netty-common-4.1.33.Final-sources.jar!/io/netty/util/ResourceLeakDetector.java
public class ResourceLeakDetector<T> {
private static final String PROP_LEVEL_OLD = "io.netty.leakDetectionLevel";
private static final String PROP_LEVEL = "io.netty.leakDetection.level";
private static final Level DEFAULT_LEVEL = Level.SIMPLE;
private static final String PROP_TARGET_RECORDS = "io.netty.leakDetection.targetRecords";
private static final int DEFAULT_TARGET_RECORDS = 4;
private static final String PROP_SAMPLING_INTERVAL = "io.netty.leakDetection.samplingInterval";
// There is a minor performance benefit in TLR if this is a power of 2.
private static final int DEFAULT_SAMPLING_INTERVAL = 128;
private static final int TARGET_RECORDS;
static final int SAMPLING_INTERVAL;
/**
* Represents the level of resource leak detection.
*/
public enum Level {
/**
* Disables resource leak detection.
*/
DISABLED,
/**
* Enables simplistic sampling resource leak detection which reports there is a leak or not,
* at the cost of small overhead (default).
*/
SIMPLE,
/**
* Enables advanced sampling resource leak detection which reports where the leaked object was accessed
* recently at the cost of high overhead.
*/
ADVANCED,
/**
* Enables paranoid resource leak detection which reports where the leaked object was accessed recently,
* at the cost of the highest possible overhead (for testing purposes only).
*/
PARANOID;
/**
* Returns level based on string value. Accepts also string that represents ordinal number of enum.
*
* @param levelStr - level string : DISABLED, SIMPLE, ADVANCED, PARANOID. Ignores case.
* @return corresponding level or SIMPLE level in case of no match.
*/
static Level parseLevel(String levelStr) {
String trimmedLevelStr = levelStr.trim();
for (Level l : values()) {
if (trimmedLevelStr.equalsIgnoreCase(l.name()) || trimmedLevelStr.equals(String.valueOf(l.ordinal()))) {
return l;
}
}
return DEFAULT_LEVEL;
}
}
private static Level level;
private static final InternalLogger logger = InternalLoggerFactory.getInstance(ResourceLeakDetector.class);
static {
final boolean disabled;
if (SystemPropertyUtil.get("io.netty.noResourceLeakDetection") != null) {
disabled = SystemPropertyUtil.getBoolean("io.netty.noResourceLeakDetection", false);
logger.debug("-Dio.netty.noResourceLeakDetection: {}", disabled);
logger.warn(
"-Dio.netty.noResourceLeakDetection is deprecated. Use '-D{}={}' instead.",
PROP_LEVEL, DEFAULT_LEVEL.name().toLowerCase());
} else {
disabled = false;
}
Level defaultLevel = disabled? Level.DISABLED : DEFAULT_LEVEL;
// First read old property name
String levelStr = SystemPropertyUtil.get(PROP_LEVEL_OLD, defaultLevel.name());
// If new property name is present, use it
levelStr = SystemPropertyUtil.get(PROP_LEVEL, levelStr);
Level level = Level.parseLevel(levelStr);
TARGET_RECORDS = SystemPropertyUtil.getInt(PROP_TARGET_RECORDS, DEFAULT_TARGET_RECORDS);
SAMPLING_INTERVAL = SystemPropertyUtil.getInt(PROP_SAMPLING_INTERVAL, DEFAULT_SAMPLING_INTERVAL);
ResourceLeakDetector.level = level;
if (logger.isDebugEnabled()) {
logger.debug("-D{}: {}", PROP_LEVEL, level.name().toLowerCase());
logger.debug("-D{}: {}", PROP_TARGET_RECORDS, TARGET_RECORDS);
}
}
/**
* @deprecated Use {@link #setLevel(Level)} instead.
*/
@Deprecated
public static void setEnabled(boolean enabled) {
setLevel(enabled? Level.SIMPLE : Level.DISABLED);
}
/**
* Returns {@code true} if resource leak detection is enabled.
*/
public static boolean isEnabled() {
return getLevel().ordinal() > Level.DISABLED.ordinal();
}
/**
* Sets the resource leak detection level.
*/
public static void setLevel(Level level) {
if (level == null) {
throw new NullPointerException("level");
}
ResourceLeakDetector.level = level;
}
/**
* Returns the current resource leak detection level.
*/
public static Level getLevel() {
return level;
}
/** the collection of active resources */
private final Set<DefaultResourceLeak<?>> allLeaks =
Collections.newSetFromMap(new ConcurrentHashMap<DefaultResourceLeak<?>, Boolean>());
private final ReferenceQueue<Object> refQueue = new ReferenceQueue<Object>();
private final ConcurrentMap<String, Boolean> reportedLeaks = PlatformDependent.newConcurrentHashMap();
private final String resourceType;
private final int samplingInterval;
//......
/**
* 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}
*/
@SuppressWarnings("unchecked")
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;
}
if (level.ordinal() < Level.PARANOID.ordinal()) {
if ((PlatformDependent.threadLocalRandom().nextInt(samplingInterval)) == 0) {
reportLeak();
return new DefaultResourceLeak(obj, refQueue, allLeaks);
}
return null;
}
reportLeak();
return new DefaultResourceLeak(obj, refQueue, allLeaks);
}
private void reportLeak() {
if (!logger.isErrorEnabled()) {
clearRefQueue();
return;
}
// Detect and report previous leaks.
for (;;) {
@SuppressWarnings("unchecked")
DefaultResourceLeak ref = (DefaultResourceLeak) refQueue.poll();
if (ref == null) {
break;
}
if (!ref.dispose()) {
continue;
}
String records = ref.toString();
if (reportedLeaks.putIfAbsent(records, Boolean.TRUE) == null) {
if (records.isEmpty()) {
reportUntracedLeak(resourceType);
} else {
reportTracedLeak(resourceType, records);
}
}
}
}
/**
* This method is called when a traced leak is detected. It can be overridden for tracking how many times leaks
* have been detected.
*/
protected void reportTracedLeak(String resourceType, String records) {
logger.error(
"LEAK: {}.release() was not called before it's garbage-collected. " +
"See http://netty.io/wiki/reference-counted-objects.html for more information.{}",
resourceType, records);
}
/**
* This method is called when an untraced leak is detected. It can be overridden for tracking how many times leaks
* have been detected.
*/
protected void reportUntracedLeak(String resourceType) {
logger.error("LEAK: {}.release() was not called before it's garbage-collected. " +
"Enable advanced leak reporting to find out where the leak occurred. " +
"To enable advanced leak reporting, " +
"specify the JVM option '-D{}={}' or call {}.setLevel() " +
"See http://netty.io/wiki/reference-counted-objects.html for more information.",
resourceType, PROP_LEVEL, Level.ADVANCED.name().toLowerCase(), simpleClassName(this));
}
//......
}
- ResourceLeakDetector使用Level枚举定义了四种不同的leak detection级别，分别是DISABLED、SIMPLE、ADVANCED、PARANOID；默认level为SIMPLE；可以使用-Dio.netty.leakDetection.level=advanced来进行设置
-
ResourceLeakDetector的静态代码块会读取io.netty.noResourceLeakDetection系统属性，如果显示设置为false，则变更defaultLevel为DISABLED；如果没有设置，则默认disabled为false，defaultLevel为SIMPLE；ResourceLeakDetector还有TARGET_RECORDS(
io.netty.leakDetection.targetRecords)及SAMPLING_INTERVAL(io.netty.leakDetection.samplingInterval)两个属性，其中targetRecords默认为4，samplingInterval默认为128 -
ResourceLeakDetector提供了track方法用于创建ResourceLeakTracker；track方法内部调用track0方法，如果level为PARANOID则立即调用reportLeak，创建DefaultResourceLeak，否则利用随机数来判断(
PlatformDependent.threadLocalRandom().nextInt(samplingInterval)) == 0)是否调用reportLeak并创建DefaultResourceLeak；reportLeak方法有个for循环，不断从refQueue取DefaultResourceLeak，然后调用reportUntracedLeak或者reportTracedLeak进行error
DefaultResourceLeak
netty-common-4.1.33.Final-sources.jar!/io/netty/util/ResourceLeakDetector.java
private static final class DefaultResourceLeak<T>
extends WeakReference<Object> implements ResourceLeakTracker<T>, ResourceLeak {
@SuppressWarnings("unchecked") // generics and updaters do not mix.
private static final AtomicReferenceFieldUpdater<DefaultResourceLeak<?>, Record> headUpdater =
(AtomicReferenceFieldUpdater)
AtomicReferenceFieldUpdater.newUpdater(DefaultResourceLeak.class, Record.class, "head");
@SuppressWarnings("unchecked") // generics and updaters do not mix.
private static final AtomicIntegerFieldUpdater<DefaultResourceLeak<?>> droppedRecordsUpdater =
(AtomicIntegerFieldUpdater)
AtomicIntegerFieldUpdater.newUpdater(DefaultResourceLeak.class, "droppedRecords");
@SuppressWarnings("unused")
private volatile Record head;
@SuppressWarnings("unused")
private volatile int droppedRecords;
private final Set<DefaultResourceLeak<?>> allLeaks;
private final int trackedHash;
DefaultResourceLeak(
Object referent,
ReferenceQueue<Object> refQueue,
Set<DefaultResourceLeak<?>> allLeaks) {
super(referent, refQueue);
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 WeakReference.
trackedHash = System.identityHashCode(referent);
allLeaks.add(this);
// Create a new Record so we always have the creation stacktrace included.
headUpdater.set(this, new Record(Record.BOTTOM));
this.allLeaks = allLeaks;
}
@Override
public void record() {
record0(null);
}
@Override
public void record(Object hint) {
record0(hint);
}
/**
* This method works by exponentially backing off as more records are present in the stack. Each record has a
* 1 / 2^n chance of dropping the top most record and replacing it with itself. This has a number of convenient
* properties:
*
* <ol>
* <li> The current record is always recorded. This is due to the compare and swap dropping the top most
* record, rather than the to-be-pushed record.
* <li> The very last access will always be recorded. This comes as a property of 1.
* <li> It is possible to retain more records than the target, based upon the probability distribution.
* <li> It is easy to keep a precise record of the number of elements in the stack, since each element has to
* know how tall the stack is.
* </ol>
*
* In this particular implementation, there are also some advantages. A thread local random is used to decide
* if something should be recorded. This means that if there is a deterministic access pattern, it is now
* possible to see what other accesses occur, rather than always dropping them. Second, after
* {@link #TARGET_RECORDS} accesses, backoff occurs. This matches typical access patterns,
* where there are either a high number of accesses (i.e. a cached buffer), or low (an ephemeral buffer), but
* not many in between.
*
* The use of atomics avoids serializing a high number of accesses, when most of the records will be thrown
* away. High contention only happens when there are very few existing records, which is only likely when the
* object isn't shared! If this is a problem, the loop can be aborted and the record dropped, because another
* thread won the race.
*/
private void record0(Object hint) {
// Check TARGET_RECORDS > 0 here to avoid similar check before remove from and add to lastRecords
if (TARGET_RECORDS > 0) {
Record oldHead;
Record prevHead;
Record newHead;
boolean dropped;
do {
if ((prevHead = oldHead = headUpdater.get(this)) == null) {
// already closed.
return;
}
final int numElements = oldHead.pos + 1;
if (numElements >= TARGET_RECORDS) {
final int backOffFactor = Math.min(numElements - TARGET_RECORDS, 30);
if (dropped = PlatformDependent.threadLocalRandom().nextInt(1 << backOffFactor) != 0) {
prevHead = oldHead.next;
}
} else {
dropped = false;
}
newHead = hint != null ? new Record(prevHead, hint) : new Record(prevHead);
} while (!headUpdater.compareAndSet(this, oldHead, newHead));
if (dropped) {
droppedRecordsUpdater.incrementAndGet(this);
}
}
}
boolean dispose() {
clear();
return allLeaks.remove(this);
}
@Override
public boolean close() {
if (allLeaks.remove(this)) {
// Call clear so the reference is not even enqueued.
clear();
headUpdater.set(this, null);
return true;
}
return false;
}
@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);
try {
return close();
} finally {
// This method will do `synchronized(trackedObject)` and we should be sure this will not cause deadlock.
// It should not, because somewhere up the callstack should be a (successful) `trackedObject.release`,
// therefore it is unreasonable that anyone else, anywhere, is holding a lock on the trackedObject.
// (Unreasonable but possible, unfortunately.)
reachabilityFence0(trackedObject);
}
}
/**
* Ensures that the object referenced by the given reference remains
* <a href="package-summary.html#reachability"><em>strongly reachable</em></a>,
* regardless of any prior actions of the program that might otherwise cause
* the object to become unreachable; thus, the referenced object is not
* reclaimable by garbage collection at least until after the invocation of
* this method.
*
* <p> Recent versions of the JDK have a nasty habit of prematurely deciding objects are unreachable.
* see: https://stackoverflow.com/questions/26642153/finalize-called-on-strongly-reachable-object-in-java-8
* The Java 9 method Reference.reachabilityFence offers a solution to this problem.
*
* <p> This method is always implemented as a synchronization on {@code ref}, not as
* {@code Reference.reachabilityFence} for consistency across platforms and to allow building on JDK 6-8.
* <b>It is the caller's responsibility to ensure that this synchronization will not cause deadlock.</b>
*
* @param ref the reference. If {@code null}, this method has no effect.
* @see java.lang.ref.Reference#reachabilityFence
*/
private static void reachabilityFence0(Object ref) {
if (ref != null) {
// Empty synchronized is ok: https://stackoverflow.com/a/31933260/1151521
synchronized (ref) { }
}
}
@Override
public String toString() {
Record oldHead = headUpdater.getAndSet(this, null);
if (oldHead == null) {
// Already closed
return EMPTY_STRING;
}
final int dropped = droppedRecordsUpdater.get(this);
int duped = 0;
int present = oldHead.pos + 1;
// Guess about 2 kilobytes per stack trace
StringBuilder buf = new StringBuilder(present * 2048).append(NEWLINE);
buf.append("Recent access records: ").append(NEWLINE);
int i = 1;
Set<String> seen = new HashSet<String>(present);
for (; oldHead != Record.BOTTOM; oldHead = oldHead.next) {
String s = oldHead.toString();
if (seen.add(s)) {
if (oldHead.next == Record.BOTTOM) {
buf.append("Created at:").append(NEWLINE).append(s);
} else {
buf.append('#').append(i++).append(':').append(NEWLINE).append(s);
}
} else {
duped++;
}
}
if (duped > 0) {
buf.append(": ")
.append(duped)
.append(" leak records were discarded because they were duplicates")
.append(NEWLINE);
}
if (dropped > 0) {
buf.append(": ")
.append(dropped)
.append(" leak records were discarded because the leak record count is targeted to ")
.append(TARGET_RECORDS)
.append(". Use system property ")
.append(PROP_TARGET_RECORDS)
.append(" to increase the limit.")
.append(NEWLINE);
}
buf.setLength(buf.length() - NEWLINE.length());
return buf.toString();
}
}
-
DefaultResourceLeak是ResourceLeakDetector定义的私有静态类，它继承了WeakReference类，同时实现了ResourceLeakTracker(
定义了record、close方法)接口；record方法内部调用的是record0方法，它会更新newHead为新的Record；close方法会移除allLeaks，allLeaks由ResourceLeakDetector创建DefaultResourceLeak时传入，每创建一个DefaultResourceLeak，DefaultResourceLeak会把自己加入到allLeaks中
SimpleLeakAwareByteBuf
netty-netty-4.1.33.Final/buffer/src/main/java/io/netty/buffer/SimpleLeakAwareByteBuf.java
class SimpleLeakAwareByteBuf extends WrappedByteBuf {
/**
* 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 wrapped, ByteBuf trackedByteBuf, ResourceLeakTracker<ByteBuf> leak) {
super(wrapped);
this.trackedByteBuf = ObjectUtil.checkNotNull(trackedByteBuf, "trackedByteBuf");
this.leak = ObjectUtil.checkNotNull(leak, "leak");
}
SimpleLeakAwareByteBuf(ByteBuf wrapped, ResourceLeakTracker<ByteBuf> leak) {
this(wrapped, wrapped, leak);
}
//......
@Override
public boolean release() {
if (super.release()) {
closeLeak();
return true;
}
return false;
}
@Override
public 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;
}
private ByteBuf unwrappedDerived(ByteBuf derived) {
// We only need to unwrap SwappedByteBuf implementations as these will be the only ones that may end up in
// the AbstractLeakAwareByteBuf implementations beside slices / duplicates and "real" buffers.
ByteBuf unwrappedDerived = unwrapSwapped(derived);
if (unwrappedDerived instanceof AbstractPooledDerivedByteBuf) {
// Update the parent to point to this buffer so we correctly close the ResourceLeakTracker.
((AbstractPooledDerivedByteBuf) unwrappedDerived).parent(this);
ResourceLeakTracker<ByteBuf> newLeak = AbstractByteBuf.leakDetector.track(derived);
if (newLeak == null) {
// No leak detection, just return the derived buffer.
return derived;
}
return newLeakAwareByteBuf(derived, newLeak);
}
return newSharedLeakAwareByteBuf(derived);
}
//......
}
- SimpleLeakAwareByteBuf继承了WrappedByteBuf，它的构造器要求传入ResourceLeakTracker
- SimpleLeakAwareByteBuf覆盖了WrappedByteBuf的retainedSlice、retainedDuplicate、readRetainedSlice方法，它们内部都会调用unwrappedDerived方法，unwrappedDerived方法在unwrappedDerived对象是AbstractPooledDerivedByteBuf类型时会调用AbstractByteBuf.leakDetector.track进行track
- SimpleLeakAwareByteBuf也覆盖了WrappedByteBuf的release方法，在调用父类release成功时会再调用closeLeak方法，使用leak.close(trackedByteBuf)来释放trackedByteBuf
AdvancedLeakAwareByteBuf
netty-netty-4.1.33.Final/buffer/src/main/java/io/netty/buffer/AdvancedLeakAwareByteBuf.java
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;
private static final InternalLogger logger = InternalLoggerFactory.getInstance(AdvancedLeakAwareByteBuf.class);
static {
ACQUIRE_AND_RELEASE_ONLY = SystemPropertyUtil.getBoolean(PROP_ACQUIRE_AND_RELEASE_ONLY, false);
if (logger.isDebugEnabled()) {
logger.debug("-D{}: {}", PROP_ACQUIRE_AND_RELEASE_ONLY, ACQUIRE_AND_RELEASE_ONLY);
}
ResourceLeakDetector.addExclusions(
AdvancedLeakAwareByteBuf.class, "touch", "recordLeakNonRefCountingOperation");
}
AdvancedLeakAwareByteBuf(ByteBuf buf, ResourceLeakTracker<ByteBuf> leak) {
super(buf, leak);
}
AdvancedLeakAwareByteBuf(ByteBuf wrapped, ByteBuf trackedByteBuf, ResourceLeakTracker<ByteBuf> leak) {
super(wrapped, trackedByteBuf, leak);
}
static void recordLeakNonRefCountingOperation(ResourceLeakTracker<ByteBuf> leak) {
if (!ACQUIRE_AND_RELEASE_ONLY) {
leak.record();
}
}
//......
@Override
public ByteBuf order(ByteOrder endianness) {
recordLeakNonRefCountingOperation(leak);
return super.order(endianness);
}
@Override
public ByteBuf slice() {
recordLeakNonRefCountingOperation(leak);
return super.slice();
}
@Override
public ByteBuf slice(int index, int length) {
recordLeakNonRefCountingOperation(leak);
return super.slice(index, length);
}
//......
@Override
public ByteBuf retain() {
leak.record();
return super.retain();
}
@Override
public ByteBuf retain(int increment) {
leak.record();
return super.retain(increment);
}
@Override
public boolean release() {
leak.record();
return super.release();
}
@Override
public boolean release(int decrement) {
leak.record();
return super.release(decrement);
}
@Override
public ByteBuf touch() {
leak.record();
return this;
}
@Override
public ByteBuf touch(Object hint) {
leak.record(hint);
return this;
}
//......
}
- AdvancedLeakAwareByteBuf继承了SimpleLeakAwareByteBuf，它对方法进行了覆盖，这些覆盖的方法要么内部通过recordLeakNonRefCountingOperation调用leak.record，要么直接调用leak.record
小结
-
ResourceLeakDetector使用Level枚举定义了四种不同的leak detection级别，分别是DISABLED、SIMPLE、ADVANCED、PARANOID；默认level为SIMPLE；可以使用-Dio.netty.leakDetection.level=advanced来进行设置；ResourceLeakDetector还有TARGET_RECORDS(
io.netty.leakDetection.targetRecords)及SAMPLING_INTERVAL(io.netty.leakDetection.samplingInterval)两个属性，其中targetRecords默认为4，samplingInterval默认为128 -
ResourceLeakDetector提供了track方法用于创建ResourceLeakTracker；track方法内部调用track0方法，如果level为PARANOID则立即调用reportLeak，创建DefaultResourceLeak，否则利用随机数来判断(
PlatformDependent.threadLocalRandom().nextInt(samplingInterval)) == 0)是否调用reportLeak并创建DefaultResourceLeak；reportLeak方法有个for循环，不断从refQueue取DefaultResourceLeak，然后调用reportUntracedLeak或者reportTracedLeak进行error -
DefaultResourceLeak是ResourceLeakDetector定义的私有静态类，它继承了WeakReference类，同时实现了ResourceLeakTracker(
定义了record、close方法)接口；record方法内部调用的是record0方法，它会更新newHead为新的Record；close方法会移除allLeaks，allLeaks由ResourceLeakDetector创建DefaultResourceLeak时传入，每创建一个DefaultResourceLeak，DefaultResourceLeak会把自己加入到allLeaks中 - SimpleLeakAwareByteBuf继承了WrappedByteBuf，它的构造器要求传入ResourceLeakTracker；SimpleLeakAwareByteBuf覆盖了WrappedByteBuf的retainedSlice、retainedDuplicate、readRetainedSlice方法，它们内部都会调用unwrappedDerived方法，unwrappedDerived方法在unwrappedDerived对象是AbstractPooledDerivedByteBuf类型时会调用AbstractByteBuf.leakDetector.track进行track；SimpleLeakAwareByteBuf也覆盖了WrappedByteBuf的release方法，在调用父类release成功时会再调用closeLeak方法，使用leak.close(trackedByteBuf)来释放trackedByteBuf
- AdvancedLeakAwareByteBuf继承了SimpleLeakAwareByteBuf，它对方法进行了覆盖，这些覆盖的方法要么内部通过recordLeakNonRefCountingOperation调用leak.record，要么直接调用leak.record；另外有SimpleLeakAwareCompositeByteBuf与AdvancedLeakAwareCompositeByteBuf，它们对leak detect的支持类似SimpleLeakAwareByteBuf与AdvancedLeakAwareByteBuf
doc
- Netty 的资源泄露探测机制
- A Netty ByteBuf Memory Leak Story and the Lessons Learned
- In 4.0.23.Final, Seeing io.netty.util.ResourceLeakDetector - LEAK: ByteBuf.release() was not called before it's garbage-collected #2774
正文到此结束
- 本文标签： rmi tar FIT parse client Atom mmap Reactor NSA Java 9 build queue Collection synchronized https IDE HTML final message IO JVM ip UI App volatile ACE executor equals 2019 db map CTO list http cache ConcurrentHashMap rand unix cat CEO value entity ORM HashMap tag update find 代码 Collections src Action bug HashSet java swap Netty DOM dist Property id stream
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
近期评论
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
搞个gitee的项目
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
