Fix memory leak on netty
MODIFY
instrumentation-api-caching/src/main/java/com/github/benmanes/caffeine/cache/CacheImplementations.java
instrumentation-api-caching/src/main/java/com/github/benmanes/caffeine/cache/CacheImplementations.java
@@ -21,5 +21,9 @@ final class CacheImplementations {
WSMS<?, ?> wsms; // cache
FSMS<?, ?> fsms; // node
+ // Weak keys, weak values
+ WI<?, ?> wi; // cache
+ FW<?, ?> fw; // node
+
private CacheImplementations() {}
}

MODIFY
instrumentation-api-caching/src/main/java/io/opentelemetry/instrumentation/api/caching/CacheBuilder.java
instrumentation-api-caching/src/main/java/io/opentelemetry/instrumentation/api/caching/CacheBuilder.java
@@ -14,6 +14,7 @@ public final class CacheBuilder {
private static final long UNSET = -1;
private boolean weakKeys;
+ private boolean weakValues;
private long maximumSize = UNSET;
private Executor executor = null;
@@ -32,6 +33,12 @@ public final class CacheBuilder {
return this;
}
+ /** Sets that values should be referenced weakly. */
+ public CacheBuilder setWeakValues() {
+ this.weakValues = true;
+ return this;
+ }
+
// Visible for testing
CacheBuilder setExecutor(Executor executor) {
this.executor = executor;
@@ -40,13 +47,16 @@ public final class CacheBuilder {
/** Returns a new {@link Cache} with the settings of this {@link CacheBuilder}. */
public <K, V> Cache<K, V> build() {
- if (weakKeys && maximumSize == UNSET) {
+ if (weakKeys && !weakValues && maximumSize == UNSET) {
return new WeakLockFreeCache<>();
}
Caffeine<?, ?> caffeine = Caffeine.newBuilder();
if (weakKeys) {
caffeine.weakKeys();
}
+ if (weakValues) {
+ caffeine.weakValues();
+ }
if (maximumSize != UNSET) {
caffeine.maximumSize(maximumSize);
}

MODIFY
instrumentation/netty/netty-4-common/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/netty/common/FutureListenerWrappers.java
instrumentation/netty/netty-4-common/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/netty/common/FutureListenerWrappers.java
@@ -11,55 +11,58 @@ import io.netty.util.concurrent.GenericProgressiveFutureListener;
import io.netty.util.concurrent.ProgressiveFuture;
import io.opentelemetry.context.Context;
import io.opentelemetry.context.Scope;
-import io.opentelemetry.javaagent.instrumentation.api.ContextStore;
+import io.opentelemetry.instrumentation.api.caching.Cache;
public final class FutureListenerWrappers {
+ // Instead of ContextStore use Cache with weak keys and weak values to store link between original
+ // listener and wrapper. ContextStore works fine when wrapper is stored in a field on original
+ // listener, but when listener class is a lambda instead of field it gets stored in a map with
+ // weak keys where original listener is key and wrapper is value. As wrapper has a strong
+ // reference to original listener this causes a memory leak.
+ // Also note that it's ok if the value is collected prior to the key, since this cache is only
+ // used to remove the wrapped listener from the netty future, and if the value is collected prior
+ // to the key, that means it's no longer used (referenced) by the netty future anyways.
+ private static final Cache<
+ GenericFutureListener<? extends Future<?>>, GenericFutureListener<? extends Future<?>>>
+ wrappers = Cache.newBuilder().setWeakKeys().setWeakValues().build();
+
@SuppressWarnings("unchecked")
- public static GenericFutureListener<? extends Future<? super Void>> wrap(
- ContextStore<GenericFutureListener, GenericFutureListener> contextStore,
- Context context,
- GenericFutureListener<? extends Future<? super Void>> delegate) {
+ public static GenericFutureListener<?> wrap(
+ Context context, GenericFutureListener<? extends Future<?>> delegate) {
if (delegate instanceof WrappedFutureListener
|| delegate instanceof WrappedProgressiveFutureListener) {
return delegate;
}
- return (GenericFutureListener<? extends Future<? super Void>>)
- contextStore.putIfAbsent(
+ return wrappers.computeIfAbsent(
delegate,
- () -> {
+ (key) -> {
if (delegate instanceof GenericProgressiveFutureListener) {
return new WrappedProgressiveFutureListener(
- context,
- (GenericProgressiveFutureListener<ProgressiveFuture<? super Void>>) delegate);
+ context, (GenericProgressiveFutureListener<ProgressiveFuture<?>>) delegate);
} else {
- return new WrappedFutureListener(
- context, (GenericFutureListener<Future<? super Void>>) delegate);
+ return new WrappedFutureListener(context, (GenericFutureListener<Future<?>>) delegate);
}
});
}
- public static GenericFutureListener<? extends Future<? super Void>> getWrapper(
- ContextStore<GenericFutureListener, GenericFutureListener> contextStore,
- GenericFutureListener<? extends Future<? super Void>> delegate) {
- GenericFutureListener<? extends Future<? super Void>> wrapper =
- (GenericFutureListener<? extends Future<? super Void>>) contextStore.get(delegate);
+ public static GenericFutureListener<? extends Future<?>> getWrapper(
+ GenericFutureListener<? extends Future<?>> delegate) {
+ GenericFutureListener<? extends Future<?>> wrapper = wrappers.get(delegate);
return wrapper == null ? delegate : wrapper;
}
- private static final class WrappedFutureListener
- implements GenericFutureListener<Future<? super Void>> {
+ private static final class WrappedFutureListener implements GenericFutureListener<Future<?>> {
private final Context context;
- private final GenericFutureListener<Future<? super Void>> delegate;
+ private final GenericFutureListener<Future<?>> delegate;
- private WrappedFutureListener(
- Context context, GenericFutureListener<Future<? super Void>> delegate) {
+ private WrappedFutureListener(Context context, GenericFutureListener<Future<?>> delegate) {
this.context = context;
this.delegate = delegate;
}
@Override
- public void operationComplete(Future<? super Void> future) throws Exception {
+ public void operationComplete(Future<?> future) throws Exception {
try (Scope ignored = context.makeCurrent()) {
delegate.operationComplete(future);
}
@@ -67,29 +70,27 @@ public final class FutureListenerWrappers {
}
private static final class WrappedProgressiveFutureListener
- implements GenericProgressiveFutureListener<ProgressiveFuture<? super Void>> {
+ implements GenericProgressiveFutureListener<ProgressiveFuture<?>> {
private final Context context;
- private final GenericProgressiveFutureListener<ProgressiveFuture<? super Void>> delegate;
+ private final GenericProgressiveFutureListener<ProgressiveFuture<?>> delegate;
private WrappedProgressiveFutureListener(
- Context context,
- GenericProgressiveFutureListener<ProgressiveFuture<? super Void>> delegate) {
+ Context context, GenericProgressiveFutureListener<ProgressiveFuture<?>> delegate) {
this.context = context;
this.delegate = delegate;
}
@Override
- public void operationProgressed(
- ProgressiveFuture<? super Void> progressiveFuture, long l, long l1) throws Exception {
+ public void operationProgressed(ProgressiveFuture<?> progressiveFuture, long l, long l1)
+ throws Exception {
try (Scope ignored = context.makeCurrent()) {
delegate.operationProgressed(progressiveFuture, l, l1);
}
}
@Override
- public void operationComplete(ProgressiveFuture<? super Void> progressiveFuture)
- throws Exception {
+ public void operationComplete(ProgressiveFuture<?> progressiveFuture) throws Exception {
try (Scope ignored = context.makeCurrent()) {
delegate.operationComplete(progressiveFuture);
}

MODIFY
instrumentation/netty/netty-4-common/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/netty/common/NettyFutureInstrumentation.java
instrumentation/netty/netty-4-common/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/netty/common/NettyFutureInstrumentation.java
@@ -17,8 +17,6 @@ import io.netty.util.concurrent.GenericFutureListener;
import io.opentelemetry.context.Context;
import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation;
import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer;
-import io.opentelemetry.javaagent.instrumentation.api.ContextStore;
-import io.opentelemetry.javaagent.instrumentation.api.InstrumentationContext;
import io.opentelemetry.javaagent.instrumentation.api.Java8BytecodeBridge;
import net.bytebuddy.asm.Advice;
import net.bytebuddy.description.type.TypeDescription;
@@ -60,11 +58,8 @@ public class NettyFutureInstrumentation implements TypeInstrumentation {
@Advice.OnMethodEnter
public static void wrapListener(
@Advice.Argument(value = 0, readOnly = false)
- GenericFutureListener<? extends Future<? super Void>> listener) {
- ContextStore<GenericFutureListener, GenericFutureListener> contextStore =
- InstrumentationContext.get(GenericFutureListener.class, GenericFutureListener.class);
- listener =
- FutureListenerWrappers.wrap(contextStore, Java8BytecodeBridge.currentContext(), listener);
+ GenericFutureListener<? extends Future<?>> listener) {
+ listener = FutureListenerWrappers.wrap(Java8BytecodeBridge.currentContext(), listener);
}
}
@@ -72,16 +67,14 @@ public class NettyFutureInstrumentation implements TypeInstrumentation {
@Advice.OnMethodEnter
public static void wrapListener(
@Advice.Argument(value = 0, readOnly = false)
- GenericFutureListener<? extends Future<? super Void>>[] listeners) {
+ GenericFutureListener<? extends Future<?>>[] listeners) {
- ContextStore<GenericFutureListener, GenericFutureListener> contextStore =
- InstrumentationContext.get(GenericFutureListener.class, GenericFutureListener.class);
Context context = Java8BytecodeBridge.currentContext();
@SuppressWarnings("unchecked")
- GenericFutureListener<? extends Future<? super Void>>[] wrappedListeners =
+ GenericFutureListener<? extends Future<?>>[] wrappedListeners =
new GenericFutureListener[listeners.length];
for (int i = 0; i < listeners.length; ++i) {
- wrappedListeners[i] = FutureListenerWrappers.wrap(contextStore, context, listeners[i]);
+ wrappedListeners[i] = FutureListenerWrappers.wrap(context, listeners[i]);
}
listeners = wrappedListeners;
}
@@ -91,10 +84,8 @@ public class NettyFutureInstrumentation implements TypeInstrumentation {
@Advice.OnMethodEnter
public static void wrapListener(
@Advice.Argument(value = 0, readOnly = false)
- GenericFutureListener<? extends Future<? super Void>> listener) {
- ContextStore<GenericFutureListener, GenericFutureListener> contextStore =
- InstrumentationContext.get(GenericFutureListener.class, GenericFutureListener.class);
- listener = FutureListenerWrappers.getWrapper(contextStore, listener);
+ GenericFutureListener<? extends Future<?>> listener) {
+ listener = FutureListenerWrappers.getWrapper(listener);
}
}
@@ -102,15 +93,13 @@ public class NettyFutureInstrumentation implements TypeInstrumentation {
@Advice.OnMethodEnter
public static void wrapListener(
@Advice.Argument(value = 0, readOnly = false)
- GenericFutureListener<? extends Future<? super Void>>[] listeners) {
+ GenericFutureListener<? extends Future<?>>[] listeners) {
- ContextStore<GenericFutureListener, GenericFutureListener> contextStore =
- InstrumentationContext.get(GenericFutureListener.class, GenericFutureListener.class);
@SuppressWarnings("unchecked")
- GenericFutureListener<? extends Future<? super Void>>[] wrappedListeners =
+ GenericFutureListener<? extends Future<?>>[] wrappedListeners =
new GenericFutureListener[listeners.length];
for (int i = 0; i < listeners.length; ++i) {
- wrappedListeners[i] = FutureListenerWrappers.getWrapper(contextStore, listeners[i]);
+ wrappedListeners[i] = FutureListenerWrappers.getWrapper(listeners[i]);
}
listeners = wrappedListeners;
}

Fix memory leak on netty (#3059)