CallTree recycling (#1592)

Author: eyalkoren

CHANGELOG.asciidoc

@@ -43,9 +43,10 @@ endif::[]
 * Serialize all stack trace frames when setting `stack_trace_limit=-1` instead of none - {pull}1571[#1571]
 * Fix `UnsupportedOperationException` when calling `ServletContext.getClassLoader()` - {pull}1576[#1576]
 * Fix improper request body capturing - {pull}1579[#1579]
-* avoid `NullPointerException` due to null return values instrumentation advices - {pull}1601[#1601]
+* Avoid `NullPointerException` due to null return values instrumentation advices - {pull}1601[#1601]
 * Update async-profiler to 1.8.3 {pull}1602[1602]
 * Use null-safe data structures to avoid `NullPointerException` {pull}1597[1597]
+* Fix memory leak in sampling profiler mechanism - {pull}1592[#1592]
 
 [float]
 ===== Refactors

apm-agent-plugins/apm-profiling-plugin/src/main/java/co/elastic/apm/agent/profiler/CallTree.java

@@ -436,7 +436,16 @@ private void fillStackTrace(List<StackFrame> stackTrace) {
         }
     }
 
-    public void recycle(ObjectPool<CallTree> pool) {
+    /**
+     * Recycles this subtree to the provided pool recursively.
+     * Note that this method ends by recycling {@code this} node (i.e. - this subtree root), which means that
+     * <b>the caller of this method should make sure that no reference to this object is held anywhere</b>.
+     * <p>ALSO NOTE: MAKE SURE NOT TO CALL THIS METHOD FOR {@link CallTree.Root} INSTANCES.</p>
+     *
+     * @param pool the pool to which all subtree nodes are to be recycled
+     */
+    public final void recycle(ObjectPool<CallTree> pool) {
+        assert !(this instanceof Root);
         List<CallTree> children = this.children;
         for (int i = 0, size = children.size(); i < size; i++) {
             children.get(i).recycle(pool);
@@ -719,11 +728,21 @@ public long getEpochMicros(long nanoTime) {
             return rootContext.getClock().getEpochMicros(nanoTime);
         }
 
-        public void recycle(ObjectPool<CallTree> pool) {
+        /**
+         * Recycles this tree to the provided pools.
+         * First, all child subtrees are recycled recursively to the children pool.
+         * Then, {@code this} root node is recycled to the root pool. This means that <b>the caller of this method
+         * should make sure that no reference to this root object is held anywhere</b>.
+         *
+         * @param childrenPool object pool for all non-root nodes
+         * @param rootPool     object pool for root nodes
+         */
+        public void recycle(ObjectPool<CallTree> childrenPool, ObjectPool<CallTree.Root> rootPool) {
             List<CallTree> children = getChildren();
             for (int i = 0, size = children.size(); i < size; i++) {
-                children.get(i).recycle(pool);
+                children.get(i).recycle(childrenPool);
             }
+            rootPool.recycle(this);
         }
 
         public void end(ObjectPool<CallTree> pool, long minDurationNs) {
@@ -733,8 +752,11 @@ public void end(ObjectPool<CallTree> pool, long minDurationNs) {
         @Override
         public void resetState() {
             super.resetState();
+            rootContext.resetState();
             activeSpan = null;
             activationTimestamp = -1;
+            Arrays.fill(activeSpanSerialized, (byte) 0);
+            previousTopOfStack = null;
             topOfStack = null;
             activeSet.clear();
         }

apm-agent-plugins/apm-profiling-plugin/src/main/java/co/elastic/apm/agent/profiler/SamplingProfiler.java

@@ -83,12 +83,12 @@
  * </p>
  * <p>
  * The {@link #onActivation} and {@link #onDeactivation} methods are called by {@link ProfilingActivationListener}
- * which register an {@link ActivationEvent} in to a {@linkplain #eventBuffer ring buffer} whenever a {@link Span}
+ * which register an {@link ActivationEvent} to a {@linkplain #eventBuffer ring buffer} whenever a {@link Span}
  * gets {@link Span#activate()}d or {@link Span#deactivate()}d while a {@linkplain #profilingSessionOngoing profiling session is ongoing}.
  * A background thread consumes the {@link ActivationEvent}s and writes them to a {@linkplain #activationEventsBuffer direct buffer}
  * which is flushed to a {@linkplain #activationEventsFileChannel file}.
  * That is necessary because within a profiling session (which lasts 10s by default) there may be many more {@link ActivationEvent}s
- * than the ring buffer can hold {@link #RING_BUFFER_SIZE}.
+ * than the ring buffer {@link #RING_BUFFER_SIZE can hold}.
  * The file can hold {@link #ACTIVATION_EVENTS_IN_FILE} events and each is {@link ActivationEvent#SERIALIZED_SIZE} in size.
  * This process is completely garbage free thanks to the {@link RingBuffer} acting as an object pool for {@link ActivationEvent}s.
  * </p>
@@ -335,12 +335,15 @@ public void run() {
         if (!config.isProfilingEnabled() || !tracer.isRunning()) {
             if (jfrParser != null) {
                 jfrParser = null;
-                rootPool.clear();
-                callTreePool.clear();
             }
             if (!scheduler.isShutdown()) {
                 scheduler.schedule(this, config.getProfilingInterval().getMillis(), TimeUnit.MILLISECONDS);
             }
+            try {
+                clear();
+            } catch (Throwable throwable) {
+                logger.error("Error while trying to clear profiler constructs", throwable);
+            }
             return;
         }
 
@@ -391,6 +394,8 @@ private void profile(TimeDuration sampleRate, TimeDuration profilingDuration) th
             if (!profiledThreads.isEmpty()) {
                 restoreFilterState(asyncProfiler);
             }
+            // Doesn't need to be atomic as this field is being updated only by a single thread
+            //noinspection NonAtomicOperationOnVolatileField
             profilingSessions++;
 
             // When post-processing is disabled activation events are ignored, but we still need to invoke this method
@@ -418,17 +423,22 @@ private void profile(TimeDuration sampleRate, TimeDuration profilingDuration) th
      * we have to tell async-profiler which threads it should profile after re-starting it.
      */
     private void restoreFilterState(AsyncProfiler asyncProfiler) {
-        threadMatcher.forEachThread(new ThreadMatcher.NonCapturingPredicate<Thread, Long2ObjectHashMap<?>.KeySet>() {
-            @Override
-            public boolean test(Thread thread, Long2ObjectHashMap<?>.KeySet profiledThreads) {
-                return profiledThreads.contains(thread.getId());
-            }
-        }, profiledThreads.keySet(), new ThreadMatcher.NonCapturingConsumer<Thread, AsyncProfiler>() {
-            @Override
-            public void accept(Thread thread, AsyncProfiler asyncProfiler) {
-                asyncProfiler.enableProfilingThread(thread);
-            }
-        }, asyncProfiler);
+        threadMatcher.forEachThread(
+            new ThreadMatcher.NonCapturingPredicate<Thread, Long2ObjectHashMap<?>.KeySet>() {
+                @Override
+                public boolean test(Thread thread, Long2ObjectHashMap<?>.KeySet profiledThreads) {
+                    return profiledThreads.contains(thread.getId());
+                }
+            },
+            profiledThreads.keySet(),
+            new ThreadMatcher.NonCapturingConsumer<Thread, AsyncProfiler>() {
+                @Override
+                public void accept(Thread thread, AsyncProfiler asyncProfiler) {
+                    asyncProfiler.enableProfilingThread(thread);
+                }
+            },
+            asyncProfiler
+        );
     }
 
     private void consumeActivationEventsFromRingBufferAndWriteToFile(TimeDuration profilingDuration) throws Exception {
@@ -623,7 +633,9 @@ private static long peekLong(ByteBuffer buf) {
 
     public void resetActivationEventBuffer() throws IOException {
         ((Buffer) activationEventsBuffer).clear();
-        activationEventsFileChannel.position(0L);
+        if (activationEventsFileChannel != null && activationEventsFileChannel.isOpen()) {
+            activationEventsFileChannel.position(0L);
+        }
     }
 
     private void flushActivationEvents() throws IOException {
@@ -691,7 +703,7 @@ void setProfilingSessionOngoing(boolean profilingSessionOngoing) {
 
     public void clearProfiledThreads() {
         for (CallTree.Root root : profiledThreads.values()) {
-            root.recycle(callTreePool);
+            root.recycle(callTreePool, rootPool);
         }
         profiledThreads.clear();
     }
@@ -702,7 +714,6 @@ CallTree.Root getRoot() {
     }
 
     void clear() throws IOException {
-        profiledThreads.clear();
         // consume all remaining events from the ring buffer
         try {
             poller.poll(new EventPoller.Handler<ActivationEvent>() {
@@ -716,6 +727,9 @@ public boolean onEvent(ActivationEvent event, long sequence, boolean endOfBatch)
             throw new RuntimeException(e);
         }
         resetActivationEventBuffer();
+        profiledThreads.clear();
+        callTreePool.clear();
+        rootPool.clear();
     }
 
     int getProfilingSessions() {
@@ -834,8 +848,7 @@ private void startProfiling(SamplingProfiler samplingProfiler) {
                 if (logger.isDebugEnabled()) {
                     logger.warn("Illegal state when stopping profiling for thread {}: orphaned root", threadId);
                 }
-                orphaned.recycle(samplingProfiler.callTreePool);
-                samplingProfiler.rootPool.recycle(orphaned);
+                orphaned.recycle(samplingProfiler.callTreePool, samplingProfiler.rootPool);
             }
         }
 
@@ -867,17 +880,19 @@ private void stopProfiling(SamplingProfiler samplingProfiler) {
                     logger.debug("End call tree ({}) for thread {}", deserialize(samplingProfiler, traceContextBuffer), threadId);
                 }
                 samplingProfiler.profiledThreads.remove(threadId);
-                callTree.end(samplingProfiler.callTreePool, samplingProfiler.getInferredSpansMinDurationNs());
-                int createdSpans = callTree.spanify();
-                if (logger.isDebugEnabled()) {
-                    if (createdSpans > 0) {
-                        logger.debug("Created spans ({}) for thread {}", createdSpans, threadId);
-                    } else {
-                        logger.debug("Created no spans for thread {} (count={})", threadId, callTree.getCount());
+                try {
+                    callTree.end(samplingProfiler.callTreePool, samplingProfiler.getInferredSpansMinDurationNs());
+                    int createdSpans = callTree.spanify();
+                    if (logger.isDebugEnabled()) {
+                        if (createdSpans > 0) {
+                            logger.debug("Created spans ({}) for thread {}", createdSpans, threadId);
+                        } else {
+                            logger.debug("Created no spans for thread {} (count={})", threadId, callTree.getCount());
+                        }
                     }
+                } finally {
+                     callTree.recycle(samplingProfiler.callTreePool, samplingProfiler.rootPool);
                 }
-                callTree.recycle(samplingProfiler.callTreePool);
-                samplingProfiler.rootPool.recycle(callTree);
             }
         }
 

apm-agent-plugins/apm-profiling-plugin/src/test/java/co/elastic/apm/agent/profiler/CallTreeSpanifyTest.java

@@ -37,7 +37,9 @@
 import org.junit.jupiter.api.Test;
 import org.stagemonitor.configuration.ConfigurationRegistry;
 
+import java.io.IOException;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.TimeUnit;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -59,7 +61,8 @@ void setUp() {
     }
 
     @AfterEach
-    void tearDown() {
+    void tearDown() throws IOException {
+        Objects.requireNonNull(tracer.getLifecycleListener(ProfilingFactory.class)).getProfiler().clear();
         reporter.assertRecycledAfterDecrementingReferences();
         tracer.stop();
     }

apm-agent-plugins/apm-profiling-plugin/src/test/java/co/elastic/apm/agent/profiler/CallTreeTest.java

@@ -45,6 +45,7 @@
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -57,7 +58,6 @@
 
 import static java.util.stream.Collectors.toMap;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 class CallTreeTest {
@@ -77,12 +77,13 @@ void setUp() {
     }
 
     @AfterEach
-    void tearDown() {
+    void tearDown() throws IOException {
+        Objects.requireNonNull(tracer.getLifecycleListener(ProfilingFactory.class)).getProfiler().clear();
         tracer.stop();
     }
 
     @Test
-    void testCallTree() throws Exception {
+    void testCallTree() {
         TraceContext traceContext = TraceContext.with64BitId(MockTracer.create());
         CallTree.Root root = CallTree.createRoot(NoopObjectPool.ofRecyclable(() -> new CallTree.Root(tracer)), traceContext.serialize(), traceContext.getServiceName(), 0);
         ObjectPool<CallTree> callTreePool = ListBasedObjectPool.ofRecyclable(new ArrayList<>(), Integer.MAX_VALUE, CallTree::new);
@@ -921,7 +922,6 @@ public static CallTree.Root getCallTree(ElasticApmTracer tracer, String[] stackT
         transaction.deactivate().end(nanoClock.nanoTime() / 1000);
         assertThat(root).isNotNull();
         root.end(callTreePool, 0);
-        profiler.clear();
         return root;
     }