diff --git a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java
index b47b0d81717..1514c649177 100644
--- a/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java
+++ b/dd-java-agent/agent-bootstrap/src/main/java/datadog/trace/bootstrap/Agent.java
@@ -606,8 +606,9 @@ private static void startProfilingAgent(final URL bootstrapURL, final boolean is
Thread.currentThread().setContextClassLoader(classLoader);
final Class profilingAgentClass =
classLoader.loadClass("com.datadog.profiling.agent.ProfilingAgent");
- final Method profilingInstallerMethod = profilingAgentClass.getMethod("run", Boolean.TYPE);
- profilingInstallerMethod.invoke(null, isStartingFirst);
+ final Method profilingInstallerMethod =
+ profilingAgentClass.getMethod("run", Boolean.TYPE, ClassLoader.class);
+ profilingInstallerMethod.invoke(null, isStartingFirst, AGENT_CLASSLOADER);
/*
* Install the tracer hooks only when not using 'early start'.
* The 'early start' is happening so early that most of the infrastructure has not been set up yet.
diff --git a/dd-java-agent/agent-profiling/agent-profiling.gradle b/dd-java-agent/agent-profiling/agent-profiling.gradle
index 4ed4f5ec6d3..11035867722 100644
--- a/dd-java-agent/agent-profiling/agent-profiling.gradle
+++ b/dd-java-agent/agent-profiling/agent-profiling.gradle
@@ -18,6 +18,7 @@ dependencies {
api project(':dd-java-agent:agent-profiling:profiling-auxiliary')
implementation project(path: ':dd-java-agent:agent-profiling:profiling-auxiliary-async', configuration: 'shadow')
+ api project(':dd-java-agent:agent-profiling:profiling-context')
api project(':dd-java-agent:agent-profiling:profiling-uploader')
api project(':dd-java-agent:agent-profiling:profiling-controller')
api project(':dd-java-agent:agent-profiling:profiling-controller-openjdk')
diff --git a/dd-java-agent/agent-profiling/profiling-auxiliary-async/profiling-auxiliary-async.gradle b/dd-java-agent/agent-profiling/profiling-auxiliary-async/profiling-auxiliary-async.gradle
index 2fbc5538607..cf97bf2abfc 100644
--- a/dd-java-agent/agent-profiling/profiling-auxiliary-async/profiling-auxiliary-async.gradle
+++ b/dd-java-agent/agent-profiling/profiling-auxiliary-async/profiling-auxiliary-async.gradle
@@ -19,7 +19,9 @@ excludedClassesCoverage += [
'com.datadog.profiling.auxiliary.async.AsyncProfilerRecording',
'com.datadog.profiling.auxiliary.async.AsyncProfilerRecordingData',
// the config event is a simple data holder
- 'com.datadog.profiling.auxiliary.async.AsyncProfilerConfigEvent'
+ 'com.datadog.profiling.auxiliary.async.AsyncProfilerConfigEvent',
+ // simple forwarder class to the native implementation
+ "com.datadog.profiling.auxiliary.async.ContextIntervalEventsWriter",
]
dependencies {
diff --git a/dd-java-agent/agent-profiling/profiling-context/profiling-context.gradle b/dd-java-agent/agent-profiling/profiling-context/profiling-context.gradle
new file mode 100644
index 00000000000..ab6bdaf7d14
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/profiling-context.gradle
@@ -0,0 +1,54 @@
+apply from: "$rootDir/gradle/java.gradle"
+apply plugin: 'idea'
+
+ext {
+ minimumBranchCoverage = 0.6
+ minimumInstructionCoverage = 0.8
+ excludedClassesCoverage = [
+ // simple DTO
+ "com.datadog.profiling.context.PositionDecoder.*",
+ "com.datadog.profiling.context.JfrTimestampPatch",
+ // jacoco does not allow per-method excludes so, here we go
+ "com.datadog.profiling.context.ProfilerTracingContextTrackerFactory*",
+ "com.datadog.profiling.context.ProfilerTracingContextTracker.TimeTicksProvider*",
+ ]
+}
+
+sourceSets {
+ "main_java11" {
+ java.srcDirs "${project.projectDir}/src/main/java11"
+ }
+}
+
+dependencies {
+ api project(':internal-api')
+
+ implementation deps.slf4j
+ main_java11CompileOnly deps.slf4j
+ implementation sourceSets.main_java11.output
+
+ testImplementation deps.junit5
+ testImplementation deps.mockito
+}
+
+sourceCompatibility = JavaVersion.VERSION_1_8
+targetCompatibility = JavaVersion.VERSION_1_8
+
+compileMain_java11Java.doFirst {
+ setJavaVersion(it, 11)
+ sourceCompatibility = JavaVersion.VERSION_1_8
+ targetCompatibility = JavaVersion.VERSION_1_8
+}
+
+jar {
+ from sourceSets.main_java11.output
+}
+forbiddenApisMain_java11 {
+ failOnMissingClasses = false
+}
+
+idea {
+ module {
+ jdkName = '11'
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/Allocator.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/Allocator.java
new file mode 100644
index 00000000000..bf938c45db1
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/Allocator.java
@@ -0,0 +1,29 @@
+package com.datadog.profiling.context;
+
+import com.datadog.profiling.context.allocator.AllocatedBuffer;
+
+/** A simple allocator definition */
+public interface Allocator {
+ /**
+ * The size of the chunk the allocator opearates with
+ *
+ * @return size of the chunk the allocator opearates with
+ */
+ int getChunkSize();
+
+ /**
+ * Allocate the {@code size} number of bytes aligned to the chunk size
+ *
+ * @param size the number of bytes to allocate
+ * @return the allocated byffer or {@literal null} when allocation fails
+ */
+ AllocatedBuffer allocate(int size);
+
+ /**
+ * Allocate the {@code chunks} number of chunks
+ *
+ * @param chunks the number of chunks to allocate
+ * @return the allocated byffer or {@literal null} when allocation fails
+ */
+ AllocatedBuffer allocateChunks(int chunks);
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/IntervalEncoder.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/IntervalEncoder.java
new file mode 100644
index 00000000000..e1b7373188d
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/IntervalEncoder.java
@@ -0,0 +1,330 @@
+package com.datadog.profiling.context;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Encodes the collected interval into a binary compressed format.
+ *
+ * The binary blob can be split into a prologue and data chunk whereas the data chunk immediately
+ * follows the prologue.
+ *
+ *
The prologue consists of a header and thread map.
+ *
+ *
The header is containing information about the number of covered threads, the start timestamp
+ * in ticks and the frequency multiplier allowing to convert ticks to nanoseconds.
+ *
+ *
The thread map is a list of [thread id, interval count] tuples.
+ *
+ *
The data chunk holds the list of intervals in the form of delta encoded offsets against the
+ * start timestamp.
+ *
+ *
Each thread has its own delta encoded sequence - meaning that the start of the first interval
+ * for each thread is encoded as the delta against the blob start timestamp, whereas the rest of the
+ * timestamps are delta encoded against the previous timestamp. In this sequence each two subsequent
+ * timestamps are forming an interval entry.
+ *
+ *
All the timestamp delta values are further group varint
+ * encoded with the assumption that the most of the deltas will fit into 1 or 2 bytes.
+ *
+ *
The bitmap describing the size of each group varint encoded value in number of byte is located
+ * at the end of the data chunk.
+ *
+ *
+ * Formal description
+ *
+ *
+ *
+ *
Prologue Layout
+ *
+ * Header
+ *
+ *
+ *
+ * | Attribute
+ * | Description |
+ * Byte Size |
+ * Compression |
+ *
+ *
+ * | Data Chunk Offset |
+ * Offset to the data chunk start |
+ * 4 bytes |
+ * RAW |
+ *
+ *
+ * | Start Epoch Millis |
+ * The system epoch millis to peg the start ticks againstt |
+ * 1-9 bytes |
+ * LEB128 Varint |
+ *
+ *
+ * | Frequency Multiplier |
+ * The number of ticks per 1000ns |
+ * 1-5 bytes |
+ * LEB128 Varint |
+ *
+ *
+ * | Number of Threads |
+ * Number of threads with recorded intervals |
+ * 1-5 bytes |
+ * LEB128 Varint |
+ *
+ *
+ *
+ *
+ *
+ * Thread Map
+ *
+ *
+ *
+ * | Attribute
+ * | Description |
+ * Byte Size |
+ * Compression |
+ *
+ *
+ * | Thread ID |
+ * The Java thread ID |
+ * 1-5 bytes |
+ * LEB128 Varint |
+ *
+ *
+ * | Number of Intervals |
+ * The number of intervals stored for this thread |
+ * 1-5 bytes |
+ * LEB128 Varint |
+ *
+ *
+ *
+ *
+ *
+ * Data Chunk Layout
+ *
+ * Header
+ *
+ *
+ *
+ * | Attribute
+ * | Description |
+ * Byte Size |
+ * Compression |
+ *
+ *
+ * | Group Varint Bitmap Offset |
+ * Offset to the group varint size bitmap |
+ * 4 bytes |
+ * RAW |
+ *
+ *
+ *
+ *
+ *
+ * Thread interval sequence
+ *
+ * The following tuple is repeated for as many times as there are intervals for the given
+ * thread.
+ *
+ *
+ *
+ * | Attribute
+ * | Description |
+ * Byte Size |
+ * Compression |
+ *
+ *
+ * | Start delta |
+ * Delta ticks to the end of the previous interval or to the blob start timestamp if this is the first interval |
+ * 1-4 bytes |
+ * Group Varint |
+ *
+ *
+ * | End delta |
+ * Delta ticks to the start of this interval |
+ * 1-4 bytes |
+ * Group Varint |
+ *
+ *
+ *
+ *
+ *
+ * Group Varint Size Bitmap
+ *
+ * The group varint encoding of long (8 bytes) numbers is using a separate size map where each 8
+ * subsequent long numbers are encoded in the correspondingly subsequent 3 bytes where each 3 bits
+ * are representing the number of bytes the referenced long number can be represented in. The value
+ * of those 3 bits is offset by 1 meaning that if the value is 0b000 the encoded length becomes 1
+ * and 0b111 is actually 8.
+ *
+ * The following triple is repeated as many times as necessary to cover all intervals
+ *
+ *
+ *
+ * | Attribute
+ * | Description |
+ * Byte Size |
+ * Compression |
+ *
+ *
+ * | byte 1 |
+ * The following long number lengths are encoded in this byte - [111_222_33] |
+ * 1 byte |
+ * RAW |
+ *
+ *
+ * | byte 2 |
+ * The following long number lengths are encoded in this byte - [3_444_555_6] |
+ * 1 byte |
+ * * RAW |
+ *
+ *
+ * | byte 3 |
+ * The following long number lengths are encoded in this byte - [66_777_888] |
+ * 1 byte |
+ * RAW |
+ *
+ *
+ */
+final class IntervalEncoder {
+ private final ByteBuffer prologueBuffer;
+ private final ByteBuffer dataChunkBuffer;
+ private final ByteBuffer groupVarintMapBuffer;
+ private final int threadCount;
+ private int threadIndex = 0;
+
+ private int maskPos = 0;
+ private int maskOffset = 0;
+
+ private boolean encoderFinished = false;
+ private boolean threadInFlight = false;
+ private final LEB128Support leb128Support = new LEB128Support();
+
+ final class ThreadEncoder {
+ private final byte[] elements = new byte[8];
+
+ private long runningTimestamp;
+ private final long threadId;
+ private int intervals = 0;
+
+ private boolean threadFinished = false;
+
+ ThreadEncoder(long threadId) {
+ this.threadId = threadId;
+ this.runningTimestamp = 0;
+ }
+
+ void recordInterval(long from, long till) {
+ if (threadFinished) {
+ throw new IllegalStateException("Illegal state: threadFinished=" + threadFinished);
+ }
+ intervals++;
+ putLongValue(from - runningTimestamp);
+ putLongValue(till - from);
+ runningTimestamp = till;
+ }
+
+ IntervalEncoder finish() {
+ if (threadFinished) {
+ throw new IllegalStateException("Illegal state: threadFinished=" + threadFinished);
+ }
+ leb128Support.putVarint(prologueBuffer, threadId);
+ leb128Support.putVarint(prologueBuffer, intervals);
+ threadInFlight = false;
+ threadFinished = true;
+ return IntervalEncoder.this;
+ }
+
+ private void putLongValue(long value) {
+ int size = leb128Support.longSize(value);
+
+ int base = size - 1;
+ for (int i = 0; i < size; i++) {
+ elements[base - i] = (byte) (value & 0xff);
+ value = value >>> 8;
+ }
+ dataChunkBuffer.put(elements, 0, size);
+ storeLongValueSize(size);
+ }
+
+ private void storeLongValueSize(int size) {
+ groupVarintMapBuffer.position(maskPos);
+ groupVarintMapBuffer.mark();
+ int mask = groupVarintMapBuffer.getInt();
+
+ // record the current length
+ // need to do 'size - 1' to be able to squeeze the length into 3 bits -> 0 means size of 1
+ // etc.
+ mask = mask | (((size - 1) & 0x7) << (29 - maskOffset));
+ // and rewrite the mask
+ groupVarintMapBuffer.reset();
+ groupVarintMapBuffer.putInt(mask);
+ if ((maskOffset += 3) > 21) {
+ maskOffset = 0;
+ maskPos += 3;
+ }
+ }
+ }
+
+ /**
+ * @param timestampMillis the timestamp in epoch millis
+ * @param freqMultiplier number of ticks per 1000ns
+ * @param threadCount number of tracked threads
+ * @param maxSize max data size
+ */
+ IntervalEncoder(long timestampMillis, long freqMultiplier, int threadCount, int maxSize) {
+ this.threadCount = threadCount;
+
+ this.prologueBuffer =
+ ByteBuffer.allocate(
+ leb128Support.varintSize(timestampMillis)
+ + leb128Support.varintSize(freqMultiplier)
+ + leb128Support.varintSize(threadCount)
+ + threadCount * 18);
+ this.dataChunkBuffer = ByteBuffer.allocate(maxSize * 8 + 4);
+ this.groupVarintMapBuffer =
+ ByteBuffer.allocate(leb128Support.align((int) (Math.ceil(maxSize / 8d) * 3), 4));
+
+ prologueBuffer.putInt(0); // pre-allocate space for the datachunk offset
+ dataChunkBuffer.putInt(0); // pre-allocate space for the group varint map offset
+ leb128Support.putVarint(prologueBuffer, timestampMillis);
+ leb128Support.putVarint(prologueBuffer, freqMultiplier);
+ leb128Support.putVarint(prologueBuffer, threadCount);
+ }
+
+ ThreadEncoder startThread(long threadId) {
+ if (threadInFlight || threadIndex++ >= threadCount) {
+ throw new IllegalStateException(
+ "Illegal state: threadInFlight="
+ + threadInFlight
+ + ", threadIndex="
+ + threadIndex
+ + ", threadCount="
+ + threadCount);
+ }
+ threadInFlight = true;
+ return new ThreadEncoder(threadId);
+ }
+
+ ByteBuffer finish() {
+ if (encoderFinished || threadInFlight) {
+ throw new IllegalStateException(
+ "Illegal state: encoderFinished="
+ + encoderFinished
+ + ", threadInFlight="
+ + threadInFlight);
+ }
+ encoderFinished = true;
+ ByteBuffer buffer =
+ ByteBuffer.allocate(
+ prologueBuffer.position()
+ + dataChunkBuffer.position()
+ + groupVarintMapBuffer.position());
+ prologueBuffer.putInt(0, prologueBuffer.position()); // store the data chunk offset
+ dataChunkBuffer.putInt(0, dataChunkBuffer.position()); // store the group varint bitmap offset
+ // and now store the parts of the blob
+ buffer.put((ByteBuffer) prologueBuffer.flip()); // prologue
+ buffer.put((ByteBuffer) dataChunkBuffer.flip()); // data chunk
+ buffer.put((ByteBuffer) groupVarintMapBuffer.flip()); // group varint bitmap
+ return (ByteBuffer) buffer.flip();
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/IntervalSequencePruner.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/IntervalSequencePruner.java
new file mode 100644
index 00000000000..b1bff41316b
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/IntervalSequencePruner.java
@@ -0,0 +1,147 @@
+package com.datadog.profiling.context;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A utility class to 'prune' the interval sequence.
+ * An interval sequence is the result of a serie of activations and deactivations of a specific span
+ * on a specific thread. All intervals present in the sequence are related to exactly one span and
+ * thread and therefore it is possible to make assumptions about time linearity and do a simple
+ * deduplication and simplification of the recorded raw data.
+ */
+final class IntervalSequencePruner {
+ private static final Logger log = LoggerFactory.getLogger(IntervalSequencePruner.class);
+
+ /**
+ * All zero values should be removed from the sequence.
+ * {@literal 0} is used as a marker for pruned values. This possible thanks to the knowledge of
+ * the pruned data - there will never be a valid zero value store before pruning.
+ */
+ private static final class PruningLongIterator implements LongIterator {
+ private final LongIterator wrapped;
+ private long cachedValue = 0L;
+
+ PruningLongIterator(LongIterator wrapped) {
+ this.wrapped = wrapped;
+ }
+
+ @Override
+ public boolean hasNext() {
+ if (cachedValue != 0) {
+ // previously computed cached value available
+ return true;
+ }
+ while (wrapped.hasNext()) {
+ long val = wrapped.next();
+ if (val != 0) {
+ cachedValue = val;
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public long next() {
+ try {
+ return cachedValue;
+ } finally {
+ cachedValue = 0;
+ }
+ }
+ }
+
+ /**
+ * Perform deduplication and simplification of the original sequence raw data.
+ *
+ * The deduplication and simplification rules are
+ *
+ *
+ * - All activations happening right after each other are 'collapsed' to the first one
+ *
- All 'true' deactivations happening right after each other are 'collapsed' to the last one
+ *
- All 'maybe' deactivations are
+ *
+ * - turned into 'true' deactivations if they follow 'true' deactivation
+ *
- pruned if they are followed by an activate - the pruning constitues of removing
+ * both the deactivation and the following activation, effectively extending the
+ * original interval
+ *
+ * After the 'maybe' deactivation is turned into 'true' deactivation or pruned the other
+ * deduplication rules apply as necessary.
+ *
+ *
+ * @param sequence the raw data sequence
+ * @param timestampDelta the timestamp delta to use for the synthetic end transition if necessary
+ * @return a {@linkplain LongIterator} instance providing access to the pruned data
+ */
+ LongIterator pruneIntervals(LongSequence sequence, long timestampDelta) {
+ int lastTransition = ProfilerTracingContextTracker.TRANSITION_NONE;
+ int finishIndexStart = -1;
+ int sequenceOffset = 0;
+ LongIterator iterator = sequence.iterator();
+ while (iterator.hasNext()) {
+ long value = iterator.next();
+ int transition = (int) ((value & ProfilerTracingContextTracker.TRANSITION_MASK) >>> 62);
+ if (transition == ProfilerTracingContextTracker.TRANSITION_STARTED) {
+ if (lastTransition == ProfilerTracingContextTracker.TRANSITION_STARTED) {
+ // skip duplicated starts
+ sequence.set(sequenceOffset, 0L);
+ } else if (lastTransition > ProfilerTracingContextTracker.TRANSITION_STARTED) {
+ int collapsedLength = sequenceOffset - finishIndexStart - 1;
+ if (collapsedLength > 0) {
+ for (int i = 0; i < collapsedLength; i++) {
+ sequence.set(finishIndexStart + i, 0L);
+ }
+ }
+ finishIndexStart = -1;
+
+ if (lastTransition == ProfilerTracingContextTracker.TRANSITION_MAYBE_FINISHED) {
+ // skip 'maybe finished' followed by started
+ sequence.set(sequenceOffset - 1, 0L);
+ // also, ignore the start - instead just continue the previous interval
+ sequence.set(sequenceOffset, 0L);
+ }
+ }
+ } else if (transition == ProfilerTracingContextTracker.TRANSITION_MAYBE_FINISHED) {
+ if (lastTransition == ProfilerTracingContextTracker.TRANSITION_NONE) {
+ // dangling transition - remove it
+ log.debug("Dangling 'maybe finished' transition");
+ sequence.set(sequenceOffset, 0L);
+ transition = lastTransition;
+ } else if (lastTransition == ProfilerTracingContextTracker.TRANSITION_STARTED) {
+ finishIndexStart = sequenceOffset;
+ } else if (lastTransition == ProfilerTracingContextTracker.TRANSITION_FINISHED) {
+ // 'finish' followed by 'maybe finished' turns into 'finished'
+ transition = lastTransition;
+ }
+ } else { // if (transition == ProfilerTracingContextTracker.TRANSITION_FINISHED)
+ if (lastTransition == ProfilerTracingContextTracker.TRANSITION_NONE) {
+ // dangling transition - remove it
+ log.debug("Dangling 'finished' transition");
+ sequence.set(sequenceOffset, 0L);
+ transition = lastTransition;
+ } else if (lastTransition == ProfilerTracingContextTracker.TRANSITION_STARTED) {
+ // mark the start of 'finished' sequence
+ finishIndexStart = sequenceOffset;
+ }
+ }
+ lastTransition = transition;
+ sequenceOffset++;
+ }
+ if (finishIndexStart > -1) {
+ int maxIndex = sequence.size() - 1;
+ for (int i = finishIndexStart; i < maxIndex; i++) {
+ sequence.set(i, 0L);
+ }
+ }
+ if (lastTransition == ProfilerTracingContextTracker.TRANSITION_STARTED) {
+ // dangling start -> create a synthetic finished transition
+ log.debug(
+ "Dangling 'started' transition. Creating synthetic 'finished' transition @{} delta",
+ timestampDelta);
+ sequence.add(ProfilerTracingContextTracker.maskDeactivation(timestampDelta, false));
+ }
+ return new PruningLongIterator(sequence.iterator());
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/JfrTimestampPatch.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/JfrTimestampPatch.java
new file mode 100644
index 00000000000..6f86e9aaabb
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/JfrTimestampPatch.java
@@ -0,0 +1,7 @@
+package com.datadog.profiling.context;
+
+public final class JfrTimestampPatch {
+ public static void execute(ClassLoader agentClassLoader) {
+ JfrTimestampPatchImpl.execute(agentClassLoader);
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/LEB128Support.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/LEB128Support.java
new file mode 100644
index 00000000000..07055ea6ddf
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/LEB128Support.java
@@ -0,0 +1,98 @@
+package com.datadog.profiling.context;
+
+import java.nio.ByteBuffer;
+
+final class LEB128Support {
+ private static final int EXT_BIT = 0x80;
+ private static final long COMPRESSED_INT_MASK = -EXT_BIT;
+
+ int align(int value, int alignment) {
+ return value == 0 ? 0 : (((value - 1) / alignment) + 1) * alignment;
+ }
+
+ int varintSize(long value) {
+ if (value >= 0 && value < 255) {
+ return 1;
+ }
+ int pos = 63;
+ long mask = 0xFE00000000000000L;
+ while ((value & mask) == 0) {
+ pos -= 7;
+ mask = mask >>> 7;
+ }
+ return ((pos - 1) / 7) + 1;
+ }
+
+ int longSize(long value) {
+ int pos = 63;
+ long mask = 0xFF00000000000000L;
+ while (pos > 0 && (value & mask) == 0) {
+ pos -= 8;
+ mask = mask >>> 8;
+ }
+ return (pos / 8) + 1;
+ }
+
+ void putVarint(ByteBuffer buffer, long value) {
+ // if (value < 0) {
+ // // negative numbers are not supported
+ // return;
+ // }
+ if ((value & COMPRESSED_INT_MASK) == 0) {
+ buffer.put((byte) ((value & 0x7f)));
+ return;
+ }
+ buffer.put((byte) ((value & 0x7f) | EXT_BIT));
+
+ value >>= 7;
+ if ((value & COMPRESSED_INT_MASK) == 0) {
+ buffer.put((byte) ((value & 0x7f)));
+ return;
+ }
+ buffer.put((byte) ((value & 0x7f) | EXT_BIT));
+
+ value >>= 7;
+ if ((value & COMPRESSED_INT_MASK) == 0) {
+ buffer.put((byte) ((value & 0x7f)));
+ return;
+ }
+ buffer.put((byte) ((value & 0x7f) | EXT_BIT));
+
+ value >>= 7;
+ if ((value & COMPRESSED_INT_MASK) == 0) {
+ buffer.put((byte) ((value & 0x7f)));
+ return;
+ }
+ buffer.put((byte) ((value & 0x7f) | EXT_BIT));
+
+ value >>= 7;
+ if ((value & COMPRESSED_INT_MASK) == 0) {
+ buffer.put((byte) ((value & 0x7f)));
+ return;
+ }
+ buffer.put((byte) ((value & 0x7f) | EXT_BIT));
+
+ value >>= 7;
+ if ((value & COMPRESSED_INT_MASK) == 0) {
+ buffer.put((byte) ((value & 0x7f)));
+ return;
+ }
+ buffer.put((byte) ((value & 0x7f) | EXT_BIT));
+
+ value >>= 7;
+ if ((value & COMPRESSED_INT_MASK) == 0) {
+ buffer.put((byte) ((value & 0x7f)));
+ return;
+ }
+ buffer.put((byte) ((value & 0x7f) | EXT_BIT));
+
+ value >>= 7;
+ if ((value & COMPRESSED_INT_MASK) == 0) {
+ buffer.put((byte) ((value & 0x7f)));
+ return;
+ }
+ buffer.put((byte) ((value & 0x7f) | EXT_BIT));
+
+ buffer.put((byte) ((value >> 7) & 0x7f));
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/LongIterator.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/LongIterator.java
new file mode 100644
index 00000000000..43299fb19c6
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/LongIterator.java
@@ -0,0 +1,13 @@
+package com.datadog.profiling.context;
+
+/** A primitive long-value iterator */
+public interface LongIterator {
+ /** @return {@literal true} if the iterated seuqence has more elements */
+ boolean hasNext();
+
+ /**
+ * @return the next long-value from the iterated sequence
+ * @throws IllegalStateException if this method is called and there is no next value
+ */
+ long next();
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/LongSequence.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/LongSequence.java
new file mode 100644
index 00000000000..bffe864cca6
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/LongSequence.java
@@ -0,0 +1,235 @@
+package com.datadog.profiling.context;
+
+import com.datadog.profiling.context.allocator.AllocatedBuffer;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A growing sequence of long values.
+ * Allows adding new values, rewriting values at a given position, retrieving values from a position
+ * and creating {@linkplain LongIterator} over the sequence.
+ *
+ * This class is not thread-safe in a generic way. The only concurrent access that is allowed is
+ * doing 'release' while a sequence is manipulated and vice versa. For all other combinations of
+ * method invocations the results are not defined. The caller should take care of proper
+ * synchronization, if necessary.
+ */
+final class LongSequence {
+ private static final Logger log = LoggerFactory.getLogger(LongSequence.class);
+
+ private class LongIteratorImpl implements LongIterator {
+ int bufferReadSlot = 0;
+ int allIndex = 0;
+ LongIterator currentIterator = null;
+
+ @Override
+ public boolean hasNext() {
+ if (bufferReadSlot > bufferWriteSlot || allIndex >= size) {
+ return false;
+ }
+ if (currentIterator == null) {
+ currentIterator = buffers[bufferReadSlot].iterator();
+ }
+ if (currentIterator.hasNext()) {
+ return true;
+ }
+ do {
+ if (++bufferReadSlot > bufferWriteSlot) {
+ return false;
+ }
+ currentIterator = buffers[bufferReadSlot].iterator();
+ } while (!currentIterator.hasNext());
+ return true;
+ }
+
+ @Override
+ public long next() {
+ if (currentIterator == null) {
+ throw new IllegalStateException();
+ }
+
+ long value = currentIterator.next();
+ allIndex++;
+ return value;
+ }
+ }
+
+ // with capacity increment c[n] = c[n -1] + 2 * c[n - 1] we can safely accommodate 100k elements
+ private AllocatedBuffer[] buffers = new AllocatedBuffer[10];
+ private int[] bufferBoundaryMap = new int[10];
+
+ private int bufferInitSlot = 0;
+ private int bufferWriteSlot = 0;
+ private int capacity = 0;
+ private int capacityInChunks = 0;
+ private int size = 0;
+ private int sizeInBytes = 0;
+ private int threshold = 0;
+ private final Allocator allocator;
+ private final PositionDecoder positionDecoder = PositionDecoder.getInstance();
+
+ private static int align(int size) {
+ return (int) (Math.ceil(size / 8d) * 8);
+ }
+
+ private AtomicInteger state = new AtomicInteger(0);
+
+ public LongSequence(Allocator allocator) {
+ this.allocator = allocator;
+ this.bufferWriteSlot = -1;
+ }
+
+ public int add(long value) {
+ // check'n'update the state - if it is non-negative, increment the counter and proceed,
+ // otherwise bail out
+ if (state.updateAndGet(prev -> prev >= 0 ? prev + 1 : prev) < 0) {
+ // bail out if this instance was already released
+ return -1;
+ }
+
+ try {
+ if (bufferWriteSlot == -1) {
+ // first write; initialize the data structure
+ capacityInChunks = 1;
+ bufferWriteSlot = 0;
+ bufferInitSlot = 0;
+ AllocatedBuffer cBuffer = allocator.allocateChunks(capacityInChunks);
+ buffers[bufferWriteSlot] = cBuffer;
+ if (cBuffer != null) {
+ capacity = cBuffer.capacity();
+ threshold = align((int) (this.capacity * 0.75f));
+ bufferBoundaryMap[bufferWriteSlot] = capacity - 1;
+ } else {
+ capacity = 0;
+ threshold = -1;
+ }
+ } else {
+ if (threshold > -1 && sizeInBytes == threshold) {
+ // we hit the threshold - let's prepare the next-in-line buffer
+ int newCapacity = 2 * capacityInChunks; // capacity stays aligned
+ AllocatedBuffer cBuffer = allocator.allocateChunks(newCapacity);
+ if (cBuffer != null) {
+ bufferInitSlot = bufferWriteSlot + 1;
+ buffers[bufferInitSlot] = cBuffer;
+ capacityInChunks += newCapacity;
+ capacity += cBuffer.capacity(); // update the sequence capacity
+ threshold = align((int) (capacity * 0.75f)); // update the threshold
+ bufferBoundaryMap[bufferInitSlot] = capacity - 1;
+ } else {
+ threshold = -1;
+ }
+ }
+ }
+ if (bufferWriteSlot == buffers.length || buffers[bufferWriteSlot] == null) {
+ return 0;
+ }
+ while (!buffers[bufferWriteSlot].putLong(value)) {
+ bufferWriteSlot++;
+ if (bufferWriteSlot == buffers.length || buffers[bufferWriteSlot] == null) {
+ return 0;
+ }
+ }
+
+ size += 1;
+ sizeInBytes += 8;
+ return 1;
+ } finally {
+ // check'n'update the state - if it is negative before update, perform release, otherwise just
+ // decrement the counter
+ if (state.getAndUpdate(prev -> prev > 0 ? prev - 1 : prev) < 0) {
+ doRelease();
+ }
+ }
+ }
+
+ public boolean set(int index, long value) {
+ // check'n'update the state - if it is non-negative, increment the counter and proceed,
+ // otherwise bail out
+ if (state.updateAndGet(prev -> prev >= 0 ? prev + 1 : prev) < 0) {
+ // bail out if this instance was already released
+ return false;
+ }
+ try {
+ PositionDecoder.Coordinates decoded =
+ positionDecoder.decode(index * 8, bufferBoundaryMap, bufferInitSlot + 1);
+ if (decoded != null) {
+ return buffers[decoded.slot].putLong(decoded.index, value);
+ }
+ return false;
+ } finally {
+ // check'n'update the state - if it is negative before update, perform release, otherwise just
+ // decrement the counter
+ if (state.getAndUpdate(prev -> prev > 0 ? prev - 1 : prev) < 0) {
+ doRelease();
+ }
+ }
+ }
+
+ public long get(int index) {
+ // check'n'update the state - if it is non-negative, increment the counter and proceed,
+ // otherwise bail out
+ if (state.updateAndGet(prev -> prev >= 0 ? prev + 1 : prev) < 0) {
+ // bail out if this instance was already released
+ return Long.MIN_VALUE;
+ }
+ try {
+ PositionDecoder.Coordinates decoded =
+ positionDecoder.decode(index * 8, bufferBoundaryMap, bufferInitSlot + 1);
+ if (decoded != null) {
+ return buffers[decoded.slot].getLong(decoded.index);
+ }
+ return Long.MIN_VALUE;
+ } finally {
+ // check'n'update the state - if it is negative before update, perform release, otherwise just
+ // decrement the counter
+ if (state.getAndUpdate(prev -> prev > 0 ? prev - 1 : prev) < 0) {
+ doRelease();
+ }
+ }
+ }
+
+ public int size() {
+ // check'n'update the state - if it is non-negative, increment the counter and proceed,
+ // otherwise bail out
+ if (state.updateAndGet(prev -> prev >= 0 ? prev + 1 : prev) < 0) {
+ // bail out if this instance was already released
+ return 0;
+ }
+ try {
+ return size;
+ } finally {
+ // check'n'update the state - if it is negative before update, perform release, otherwise just
+ // decrement the counter
+ if (state.getAndUpdate(prev -> prev > 0 ? prev - 1 : prev) < 0) {
+ doRelease();
+ }
+ }
+ }
+
+ public void release() {
+ if (state.getAndUpdate(prev -> prev > 0 ? -1 * prev : prev) == 0) {
+ doRelease();
+ }
+ }
+
+ private void doRelease() {
+ for (int i = 0; i < buffers.length; i++) {
+ AllocatedBuffer buffer = buffers[i];
+ if (buffer != null) {
+ buffer.release();
+ }
+ }
+ // clear out the buffer slots to allow the most of the retained data to be GCed
+ // do not clear up the buffers contents as they still may be in use by an iterator instance
+ buffers = null;
+ bufferBoundaryMap = null;
+ bufferInitSlot = -1;
+ size = 0;
+ state.set(-1);
+ }
+
+ public LongIterator iterator() {
+ return new LongIteratorImpl();
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/PositionDecoder.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/PositionDecoder.java
new file mode 100644
index 00000000000..fb84f96c09b
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/PositionDecoder.java
@@ -0,0 +1,108 @@
+package com.datadog.profiling.context;
+
+import java.util.Arrays;
+import java.util.Objects;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+/**
+ * A utility class to decode a position into [slot, index] coordinates in an array of indexable
+ * buffers. It supports each of the indexable buffers having a different size.
+ */
+public final class PositionDecoder {
+ /** Coordinates data holder */
+ public static final class Coordinates {
+ public final int slot;
+ public final int index;
+
+ public Coordinates(int slot, int index) {
+ this.slot = slot;
+ this.index = index;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ Coordinates coordinates = (Coordinates) o;
+ return slot == coordinates.slot && index == coordinates.index;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(slot, index);
+ }
+ }
+
+ private static final class Singleton {
+ static final PositionDecoder INSTANCE = new PositionDecoder();
+ }
+
+ public static PositionDecoder getInstance() {
+ return Singleton.INSTANCE;
+ }
+
+ /**
+ * Turn the position into a {@linkplain Coordinates} instance
+ *
+ * @param position the position
+ * @param bufferBoundaryMap an array of buffer boundaries - a boundary is defined as 'size - 1'
+ * for each buffer element
+ * @return decoded {@linkplain Coordinates} or {@literal null}
+ */
+ @Nullable
+ public Coordinates decode(int position, int[] bufferBoundaryMap) {
+ return decode(position, bufferBoundaryMap, bufferBoundaryMap.length);
+ }
+
+ /**
+ * Turn the position into a {@linkplain Coordinates} instance
+ *
+ * @param position the position
+ * @param bufferBoundaryMap an array of buffer boundaries - a boundary is defined as 'size - 1'
+ * for each buffer element
+ * @param bufferBoundaryMapLimit limit the buffer boundaries to be used only to the first
+ * {@literal bufferBoundaryMapLimit} ones
+ * @return decoded {@linkplain Coordinates} or {@literal null}
+ */
+ @Nullable
+ public Coordinates decode(
+ int position, @Nonnull int[] bufferBoundaryMap, int bufferBoundaryMapLimit) {
+ if (bufferBoundaryMap.length == 0) {
+ return null;
+ }
+ if (bufferBoundaryMapLimit <= 0 || bufferBoundaryMapLimit > bufferBoundaryMap.length) {
+ return null;
+ }
+
+ // shortcut for a position falling within the first slot
+ if (position <= bufferBoundaryMap[0]) {
+ return new Coordinates(0, position);
+ }
+ // shortcut for positions not covered by the boundary map
+ if (position > bufferBoundaryMap[bufferBoundaryMapLimit - 1]) {
+ return null;
+ }
+
+ // shortcut to linear search for a small number of slots in use
+ if (bufferBoundaryMapLimit < 5) {
+ int slot = 0;
+ while (slot <= bufferBoundaryMapLimit && bufferBoundaryMap[slot] < position) {
+ slot++;
+ }
+ if (slot <= bufferBoundaryMapLimit) {
+ return new Coordinates(slot, position - bufferBoundaryMap[slot - 1] - 1);
+ }
+ return null;
+ }
+
+ // use binary search
+ int slot = Arrays.binarySearch(bufferBoundaryMap, position);
+ if (slot > 0) {
+ return new Coordinates(slot, position - bufferBoundaryMap[slot - 1] - 1);
+ } else {
+ slot = -1 - slot;
+ return new Coordinates(slot, position - bufferBoundaryMap[slot - 1] - 1);
+ }
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/ProfilerTracingContextTracker.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/ProfilerTracingContextTracker.java
new file mode 100644
index 00000000000..bc936da8201
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/ProfilerTracingContextTracker.java
@@ -0,0 +1,349 @@
+package com.datadog.profiling.context;
+
+import datadog.trace.api.profiling.TracingContextTracker;
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import datadog.trace.relocate.api.RatelimitedLogger;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public final class ProfilerTracingContextTracker implements TracingContextTracker {
+ private static final Logger log = LoggerFactory.getLogger(ProfilerTracingContextTracker.class);
+
+ private static final byte[] EMPTY_DATA = new byte[0];
+
+ static final int TRANSITION_STARTED = 0;
+ static final int TRANSITION_MAYBE_FINISHED = 1;
+ static final int TRANSITION_NONE = -1;
+ static final int TRANSITION_FINISHED = 2;
+
+ public interface TimeTicksProvider {
+ TimeTicksProvider SYSTEM =
+ new TimeTicksProvider() {
+ @Override
+ public long ticks() {
+ return System.nanoTime();
+ }
+
+ @Override
+ public long frequency() {
+ return 1_000_000_000L; // nanosecond frequency
+ }
+ };
+
+ long ticks();
+
+ long frequency();
+ }
+
+ static final class DelayedTrackerImpl implements TracingContextTracker.DelayedTracker {
+ volatile ProfilerTracingContextTracker ref;
+
+ DelayedTrackerImpl(ProfilerTracingContextTracker ref) {
+ this.ref = ref;
+ }
+
+ @Override
+ public void cleanup() {
+ ProfilerTracingContextTracker instance = ref;
+ if (instance != null) {
+ instance.release();
+ /*
+ It is important to remove the reference such that the tracker, and transitively the span
+ is not held by the delayed instance until its timeout
+ */
+ releaseRef();
+ }
+ }
+
+ void releaseRef() {
+ ref = null;
+ }
+
+ @Override
+ public long getDelay(TimeUnit unit) {
+ ProfilerTracingContextTracker instance = ref;
+ if (ref != null) {
+ return unit.convert(
+ instance.lastTransitionTimestamp + instance.inactivityDelay - System.nanoTime(),
+ TimeUnit.NANOSECONDS);
+ }
+ // if the rerence was cleared this instance is ready for immediate removal from the queue
+ return -1;
+ }
+
+ @Override
+ public int compareTo(Delayed o) {
+ if (o instanceof ProfilerTracingContextTracker.DelayedTrackerImpl) {
+ ProfilerTracingContextTracker thiz = ref;
+ ProfilerTracingContextTracker other = ((DelayedTrackerImpl) o).ref;
+ return Long.compare(
+ thiz != null ? thiz.lastTransitionTimestamp : -1,
+ other != null ? other.lastTransitionTimestamp : -1);
+ }
+ return 0;
+ }
+ }
+
+ private static final RatelimitedLogger warnlog =
+ new RatelimitedLogger(
+ LoggerFactory.getLogger(ProfilerTracingContextTracker.class), 30, TimeUnit.SECONDS);
+
+ private static final AtomicReferenceFieldUpdater<
+ ProfilerTracingContextTracker, DelayedTrackerImpl>
+ DELAYED_TRACKER_REF_UPDATER =
+ AtomicReferenceFieldUpdater.newUpdater(
+ ProfilerTracingContextTracker.class, DelayedTrackerImpl.class, "delayedTrackerRef");
+ private static final long TRANSITION_MAYBE_FINISHED_MASK =
+ (long) (TRANSITION_MAYBE_FINISHED) << 62;
+ private static final long TRANSITION_FINISHED_MASK = (long) (TRANSITION_FINISHED) << 62;
+ static final long TRANSITION_MASK = TRANSITION_FINISHED_MASK | TRANSITION_MAYBE_FINISHED_MASK;
+ static final long TIMESTAMP_MASK = ~TRANSITION_MASK;
+
+ private final long inactivityDelay;
+
+ private final ConcurrentMap threadSequences = new ConcurrentHashMap<>(64);
+ private final long startTimestampTicks;
+ private final long startTimestampMillis;
+ private final long delayedActivationTimestamp;
+ private final Allocator allocator;
+ private final AtomicBoolean released = new AtomicBoolean();
+ private final AgentSpan span;
+ private final TimeTicksProvider timeTicksProvider;
+ private final IntervalSequencePruner sequencePruner;
+
+ private final long initialThreadId;
+ private final AtomicBoolean initialized = new AtomicBoolean(false);
+ private final AtomicReference persisted = new AtomicReference<>(null);
+
+ private long lastTransitionTimestamp = -1;
+
+ private volatile DelayedTrackerImpl delayedTrackerRef = null;
+
+ ProfilerTracingContextTracker(
+ Allocator allocator,
+ AgentSpan span,
+ TimeTicksProvider timeTicksProvider,
+ IntervalSequencePruner sequencePruner) {
+ this(allocator, span, timeTicksProvider, sequencePruner, -1L);
+ }
+
+ ProfilerTracingContextTracker(
+ Allocator allocator,
+ AgentSpan span,
+ TimeTicksProvider timeTicksProvider,
+ IntervalSequencePruner sequencePruner,
+ long inactivityDelay) {
+ this.startTimestampTicks = timeTicksProvider.ticks();
+ this.startTimestampMillis = System.currentTimeMillis();
+ this.timeTicksProvider = timeTicksProvider;
+ this.sequencePruner = sequencePruner;
+ this.span = span;
+ this.allocator = allocator;
+ this.initialThreadId = Thread.currentThread().getId();
+ this.lastTransitionTimestamp = System.nanoTime();
+ this.inactivityDelay = inactivityDelay;
+ this.delayedActivationTimestamp = timeTicksProvider.ticks();
+ }
+
+ @Override
+ public void activateContext() {
+ activateContext(Thread.currentThread().getId());
+ }
+
+ private void activateContext(long threadId) {
+ activateContext(threadId, accessTimestamp());
+ }
+
+ void activateContext(long threadId, long timestamp) {
+ storeDelayedActivation();
+ long tsDiff = timestamp - startTimestampTicks;
+ long masked = maskActivation(tsDiff);
+ store(threadId, masked);
+ }
+
+ private long accessTimestamp() {
+ long ts = timeTicksProvider.ticks();
+ lastTransitionTimestamp = System.nanoTime();
+ return ts;
+ }
+
+ static long maskActivation(long value) {
+ return value & TIMESTAMP_MASK;
+ }
+
+ static long maskDeactivation(long value, boolean maybe) {
+ return (value & TIMESTAMP_MASK)
+ | (maybe ? TRANSITION_MAYBE_FINISHED_MASK : TRANSITION_FINISHED_MASK);
+ }
+
+ @Override
+ public void deactivateContext() {
+ deactivateContext(Thread.currentThread().getId(), false);
+ }
+
+ @Override
+ public void maybeDeactivateContext() {
+ deactivateContext(Thread.currentThread().getId(), true);
+ }
+
+ private void deactivateContext(long threadId, boolean maybe) {
+ deactivateContext(threadId, accessTimestamp(), maybe);
+ }
+
+ void deactivateContext(long threadId, long timestamp, boolean maybe) {
+ storeDelayedActivation();
+ long tsDiff = timestamp - startTimestampTicks;
+ long masked = maskDeactivation(tsDiff, maybe);
+ store(threadId, masked);
+ }
+
+ @Override
+ public byte[] persist() {
+ byte[] data = null;
+ if (persisted.compareAndSet(null, EMPTY_DATA)) {
+ try {
+ if (released.get()) {
+ // tracker was released without persisting the data
+ return null;
+ }
+ ByteBuffer buffer = encodeIntervals();
+ data = buffer.array();
+ data = Arrays.copyOf(data, buffer.limit());
+ } finally {
+ // make sure the other threads do not stay blocked even if there is an exception thrown
+ persisted.compareAndSet(EMPTY_DATA, data);
+ }
+ } else {
+ // busy wait for the data to become available
+ while ((data = persisted.get()) == EMPTY_DATA) {
+ Thread.yield();
+ }
+ }
+ return data;
+ }
+
+ LongIterator pruneIntervals(LongSequence sequence) {
+ return sequencePruner.pruneIntervals(sequence, timeTicksProvider.ticks() - startTimestampTicks);
+ }
+
+ @Override
+ public boolean release() {
+ boolean result = releaseThreadSequences();
+ releaseDelayed();
+ return result;
+ }
+
+ private boolean releaseThreadSequences() {
+ // the released flag needs to be set first such that it would prevent using the resourcese being
+ // released from 'persist()' method
+ if (released.compareAndSet(false, true)) {
+ log.trace("Releasing tracing context for span {}", span);
+ // now let's wait for any date being currently persisted
+ // 'persist()' method guarantees that the 'persisted' value will become != EMPTY_DATA
+ while (persisted.get() == EMPTY_DATA) {
+ Thread.yield();
+ }
+ // it is safe to cleanup the resources
+ threadSequences.values().forEach(LongSequence::release);
+ threadSequences.clear();
+ return true;
+ }
+ return false;
+ }
+
+ void releaseDelayed() {
+ DelayedTrackerImpl delayed = DELAYED_TRACKER_REF_UPDATER.getAndSet(this, null);
+ if (delayed != null) {
+ delayed.releaseRef();
+ }
+ }
+
+ @Override
+ public int getVersion() {
+ return 1;
+ }
+
+ @Override
+ public DelayedTracker asDelayed() {
+ return DELAYED_TRACKER_REF_UPDATER.updateAndGet(
+ this, prev -> prev != null ? prev : new DelayedTrackerImpl(this));
+ }
+
+ private void store(long threadId, long value) {
+ if (released.get()) {
+ return;
+ }
+ int added = 0;
+ LongSequence sequence =
+ threadSequences.computeIfAbsent(threadId, k -> new LongSequence(allocator));
+ synchronized (sequence) {
+ added = sequence.add(value);
+ }
+ if (added == -1) {
+ warnlog.warn(
+ "Attempting to add transition to already released context - losing tracing context data");
+ } else if (added == 0) {
+ warnlog.warn("Profiling Context Buffer is full - losing tracing context data");
+ }
+ }
+
+ private ByteBuffer encodeIntervals() {
+ int totalSequenceBufferSize = 0;
+ int maxSequenceSize = 0;
+ Set> entrySet = new HashSet<>(threadSequences.entrySet());
+ for (Map.Entry entry : entrySet) {
+ LongSequence sequence = entry.getValue();
+ maxSequenceSize = Math.max(maxSequenceSize, sequence.size());
+ totalSequenceBufferSize += sequence.size();
+ }
+
+ IntervalEncoder encoder =
+ new IntervalEncoder(
+ startTimestampMillis,
+ timeTicksProvider.frequency() / 1_000_000L,
+ threadSequences.size(),
+ totalSequenceBufferSize);
+ for (Map.Entry entry : threadSequences.entrySet()) {
+ long threadId = entry.getKey();
+ IntervalEncoder.ThreadEncoder threadEncoder = encoder.startThread(threadId);
+ LongSequence rawIntervals = entry.getValue();
+ synchronized (rawIntervals) {
+ LongIterator iterator = pruneIntervals(entry.getValue());
+ int sequenceIndex = 0;
+ while (iterator.hasNext() && sequenceIndex++ < maxSequenceSize) {
+ long from = iterator.next();
+ long maskedFrom = (from & TIMESTAMP_MASK);
+ if (iterator.hasNext()) {
+ long till = iterator.next();
+ long maskedTill = (till & TIMESTAMP_MASK);
+ if (maskedTill > maskedFrom) {
+ threadEncoder.recordInterval(maskedFrom, maskedTill);
+ }
+ }
+ }
+ }
+ threadEncoder.finish();
+ }
+ ByteBuffer buffer = encoder.finish();
+ return buffer;
+ }
+
+ private void storeDelayedActivation() {
+ if (initialized.compareAndSet(false, true)) {
+ log.trace("Storing delayed activation for span {}", span);
+ activateContext(initialThreadId, delayedActivationTimestamp);
+ }
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/ProfilerTracingContextTrackerFactory.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/ProfilerTracingContextTrackerFactory.java
new file mode 100644
index 00000000000..7816dc73f1d
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/ProfilerTracingContextTrackerFactory.java
@@ -0,0 +1,169 @@
+package com.datadog.profiling.context;
+
+import com.datadog.profiling.context.allocator.Allocators;
+import datadog.trace.api.config.ProfilingConfig;
+import datadog.trace.api.profiling.TracingContextTracker;
+import datadog.trace.api.profiling.TracingContextTrackerFactory;
+import datadog.trace.bootstrap.config.provider.ConfigProvider;
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import datadog.trace.util.AgentTaskScheduler;
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public final class ProfilerTracingContextTrackerFactory
+ implements TracingContextTrackerFactory.Implementation {
+ private static final Logger log =
+ LoggerFactory.getLogger(ProfilerTracingContextTrackerFactory.class);
+
+ private static final long DEFAULT_INACTIVITY_CHECK_PERIOD_MS = 5_000L; // 5 seconds
+
+ private final DelayQueue delayQueue = new DelayQueue<>();
+
+ public static void register(ConfigProvider configProvider) {
+ if (configProvider.getBoolean(
+ ProfilingConfig.PROFILING_TRACING_CONTEXT_ENABLED,
+ ProfilingConfig.PROFILING_TRACING_CONTEXT_ENABLED_DEFAULT)) {
+ long inactivityDelayNs =
+ TimeUnit.NANOSECONDS.convert(
+ configProvider.getInteger(
+ ProfilingConfig.PROFILING_TRACING_CONTEXT_TRACKER_INACTIVE_SEC,
+ ProfilingConfig.PROFILING_TRACING_CONTEXT_TRACKER_INACTIVE_DEFAULT),
+ TimeUnit.SECONDS);
+ int reservedMemorySize =
+ configProvider.getInteger(
+ ProfilingConfig.PROFILING_TRACING_CONTEXT_RESERVED_MEMORY_SIZE,
+ ProfilingConfig.PROFILING_TRACING_CONTEXT_RESERVED_MEMORY_SIZE_DEFAULT);
+ String reservedMemoryType =
+ configProvider.getString(
+ ProfilingConfig.PROFILING_TRACING_CONTEXT_RESERVED_MEMORY_TYPE,
+ ProfilingConfig.PROFILING_TRACING_CONTEXT_RESERVED_MEMORY_TYPE_DEFAULT);
+
+ TracingContextTrackerFactory.registerImplementation(
+ new ProfilerTracingContextTrackerFactory(
+ inactivityDelayNs,
+ DEFAULT_INACTIVITY_CHECK_PERIOD_MS,
+ reservedMemorySize,
+ reservedMemoryType));
+ }
+ }
+
+ private void initializeInactiveTrackerCleanup(long inactivityCheckPeriodMs) {
+ AgentTaskScheduler.INSTANCE.scheduleAtFixedRate(
+ target -> {
+ Collection timeouts = new ArrayList<>(500);
+ int drained = 0;
+ do {
+ drained = target.drainTo(timeouts);
+ if (drained > 0) {
+ log.debug("Drained {} inactive trackers", drained);
+ }
+ Iterator iterator = timeouts.iterator();
+ while (iterator.hasNext()) {
+ iterator.next().cleanup();
+ iterator.remove();
+ }
+ } while (drained > 0);
+ },
+ delayQueue,
+ inactivityCheckPeriodMs,
+ inactivityCheckPeriodMs,
+ TimeUnit.MILLISECONDS);
+ }
+
+ private final Allocator allocator;
+ private final IntervalSequencePruner sequencePruner = new IntervalSequencePruner();
+ private final ProfilerTracingContextTracker.TimeTicksProvider timeTicksProvider;
+ private final long inactivityDelay;
+
+ ProfilerTracingContextTrackerFactory(
+ long inactivityDelayNs, long inactivityCheckPeriodMs, int reservedMemorySize) {
+ this(inactivityDelayNs, inactivityCheckPeriodMs, reservedMemorySize, "heap");
+ }
+
+ ProfilerTracingContextTrackerFactory(
+ long inactivityDelayNs,
+ long inactivityCheckPeriodMs,
+ int reservedMemorySize,
+ String reservedMemoryType) {
+ timeTicksProvider = getTicksProvider();
+ log.debug("Tracing Context Tracker Memory Type: {}", reservedMemoryType);
+ allocator =
+ reservedMemoryType.equalsIgnoreCase("direct")
+ ? Allocators.directAllocator(reservedMemorySize, 32)
+ : Allocators.heapAllocator(reservedMemorySize, 32);
+ this.inactivityDelay = inactivityDelayNs;
+
+ if (inactivityDelay > 0) {
+ initializeInactiveTrackerCleanup(inactivityCheckPeriodMs);
+ }
+ }
+
+ private static ProfilerTracingContextTracker.TimeTicksProvider getTicksProvider() {
+ try {
+ Class clz =
+ ProfilerTracingContextTracker.class.getClassLoader().loadClass("jdk.jfr.internal.JVM");
+ MethodHandle mh;
+ long frequency = 0L;
+ try {
+ mh = MethodHandles.lookup().findStatic(clz, "getJVM", MethodType.methodType(clz));
+ Object jvm = mh.invoke();
+ mh =
+ MethodHandles.lookup()
+ .findVirtual(clz, "getTicksFrequency", MethodType.methodType(long.class));
+ mh = mh.bindTo(jvm);
+ frequency = (long) mh.invokeWithArguments();
+ } catch (NoSuchMethodException ignored) {
+ // the method is available since JDK11 only
+ }
+ mh = MethodHandles.lookup().findStatic(clz, "counterTime", MethodType.methodType(long.class));
+ // sanity check to fail early if the method handle invocation does not work
+ long ticks = (long) mh.invokeExact();
+
+ MethodHandle fixedMh = mh;
+ long fixedFrequency = frequency;
+
+ log.info("Using JFR time ticks provider");
+ return new ProfilerTracingContextTracker.TimeTicksProvider() {
+ @Override
+ public long ticks() {
+ try {
+ return (long) fixedMh.invokeExact();
+ } catch (Throwable ignored) {
+ }
+ return Long.MIN_VALUE;
+ }
+
+ @Override
+ public long frequency() {
+ return fixedFrequency;
+ }
+ };
+ } catch (Throwable t) {
+ if (log.isDebugEnabled()) {
+ log.warn("Failed to access JFR timestamps. Falling back to system nanotime.", t);
+ } else {
+ log.warn("Failed to access JFR timestamps. Falling back to system nanotime.");
+ }
+ }
+ return ProfilerTracingContextTracker.TimeTicksProvider.SYSTEM;
+ }
+
+ @Override
+ public TracingContextTracker instance(AgentSpan span) {
+ ProfilerTracingContextTracker instance =
+ new ProfilerTracingContextTracker(
+ allocator, span, timeTicksProvider, sequencePruner, inactivityDelay);
+ if (inactivityDelay > 0) {
+ delayQueue.offer(instance.asDelayed());
+ }
+ return instance;
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/AllocatedBuffer.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/AllocatedBuffer.java
new file mode 100644
index 00000000000..8b83b484f4f
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/AllocatedBuffer.java
@@ -0,0 +1,48 @@
+package com.datadog.profiling.context.allocator;
+
+import com.datadog.profiling.context.LongIterator;
+
+/**
+ * Represents the memory allocated by a {@linkplain Allocators} implementation. The buffer defines a
+ * few rudimentary operations to store and retrieve long values (both sequential and positional
+ * stores are supported) as well as a custom iterator over the long values stored in the buffer.
+ */
+public interface AllocatedBuffer {
+ /** Release the held memory */
+ void release();
+
+ /** @return the buffer capacity in bytes */
+ int capacity();
+
+ /**
+ * Add the long value to the buffer
+ *
+ * @param value the long value to store
+ * @return {@literal true} if the buffer was able to store the value, {@literal false} otherwise
+ * (eg. when the buffer capacity is exhausted)
+ */
+ boolean putLong(long value);
+
+ /**
+ * Puts the long value to the buffer at the given position
+ *
+ * @param pos the position to store the value at
+ * @param value the long value to store
+ * @return {@literal true} if the buffer was able to store the value, {@literal false} otherwise
+ * (eg. when the buffer capacity is exhausted)
+ */
+ boolean putLong(int pos, long value);
+
+ /**
+ * Retrieves the long value from the given position
+ *
+ * @param pos the position to retrieve the long value from; it is upon the caller to make sure the
+ * position was previously written to and is within the capacity
+ * @return the long value stored at that position; when a read from behind the capacity limit is
+ * attempted {@linkplain Long#MIN_VALUE} is always returned
+ */
+ long getLong(int pos);
+
+ /** @return a custom long-value iterator of all values stored in the buffer up to this point */
+ LongIterator iterator();
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/Allocators.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/Allocators.java
new file mode 100644
index 00000000000..00925a5c3e4
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/Allocators.java
@@ -0,0 +1,22 @@
+package com.datadog.profiling.context.allocator;
+
+import com.datadog.profiling.context.Allocator;
+import com.datadog.profiling.context.allocator.direct.DirectAllocator;
+import com.datadog.profiling.context.allocator.heap.HeapAllocator;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** A factory class for {@linkplain Allocator} instances */
+public abstract class Allocators {
+ private static final Logger log = LoggerFactory.getLogger(Allocators.class);
+
+ private Allocators() {}
+
+ public static Allocator directAllocator(int capacity, int chunk) {
+ return new DirectAllocator(capacity, chunk);
+ }
+
+ public static Allocator heapAllocator(int capacity, int chunk) {
+ return new HeapAllocator(capacity, chunk);
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/direct/Chunk.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/direct/Chunk.java
new file mode 100644
index 00000000000..2214465d9ad
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/direct/Chunk.java
@@ -0,0 +1,29 @@
+package com.datadog.profiling.context.allocator.direct;
+
+import java.nio.ByteBuffer;
+
+final class Chunk {
+ private final DirectAllocator allocator;
+ final ByteBuffer buffer;
+ private final int ref;
+ private int weight = 1;
+
+ Chunk(DirectAllocator allocator, ByteBuffer buffer, int ref) {
+ this.allocator = allocator;
+ this.buffer = buffer;
+ this.ref = ref;
+ }
+
+ void extend() {
+ weight++;
+ buffer.limit(buffer.limit() + allocator.getChunkSize());
+ }
+
+ void release() {
+ allocator.release(ref, weight);
+ }
+
+ int getWeight() {
+ return weight;
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/direct/DirectAllocatedBuffer.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/direct/DirectAllocatedBuffer.java
new file mode 100644
index 00000000000..2dc1816b393
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/direct/DirectAllocatedBuffer.java
@@ -0,0 +1,135 @@
+package com.datadog.profiling.context.allocator.direct;
+
+import com.datadog.profiling.context.LongIterator;
+import com.datadog.profiling.context.PositionDecoder;
+import com.datadog.profiling.context.allocator.AllocatedBuffer;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public final class DirectAllocatedBuffer implements AllocatedBuffer {
+ private static final Logger log = LoggerFactory.getLogger(DirectAllocatedBuffer.class);
+
+ private final Chunk[] chunks;
+ private final int[] chunkBoundaryMap;
+ private final int capacity;
+ private int chunkWriteIndex = 0;
+ private int valueWriteIndex = 0;
+
+ private final PositionDecoder positionDecoder = PositionDecoder.getInstance();
+
+ DirectAllocatedBuffer(int bufferSize, int chunkSize, Chunk... chunks) {
+ this.chunks = chunks;
+ this.chunkBoundaryMap = new int[chunks.length];
+ this.capacity = bufferSize;
+
+ int boundary = 0;
+ for (int i = 0; i < chunks.length; i++) {
+ boundary += chunkSize * chunks[i].getWeight();
+ chunkBoundaryMap[i] = boundary - 1;
+ }
+ }
+
+ @Override
+ public void release() {
+ for (Chunk chunk : chunks) {
+ chunk.release();
+ }
+ Arrays.fill(chunks, null);
+ }
+
+ @Override
+ public int capacity() {
+ return capacity;
+ }
+
+ @Override
+ public boolean putLong(long value) {
+ if (valueWriteIndex == chunks[chunkWriteIndex].buffer.limit()) {
+ valueWriteIndex = 0;
+ if (++chunkWriteIndex == chunks.length) {
+ return false;
+ }
+ }
+ chunks[chunkWriteIndex].buffer.putLong(value);
+ valueWriteIndex += 8;
+ return true;
+ }
+
+ @Override
+ public boolean putLong(int pos, long value) {
+ if (pos + 8 <= capacity) {
+ PositionDecoder.Coordinates coordinates = positionDecoder.decode(pos, chunkBoundaryMap);
+
+ if (coordinates.slot < chunks.length) {
+ chunks[coordinates.slot].buffer.putLong(coordinates.index, value);
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public long getLong(int pos) {
+ if (pos + 8 <= capacity) {
+ PositionDecoder.Coordinates coordinates = positionDecoder.decode(pos, chunkBoundaryMap);
+
+ return chunks[coordinates.slot].buffer.getLong(coordinates.index);
+ }
+ return Long.MIN_VALUE;
+ }
+
+ @Override
+ public LongIterator iterator() {
+ return new LongIterator() {
+ int valueReadIndex = 0;
+ int chunkReadIndex = -1;
+ int readBytes = 0;
+ byte computedHasNext = -1;
+ ByteBuffer currentBuffer = null;
+
+ @Override
+ public boolean hasNext() {
+ if (computedHasNext > -1) {
+ return computedHasNext == 1;
+ }
+
+ if (chunkReadIndex < 0) {
+ chunkReadIndex = 0;
+ currentBuffer = (ByteBuffer) chunks[chunkReadIndex].buffer.duplicate().flip();
+ }
+ if (currentBuffer.limit() != 0) {
+ if (chunkReadIndex <= chunkWriteIndex) {
+ if (valueReadIndex >= currentBuffer.limit()) {
+ chunkReadIndex++;
+ valueReadIndex = 0;
+ if (chunkReadIndex < chunks.length) {
+ currentBuffer = (ByteBuffer) chunks[chunkReadIndex].buffer.duplicate().flip();
+ if (currentBuffer.limit() != 0) {
+ computedHasNext = 1;
+ return true;
+ }
+ }
+ } else {
+ computedHasNext = 1;
+ return true;
+ }
+ }
+ }
+ currentBuffer = null;
+ computedHasNext = 0;
+ return false;
+ }
+
+ @Override
+ public long next() {
+ long value = currentBuffer.getLong();
+ valueReadIndex += 8;
+ readBytes += 8;
+ computedHasNext = -1;
+ return value;
+ }
+ };
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/direct/DirectAllocator.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/direct/DirectAllocator.java
new file mode 100644
index 00000000000..72c07a74bc2
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/direct/DirectAllocator.java
@@ -0,0 +1,271 @@
+package com.datadog.profiling.context.allocator.direct;
+
+import com.datadog.profiling.context.Allocator;
+import com.datadog.profiling.context.allocator.AllocatedBuffer;
+import datadog.trace.api.GlobalTracer;
+import datadog.trace.api.StatsDClient;
+import datadog.trace.api.Tracer;
+import datadog.trace.relocate.api.RatelimitedLogger;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantLock;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A naive implementation of a custom allocator using one big memory pool provided by a {@linkplain
+ * java.nio.DirectByteBuffer}. The allocator is splitting the provided buffer into chunks of the
+ * requested size and serves the allocation requests as the multiples of the chunk size.
+ * The free/occupied chunks are mapped in a single bitmap.
+ * This class is thread-safe.
+ * In order to improve the parallelism the implementation is using a version of striped locking when
+ * there are several subsets of chunks, each guarded by a separate lock. The number of subsets is
+ * set to be ~ the number of available CPU cores since there will never be more competing threads
+ * than that number.
+ * !!! IMPORTANT: This implementation tends to degrade in situation when the pool is almost
+ * exhausted. Therefore, this implementation is not suitable to be used in production as is and
+ * serves mostly as the starting point for a production ready version. !!!
+ */
+public final class DirectAllocator implements Allocator {
+ private static final Logger log = LoggerFactory.getLogger(DirectAllocator.class);
+ private static final RatelimitedLogger warnlog = new RatelimitedLogger(log, 30, TimeUnit.SECONDS);
+
+ private static final class AllocationResult {
+ static final AllocationResult EMPTY = new AllocationResult(0, 0);
+
+ final int allocatedChunks;
+ final int usedChunks;
+
+ AllocationResult(int allocatedChunks, int usedChunks) {
+ this.allocatedChunks = allocatedChunks;
+ this.usedChunks = usedChunks;
+ }
+ }
+
+ private static final int[] MASK_ARRAY = new int[8];
+
+ static {
+ int mask = 0x80;
+ for (int i = 0; i < 8; i++) {
+ MASK_ARRAY[i] = mask;
+ mask = mask >>> 1;
+ }
+ }
+
+ private final StatsDClient statsDClient;
+
+ private final ByteBuffer pool;
+ private final ByteBuffer memorymap;
+ private final int chunkSize;
+ private final int numChunks;
+ private final ReentrantLock[] memoryMapLocks;
+ private final int memoryMapLockCount;
+ private int lockSectionSize;
+ private final int bitmapSize;
+
+ private final long capacity;
+ private final AtomicLong allocatedBytes = new AtomicLong(0);
+ private final AtomicInteger lockSectionOffset = new AtomicInteger(0);
+
+ public DirectAllocator(int capacity, int chunkSize) {
+ log.warn(
+ "DirectAllocator is an experimental implementation. It should not be used in production.");
+ int cpus = Runtime.getRuntime().availableProcessors();
+
+ int targetNumChunks = (int) Math.ceil(capacity / (double) chunkSize);
+ this.lockSectionSize = (int) Math.ceil((targetNumChunks / (double) (cpus * 8))) * 8;
+ this.numChunks =
+ (int) (Math.ceil(targetNumChunks / (double) lockSectionSize) * lockSectionSize);
+ chunkSize = (int) Math.ceil(((double) capacity / (numChunks * chunkSize)) * chunkSize);
+ chunkSize = (int) Math.ceil(chunkSize / 8d) * 8;
+ int alignedCapacity = numChunks * chunkSize;
+ this.chunkSize = chunkSize;
+ this.pool = ByteBuffer.allocateDirect(alignedCapacity);
+ this.memorymap = ByteBuffer.allocateDirect((int) Math.ceil(numChunks / 8d));
+
+ this.memoryMapLockCount = (int) Math.ceil(numChunks / (double) lockSectionSize);
+ this.memoryMapLocks = new ReentrantLock[memoryMapLockCount];
+ for (int i = 0; i < memoryMapLockCount; i++) {
+ memoryMapLocks[i] = new ReentrantLock();
+ }
+ this.capacity = alignedCapacity;
+ this.bitmapSize = (int) Math.ceil(lockSectionSize / 8d);
+ StatsDClient statsd = StatsDClient.NO_OP;
+ try {
+ Tracer tracer = GlobalTracer.get();
+ Field fld = tracer.getClass().getDeclaredField("statsDClient");
+ fld.setAccessible(true);
+ statsd = (StatsDClient) fld.get(tracer);
+ log.debug("Set up custom StatsD Client instance {}", statsd);
+ } catch (Throwable t) {
+ t.printStackTrace();
+ }
+ statsDClient = statsd;
+ }
+
+ @Override
+ public int getChunkSize() {
+ return chunkSize;
+ }
+
+ @Override
+ public AllocatedBuffer allocate(int bytes) {
+ return allocateChunks((int) Math.ceil(bytes / (double) chunkSize));
+ }
+
+ @Override
+ public AllocatedBuffer allocateChunks(int chunks) {
+ long ts = System.nanoTime();
+ chunks = Math.min(chunks, numChunks);
+
+ long delta = chunkSize * (long) chunks;
+ long size = allocatedBytes.addAndGet(delta);
+ boolean exhausted = false;
+ while (size > capacity) {
+ long overflow = size - capacity;
+ long newDelta = delta - overflow;
+ chunks = (int) (newDelta / chunkSize);
+ size = allocatedBytes.addAndGet(chunks * (long) chunkSize - delta);
+ delta = newDelta;
+ exhausted = delta == 0;
+ }
+ if (exhausted) {
+ warnlog.warn("Capacity exhausted - buffer could not be allocated");
+ statsDClient.histogram("tracing.context.allocator.latency", System.nanoTime() - ts);
+ return null;
+ } else {
+ log.trace("Allocated {} chunks, new size={} ({})", chunks, size, this);
+ statsDClient.gauge("tracing.context.reserved.memory", size);
+ }
+ int lockSection = 0;
+ int offset = 0;
+ int allocated = 0;
+ Chunk[] chunkArray = new Chunk[chunks];
+ byte[] buffer = new byte[bitmapSize];
+ ReentrantLock sectionLock;
+
+ while (allocated < chunks) {
+ int offsetValue = 0;
+ int newOffsetValue = 0;
+ do {
+ offsetValue = lockSectionOffset.get();
+ newOffsetValue =
+ (offsetValue + 104729)
+ % memoryMapLocks
+ .length; // simple hashing using 10000th prime number and mod operation
+ } while (!lockSectionOffset.compareAndSet(offsetValue, newOffsetValue));
+ int idx;
+ for (idx = 0; idx < memoryMapLocks.length; idx++) {
+ lockSection = offsetValue % memoryMapLocks.length;
+ sectionLock = null;
+ try {
+ memoryMapLocks[lockSection].lock();
+ sectionLock = memoryMapLocks[lockSection];
+
+ int memorymapOffset = lockSection * bitmapSize;
+ for (int i = 0; i < bitmapSize; i++) {
+ buffer[i] = memorymap.get(memorymapOffset + i);
+ }
+ AllocationResult rslt =
+ allocateChunks(buffer, lockSection, chunkArray, chunks - allocated, offset);
+ offset += rslt.usedChunks;
+ allocated += rslt.allocatedChunks;
+ if (allocated == chunks) {
+ break;
+ }
+ } finally {
+ if (sectionLock != null) {
+ sectionLock.unlock();
+ }
+ }
+ }
+ }
+ statsDClient.histogram("tracing.context.allocator.latency", System.nanoTime() - ts);
+ return new DirectAllocatedBuffer(
+ chunkSize * allocated, chunkSize, Arrays.copyOf(chunkArray, offset));
+ }
+
+ private AllocationResult allocateChunks(
+ byte[] bitmap, int lockSection, Chunk[] chunks, int toAllocate, int offset) {
+ int allocated = 0;
+ int chunkCounter = 0;
+ int overlay = 0x00;
+ for (int index = 0; index < bitmap.length; index++) {
+ int slot = bitmap[index] & 0xff;
+ if (slot == 0xff) {
+ overlay = 0;
+ continue;
+ }
+ if ((overlay & 0x01) != 0) {
+ overlay = 0x01 << 8;
+ } else {
+ overlay = 0;
+ }
+ int mask = 0x80;
+ int bitIndex = 0;
+ while (slot != 0xff && allocated < toAllocate) {
+ if ((slot & mask) != 0) {
+ bitIndex++;
+ mask = mask >>> 1;
+ continue;
+ }
+ int previousMask = mask;
+ int lockOffset = lockSection * (lockSectionSize / 8);
+ int slotOffset = lockOffset + index;
+ slot |= mask;
+ memorymap.put(slotOffset, (byte) (slot & 0xff));
+
+ allocated++;
+ overlay |= mask;
+ mask = mask >>> 1;
+
+ int ref = (slotOffset * 8 + bitIndex++);
+ if (chunkCounter > 0) {
+ Chunk previous = chunks[chunkCounter + offset - 1];
+ if ((overlay & (previousMask << 1)) != 0) {
+ // can create a contiguous chunk
+ previous.extend();
+ continue;
+ }
+ }
+ pool.position(ref * chunkSize);
+ ByteBuffer sliced = pool.slice();
+ sliced.limit(chunkSize);
+ chunks[chunkCounter++ + offset] = new Chunk(this, sliced, ref);
+ }
+ }
+ return allocated == 0 ? AllocationResult.EMPTY : new AllocationResult(allocated, chunkCounter);
+ }
+
+ void release(int ref, int len) {
+ int chunks = len;
+ long delta = chunkSize * (long) len;
+ int offset = 0;
+ while (offset < len) {
+ int pos = (ref + offset) / 8;
+ int lockIndex = pos / lockSectionSize;
+ ReentrantLock lock = memoryMapLocks[lockIndex];
+ try {
+ lock.lock();
+ int slot = memorymap.get(pos);
+ int mask = 0;
+ int bitPos = (ref + offset) % 8;
+ int bitStop = Math.min(bitPos + len, 8);
+ for (int i = bitPos; i < bitStop; i++) {
+ mask |= MASK_ARRAY[i];
+ }
+ memorymap.put(pos, (byte) ((slot & ~mask) & 0xff));
+ len -= (bitStop - bitPos);
+ } finally {
+ lock.unlock();
+ }
+ }
+ long size = allocatedBytes.addAndGet(-delta);
+ log.trace("{} allocated chunks released - new size={} ({})", chunks, size, this);
+ statsDClient.gauge("tracing.context.reserved.memory", size);
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/heap/HeapAllocatedBuffer.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/heap/HeapAllocatedBuffer.java
new file mode 100644
index 00000000000..96707882456
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/heap/HeapAllocatedBuffer.java
@@ -0,0 +1,67 @@
+package com.datadog.profiling.context.allocator.heap;
+
+import com.datadog.profiling.context.LongIterator;
+import com.datadog.profiling.context.allocator.AllocatedBuffer;
+import java.nio.ByteBuffer;
+
+final class HeapAllocatedBuffer implements AllocatedBuffer {
+ private final HeapAllocator allocator;
+ private final ByteBuffer buffer;
+
+ HeapAllocatedBuffer(HeapAllocator allocator, int capacity) {
+ this.allocator = allocator;
+ buffer = ByteBuffer.allocate(capacity);
+ }
+
+ @Override
+ public void release() {
+ allocator.release(buffer.capacity());
+ }
+
+ @Override
+ public int capacity() {
+ return buffer.capacity();
+ }
+
+ @Override
+ public boolean putLong(long value) {
+ if (buffer.position() + 8 <= buffer.capacity()) {
+ buffer.putLong(value);
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public boolean putLong(int pos, long value) {
+ if (pos + 8 <= buffer.capacity()) {
+ buffer.putLong(pos, value);
+ return true;
+ }
+ return false;
+ }
+
+ @Override
+ public long getLong(int pos) {
+ if (pos + 8 <= buffer.capacity()) {
+ return buffer.getLong(pos);
+ }
+ return Long.MIN_VALUE;
+ }
+
+ @Override
+ public LongIterator iterator() {
+ ByteBuffer iterable = (ByteBuffer) buffer.duplicate().flip();
+ return new LongIterator() {
+ @Override
+ public boolean hasNext() {
+ return iterable.position() < iterable.limit();
+ }
+
+ @Override
+ public long next() {
+ return iterable.getLong();
+ }
+ };
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/heap/HeapAllocator.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/heap/HeapAllocator.java
new file mode 100644
index 00000000000..ab3062f927c
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java/com/datadog/profiling/context/allocator/heap/HeapAllocator.java
@@ -0,0 +1,97 @@
+package com.datadog.profiling.context.allocator.heap;
+
+import com.datadog.profiling.context.Allocator;
+import com.datadog.profiling.context.allocator.AllocatedBuffer;
+import datadog.trace.api.GlobalTracer;
+import datadog.trace.api.StatsDClient;
+import datadog.trace.api.Tracer;
+import datadog.trace.relocate.api.RatelimitedLogger;
+import java.lang.reflect.Field;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A straight-forward {@linkplain Allocator} implementation which satisfies the allocation requests
+ * by instantiating {@linkplain java.nio.HeapByteBuffer} instances.
+ */
+public final class HeapAllocator implements Allocator {
+ private static final Logger log = LoggerFactory.getLogger(HeapAllocator.class);
+ private static final RatelimitedLogger warnlog = new RatelimitedLogger(log, 30, TimeUnit.SECONDS);
+
+ private final int chunkSize;
+ private final AtomicLong remaining;
+ private final long topMemory;
+
+ private final StatsDClient statsDClient;
+
+ public HeapAllocator(int capacity, int chunkSize) {
+ this.chunkSize = chunkSize;
+ int numChunks = (int) Math.ceil(capacity / (double) chunkSize);
+ this.topMemory = numChunks * (long) chunkSize;
+ this.remaining = new AtomicLong(topMemory);
+
+ StatsDClient statsd = getStatsdClient();
+ statsDClient = statsd;
+ }
+
+ private static StatsDClient getStatsdClient() {
+ try {
+ Tracer tracer = GlobalTracer.get();
+ Field fld = tracer.getClass().getDeclaredField("statsDClient");
+ fld.setAccessible(true);
+ StatsDClient statsd = (StatsDClient) fld.get(tracer);
+ log.debug("Set up custom StatsD Client instance {}", statsd);
+ return statsd;
+ } catch (Throwable t) {
+ if (log.isDebugEnabled()) {
+ log.warn("Unable to obtain a StatsD client instance", t);
+ } else {
+ log.warn("Unable to obtain a StatsD client instance");
+ }
+ return StatsDClient.NO_OP;
+ }
+ }
+
+ @Override
+ public int getChunkSize() {
+ return chunkSize;
+ }
+
+ @Override
+ public AllocatedBuffer allocate(int size) {
+ long ts = System.nanoTime();
+ long newRemaining = 0;
+ try {
+ // align at chunkSize
+ size = (((size - 1) / chunkSize) + 1) * chunkSize;
+ newRemaining = remaining.addAndGet(-size);
+ while (newRemaining < 0) {
+ long restored = remaining.addAndGet(size); // restore the remaining size
+ size = (int) (newRemaining + size);
+ if (size <= chunkSize) {
+ warnlog.warn("Capacity exhausted - buffer could not be allocated");
+ return null;
+ }
+ // align at chunkSize
+ size = (((size - 1) / chunkSize) + 1) * chunkSize;
+ newRemaining = remaining.addAndGet(-size);
+ }
+ statsDClient.gauge("tracing.context.reserved.memory", topMemory - newRemaining);
+ return new HeapAllocatedBuffer(this, size);
+ } finally {
+ long timeDelta = System.nanoTime() - ts;
+ statsDClient.histogram("tracing.context.allocator.latency", timeDelta);
+ }
+ }
+
+ @Override
+ public AllocatedBuffer allocateChunks(int chunks) {
+ return allocate(chunks * chunkSize);
+ }
+
+ void release(int size) {
+ remaining.addAndGet(size);
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/main/java11/com/datadog/profiling/context/JfrTimestampPatchImpl.java b/dd-java-agent/agent-profiling/profiling-context/src/main/java11/com/datadog/profiling/context/JfrTimestampPatchImpl.java
new file mode 100644
index 00000000000..9f97cd56fd4
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/main/java11/com/datadog/profiling/context/JfrTimestampPatchImpl.java
@@ -0,0 +1,32 @@
+package com.datadog.profiling.context;
+
+import java.lang.instrument.Instrumentation;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import jdk.jfr.FlightRecorder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+final class JfrTimestampPatchImpl {
+ private static final Logger log = LoggerFactory.getLogger(JfrTimestampPatchImpl.class);
+
+ static void execute(ClassLoader agentClassLoader) {
+ try {
+ Class installerClass = agentClassLoader.loadClass("datadog.trace.agent.tooling.Utils");
+ Instrumentation instrumentation =
+ (Instrumentation) installerClass.getMethod("getInstrumentation").invoke(null);
+ Set myModules = Collections.singleton(JfrTimestampPatchImpl.class.getModule());
+ Module module = FlightRecorder.class.getModule();
+ instrumentation.redefineModule(
+ module,
+ Collections.emptySet(),
+ Map.of("jdk.jfr.internal", myModules),
+ Map.of("jdk.jfr.internal", myModules),
+ Collections.emptySet(),
+ Collections.emptyMap());
+ } catch (Throwable t) {
+ log.debug("Failed to patch the access to JFR timestamp", t);
+ }
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/AllocatorsTest.java b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/AllocatorsTest.java
new file mode 100644
index 00000000000..7f3fb5b4cb5
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/AllocatorsTest.java
@@ -0,0 +1,169 @@
+package com.datadog.profiling.context;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.datadog.profiling.context.allocator.AllocatedBuffer;
+import com.datadog.profiling.context.allocator.Allocators;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class AllocatorsTest {
+ private static final int maxChunks = 16;
+ private static final int chunkSize = 32;
+
+ private static final Allocator directAllocator =
+ Allocators.directAllocator(chunkSize * maxChunks, chunkSize);;
+ private static final Allocator heapAllocator =
+ Allocators.heapAllocator(chunkSize * maxChunks, chunkSize);
+
+ private AllocatedBuffer buffer = null;
+
+ @AfterEach
+ void teardown() {
+ if (buffer != null) {
+ buffer.release();
+ }
+ }
+
+ @ParameterizedTest
+ @MethodSource("allocators")
+ void testAllocations(Allocator allocator) {
+ AllocatedBuffer buffer1 = null;
+ AllocatedBuffer buffer2 = null;
+ AllocatedBuffer buffer3 = null;
+ try {
+ buffer1 = allocator.allocateChunks(maxChunks - 2);
+ buffer2 = allocator.allocate(chunkSize);
+ buffer3 = allocator.allocate(chunkSize / 2);
+ buffer = allocator.allocate(chunkSize / 4);
+
+ assertNotNull(buffer1);
+ assertNotNull(buffer2);
+ assertNotNull(buffer3);
+ assertNull(buffer);
+
+ assertEquals(chunkSize * (maxChunks - 2), buffer1.capacity());
+ assertEquals(chunkSize, buffer2.capacity());
+ assertEquals(chunkSize, buffer3.capacity());
+ } finally {
+ if (buffer1 != null) {
+ buffer1.release();
+ }
+ if (buffer2 != null) {
+ buffer2.release();
+ }
+ if (buffer3 != null) {
+ buffer3.release();
+ }
+ }
+ }
+
+ private static Stream allocators() {
+ return Stream.of(Arguments.of(heapAllocator), Arguments.of(directAllocator));
+ }
+
+ @Test
+ void testOverallocationDirect() {
+ buffer = directAllocator.allocateChunks(maxChunks);
+ assertNotNull(buffer);
+ assertNull(directAllocator.allocate(10));
+ }
+
+ @Test
+ void testOverallocationHeap() {
+ buffer = heapAllocator.allocateChunks(maxChunks);
+ assertNotNull(buffer);
+ assertNull(heapAllocator.allocate(10));
+ }
+
+ @Test
+ void testDirectBufferIterator() {
+ buffer = directAllocator.allocate(512);
+
+ LongIterator iterator = buffer.iterator();
+ assertFalse(iterator.hasNext());
+ assertFalse(iterator.hasNext());
+
+ buffer.putLong(0L);
+ iterator = buffer.iterator();
+ assertTrue(iterator.hasNext());
+ assertTrue(iterator.hasNext());
+
+ iterator.next();
+ assertFalse(iterator.hasNext());
+
+ // fill in data spanning multiple chunks
+ // there is already 1 value stored so we need to add only chunkSize other values
+ for (int i = 0; i < chunkSize; i++) {
+ buffer.putLong(i + 1);
+ }
+ iterator = buffer.iterator();
+ int cnt = 0;
+ while (iterator.hasNext()) {
+ assertNotNull(iterator.next());
+ cnt++;
+ }
+ assertEquals(chunkSize + 1, cnt);
+ }
+
+ @Test
+ void testHeapBufferIterator() {
+ buffer = heapAllocator.allocate(256);
+
+ LongIterator iterator = buffer.iterator();
+ assertFalse(iterator.hasNext());
+ assertFalse(iterator.hasNext());
+
+ buffer.putLong(0L);
+ iterator = buffer.iterator();
+ assertTrue(iterator.hasNext());
+ assertTrue(iterator.hasNext());
+
+ iterator.next();
+ assertFalse(iterator.hasNext());
+ }
+
+ @ParameterizedTest
+ @MethodSource("allocateBufferTestParams")
+ void testAllocatedBuffer(Allocator allocator, int numChunks, int skid) {
+ buffer =
+ skid == 0
+ ? allocator.allocateChunks(numChunks)
+ : allocator.allocate(chunkSize * numChunks + skid);
+ int iterations = (numChunks * chunkSize) / 8;
+ for (int i = 0; i < iterations; i++) {
+ assertTrue(buffer.putLong(i));
+ }
+ assertTrue(numChunks * chunkSize + skid <= buffer.capacity());
+
+ for (int i = 0; i < numChunks; i++) {
+ assertTrue(buffer.putLong(i * chunkSize, 5));
+ assertEquals(5, buffer.getLong(i * chunkSize));
+ }
+
+ if (skid == 0) {
+ assertFalse(buffer.putLong(6));
+ } else {
+ assertTrue(buffer.putLong(6));
+ }
+ assertFalse(buffer.putLong(buffer.capacity() - 7, 6));
+
+ assertEquals(Long.MIN_VALUE, buffer.getLong(buffer.capacity() - 7));
+ }
+
+ private static Stream allocateBufferTestParams() {
+ return Stream.of(
+ Arguments.of(directAllocator, 1, 0),
+ Arguments.of(directAllocator, 2, 0),
+ Arguments.of(directAllocator, 1, 1),
+ Arguments.of(directAllocator, 2, 1),
+ Arguments.of(heapAllocator, 1, 0),
+ Arguments.of(heapAllocator, 2, 0),
+ Arguments.of(heapAllocator, 1, 1),
+ Arguments.of(heapAllocator, 2, 1));
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/IntervalEncoderTest.java b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/IntervalEncoderTest.java
new file mode 100644
index 00000000000..2a47d26d212
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/IntervalEncoderTest.java
@@ -0,0 +1,84 @@
+package com.datadog.profiling.context;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.nio.ByteBuffer;
+import java.time.Instant;
+import java.util.List;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class IntervalEncoderTest {
+ private Instant now;
+ private IntervalEncoder instance;
+
+ @BeforeEach
+ void setup() {
+ now = Instant.now();
+ instance = new IntervalEncoder(now.toEpochMilli(), 1000, 2, 32);
+ }
+
+ @Test
+ void testFinishUnstarted() {
+ assertNotNull(instance.finish());
+ }
+
+ @Test
+ void testFinishThreadInFlight() {
+ IntervalEncoder.ThreadEncoder encoder = instance.startThread(1);
+ assertThrows(IllegalStateException.class, instance::finish);
+ }
+
+ @Test
+ void testDoubleFinish() {
+ instance.finish();
+ assertThrows(IllegalStateException.class, instance::finish);
+ }
+
+ @Test
+ void testStartTooManyThreads() {
+ instance.startThread(1).finish();
+ instance.startThread(2).finish();
+ assertThrows(IllegalStateException.class, () -> instance.startThread(3));
+ }
+
+ @Test
+ void testStartMultiThreads() {
+ instance.startThread(1);
+ assertThrows(IllegalStateException.class, () -> instance.startThread(2));
+
+ IntervalEncoder instance1 = new IntervalEncoder(System.currentTimeMillis(), 1000, 1, 100);
+ instance1.startThread(1);
+ assertThrows(IllegalStateException.class, () -> instance1.startThread(2));
+ }
+
+ @Test
+ void testEncodeSequence() {
+ IntervalEncoder.ThreadEncoder encoder = instance.startThread(1);
+ encoder.recordInterval(200, 300);
+ encoder.recordInterval(500, 600);
+ ByteBuffer data = encoder.finish().finish();
+ assertNotNull(data);
+ assertTrue(data.limit() > 0);
+
+ List intervals = new IntervalParser().parseIntervals(data.array());
+ IntervalParser.Interval i1 = intervals.get(0);
+ assertEquals(now.toEpochMilli() * 1_000_000L + 200, i1.from);
+ }
+
+ @Test
+ void testIntervalAfterFinish() {
+ IntervalEncoder.ThreadEncoder encoder = instance.startThread(1);
+ encoder.recordInterval(200, 300);
+ encoder.finish();
+ assertThrows(IllegalStateException.class, () -> encoder.recordInterval(500, 600));
+ }
+
+ @Test
+ void testThreadDoubleFinish() {
+ IntervalEncoder.ThreadEncoder encoder = instance.startThread(1);
+ encoder.recordInterval(200, 300);
+ encoder.finish();
+ assertThrows(IllegalStateException.class, encoder::finish);
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/IntervalParser.java b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/IntervalParser.java
new file mode 100644
index 00000000000..090a82d2b4f
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/IntervalParser.java
@@ -0,0 +1,138 @@
+package com.datadog.profiling.context;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public final class IntervalParser {
+ public static final class Interval {
+ public final long threadId;
+ public final long from;
+ public final long till;
+
+ public Interval(long threadId, long from, long till) {
+ this.threadId = threadId;
+ this.from = from;
+ this.till = till;
+ }
+ }
+
+ private static class GroupVarintIterator {
+ private final ByteBuffer dataBuffer;
+ private final ByteBuffer bitmapBuffer;
+
+ private int maskIndex = 0;
+ private int maskOffset = 0;
+
+ GroupVarintIterator(ByteBuffer dataBuffer) {
+ int index = dataBuffer.getInt();
+ this.dataBuffer = dataBuffer.slice();
+ this.bitmapBuffer = slice(dataBuffer, index);
+ }
+
+ private static ByteBuffer slice(ByteBuffer src, int index) {
+ /*
+ ByteBuffer.slice(index, size) is available only from JDK 13
+ We need to emulate that by storing current position, setting the position to index, slicing and restoring the original position
+ */
+ int currentPos = src.position();
+ try {
+ src.position(index);
+ return src.slice();
+ } finally {
+ src.position(currentPos);
+ }
+ }
+
+ boolean hasNext() {
+ return dataBuffer.position() < dataBuffer.capacity() && maskIndex < bitmapBuffer.capacity();
+ }
+
+ long next() {
+ int mask = bitmapBuffer.getInt(maskIndex);
+ int size = ((mask >>> (29 - maskOffset)) & 0x7) + 1;
+ if ((maskOffset += 3) > 21) {
+ maskIndex += 3;
+ maskOffset = 0;
+ }
+ long value = 0L;
+ for (int i = 0; i < size; i++) {
+ int item = (int) dataBuffer.get() & 0xff;
+ value = (value << 8) + item;
+ }
+ return value;
+ }
+ }
+
+ public List parseIntervals(byte[] intervalData) {
+ List result = new ArrayList<>();
+
+ parseIntervals(
+ intervalData,
+ interval -> {
+ result.add(interval);
+ return true;
+ });
+ return result;
+ }
+
+ public interface IntervalConsumer {
+ boolean accept(Interval interval);
+ }
+
+ public void parseIntervals(byte[] intervalData, IntervalConsumer consumer) {
+ ByteBuffer buffer = ByteBuffer.wrap(intervalData);
+ int chunkDataOffset = buffer.getInt();
+ long timestampMillis = getVarint(buffer);
+ long frequencyMultiplier = getVarint(buffer);
+ int numThreads = (int) getVarint(buffer);
+
+ ByteBuffer dataChunk =
+ ByteBuffer.wrap(intervalData, chunkDataOffset, intervalData.length - chunkDataOffset)
+ .slice();
+
+ GroupVarintIterator iterator = new GroupVarintIterator(dataChunk);
+ for (int thread = 0; thread < numThreads; thread++) {
+ long previousTimestamp = 0;
+ long threadId = getVarint(buffer);
+ int intervals = (int) getVarint(buffer);
+ for (int interval = 0; interval < intervals; interval++) {
+ if (iterator.hasNext()) {
+ long startTsDelta = iterator.next();
+ long endTsDelta = iterator.next();
+ long startTs = previousTimestamp + startTsDelta;
+ long endTs = startTs + endTsDelta;
+ if (!consumer.accept(
+ new Interval(
+ threadId,
+ toEpochNanos(timestampMillis, startTs, frequencyMultiplier),
+ toEpochNanos(timestampMillis, endTs, frequencyMultiplier)))) {
+ // consumer not interested in the rest of the data
+ return;
+ }
+ previousTimestamp = endTs;
+ } else {
+ throw new IllegalStateException();
+ }
+ }
+ }
+ }
+
+ private static long toEpochNanos(long startMs, long ts, long frequencyMultiplier) {
+ return TimeUnit.NANOSECONDS.convert(startMs, TimeUnit.MILLISECONDS)
+ + ((ts * 1000) / frequencyMultiplier);
+ }
+
+ private long getVarint(ByteBuffer buffer) {
+ long ret = 0;
+ for (int i = 0; i < 8; i++) {
+ byte b = buffer.get();
+ ret += (long) ((b & 0x7FL)) << (7 * i);
+ if (b >= 0) {
+ return ret;
+ }
+ }
+ return ret + ((buffer.get() & 0xFFL) << 56);
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/IntervalSequencePrunerTest.java b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/IntervalSequencePrunerTest.java
new file mode 100644
index 00000000000..a6c90074533
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/IntervalSequencePrunerTest.java
@@ -0,0 +1,286 @@
+package com.datadog.profiling.context;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import com.datadog.profiling.context.allocator.Allocators;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class IntervalSequencePrunerTest {
+ private IntervalSequencePruner instance;
+
+ @BeforeEach
+ void setup() {
+ instance = new IntervalSequencePruner();
+ }
+
+ @Test
+ void testPruneStartStop() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long stop = ProfilerTracingContextTracker.maskDeactivation(200, false);
+ ls.add(start);
+ ls.add(stop);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+ assertFalse(iterator.hasNext());
+ assertEquals(Arrays.asList(start, stop), values);
+ }
+
+ @Test
+ void testPruneMaybeStop() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long maybeStop = ProfilerTracingContextTracker.maskDeactivation(200, true);
+ ls.add(start);
+ ls.add(maybeStop);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+ assertFalse(iterator.hasNext());
+ assertEquals(Arrays.asList(start, maybeStop), values);
+ }
+
+ @Test
+ void testPruneMaybeStopWithStop() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long maybeStop = ProfilerTracingContextTracker.maskDeactivation(200, true);
+ long stop = ProfilerTracingContextTracker.maskDeactivation(300, false);
+
+ ls.add(start);
+ ls.add(maybeStop);
+ ls.add(stop);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+ assertFalse(iterator.hasNext());
+ assertEquals(Arrays.asList(start, stop), values);
+ }
+
+ @Test
+ void testPruneMaybeStopWithRestartAndStop() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long maybeStop = ProfilerTracingContextTracker.maskDeactivation(200, true);
+ long restart = ProfilerTracingContextTracker.maskActivation(300);
+ long stop = ProfilerTracingContextTracker.maskDeactivation(400, false);
+
+ ls.add(start);
+ ls.add(maybeStop);
+ ls.add(restart);
+ ls.add(stop);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+ assertFalse(iterator.hasNext());
+ assertEquals(Arrays.asList(start, stop), values);
+ }
+
+ @Test
+ void testPruneMaybeStopWithRestartDangling() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long maybeStop = ProfilerTracingContextTracker.maskDeactivation(200, true);
+ long restart = ProfilerTracingContextTracker.maskActivation(300);
+
+ ls.add(start);
+ ls.add(maybeStop);
+ ls.add(restart);
+
+ long finishTs = 400;
+ LongIterator iterator = instance.pruneIntervals(ls, finishTs);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+ assertFalse(iterator.hasNext());
+ // the deactivation at 'finishTs' is added by the pruning not to have a dangling start
+ assertEquals(
+ Arrays.asList(start, ProfilerTracingContextTracker.maskDeactivation(finishTs, false)),
+ values);
+ }
+
+ @Test
+ void testPruneMaybeStopAfterStop() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long stop = ProfilerTracingContextTracker.maskDeactivation(200, false);
+ long maybeStop = ProfilerTracingContextTracker.maskDeactivation(300, true);
+
+ ls.add(start);
+ ls.add(stop);
+ ls.add(maybeStop);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+ assertFalse(iterator.hasNext());
+ assertEquals(Arrays.asList(start, maybeStop), values);
+ }
+
+ @Test
+ void testPruneStartAfterMultiStop() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long stop1 = ProfilerTracingContextTracker.maskDeactivation(200, false);
+ long stop2 = ProfilerTracingContextTracker.maskDeactivation(300, false);
+ long stop3 = ProfilerTracingContextTracker.maskDeactivation(400, false);
+ long start1 = ProfilerTracingContextTracker.maskActivation(500);
+ long stop4 = ProfilerTracingContextTracker.maskDeactivation(600, false);
+
+ ls.add(start);
+ ls.add(stop1);
+ ls.add(stop2);
+ ls.add(stop3);
+ ls.add(start1);
+ ls.add(stop4);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+ assertFalse(iterator.hasNext());
+ assertEquals(Arrays.asList(start, stop3, start1, stop4), values);
+ }
+
+ @Test
+ void testPruneMaybeStopAfterStopMulti() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long stop = ProfilerTracingContextTracker.maskDeactivation(200, false);
+ long maybeStop = ProfilerTracingContextTracker.maskDeactivation(300, true);
+ long stop1 = ProfilerTracingContextTracker.maskDeactivation(400, false);
+ long stop2 = ProfilerTracingContextTracker.maskDeactivation(500, false);
+
+ ls.add(start);
+ ls.add(stop);
+ ls.add(maybeStop);
+ ls.add(stop1);
+ ls.add(stop2);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+ assertFalse(iterator.hasNext());
+ assertEquals(Arrays.asList(start, stop2), values);
+ }
+
+ @Test
+ void testPruneMultiStartStop() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long start1 = ProfilerTracingContextTracker.maskActivation(200);
+ long stop = ProfilerTracingContextTracker.maskDeactivation(300, false);
+
+ ls.add(start);
+ ls.add(start1);
+ ls.add(stop);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+ assertFalse(iterator.hasNext());
+ assertEquals(Arrays.asList(start, stop), values);
+ }
+
+ @Test
+ void testPruneMaybeStopMulti() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long maybeStop = ProfilerTracingContextTracker.maskDeactivation(200, true);
+ long maybeStop2 = ProfilerTracingContextTracker.maskDeactivation(300, true);
+ long maybeStop3 = ProfilerTracingContextTracker.maskDeactivation(400, true);
+
+ ls.add(start);
+ ls.add(maybeStop);
+ ls.add(maybeStop2);
+ ls.add(maybeStop3);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+ assertFalse(iterator.hasNext());
+ assertEquals(Arrays.asList(start, maybeStop3), values);
+ }
+
+ @Test
+ void testPruneStopWithNoStart() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long stop = ProfilerTracingContextTracker.maskDeactivation(100, false);
+
+ ls.add(stop);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ assertFalse(iterator.hasNext());
+ }
+
+ @Test
+ void testPruneMaybeStopWithNoStart() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long maybeStop = ProfilerTracingContextTracker.maskDeactivation(100, true);
+
+ ls.add(maybeStop);
+
+ LongIterator iterator = instance.pruneIntervals(ls, 0);
+ assertFalse(iterator.hasNext());
+ }
+
+ @Test
+ void testPruneStartWithNoStop() {
+ LongSequence ls = new LongSequence(Allocators.heapAllocator(1024, 32));
+ long start = ProfilerTracingContextTracker.maskActivation(100);
+ long stopTs = 200;
+
+ ls.add(start);
+
+ LongIterator iterator = instance.pruneIntervals(ls, stopTs);
+ List values = new ArrayList<>();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ values.add(iterator.next());
+ }
+
+ assertFalse(iterator.hasNext());
+ assertEquals(
+ Arrays.asList(start, ProfilerTracingContextTracker.maskDeactivation(stopTs, false)),
+ values);
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/LEB128SupportTest.java b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/LEB128SupportTest.java
new file mode 100644
index 00000000000..42da5c9d45d
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/LEB128SupportTest.java
@@ -0,0 +1,75 @@
+package com.datadog.profiling.context;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.nio.ByteBuffer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class LEB128SupportTest {
+ private LEB128Support instance;
+
+ @BeforeEach
+ void setup() {
+ instance = new LEB128Support();
+ }
+
+ @Test
+ void align() {
+ assertEquals(0, instance.align(0, 3));
+ assertEquals(3, instance.align(1, 3));
+ assertEquals(3, instance.align(3, 3));
+ }
+
+ @Test
+ void varintSize() {
+ ByteBuffer buffer = ByteBuffer.allocate(10);
+ long value = 0L;
+ instance.putVarint(buffer, value);
+ int size = instance.varintSize(value);
+ assertEquals(1, size);
+ assertEquals(buffer.position(), size);
+
+ buffer.rewind();
+ value = 1L;
+ instance.putVarint(buffer, value);
+ size = instance.varintSize(value);
+ assertEquals(1, size);
+ assertEquals(buffer.position(), size);
+
+ value = 0x40;
+ for (int i = 1; i < 10; i++) {
+ size = instance.varintSize(value);
+ buffer.rewind();
+ instance.putVarint(buffer, value);
+ assertEquals(i, size);
+ assertEquals(buffer.position(), size);
+
+ if (i == 9) {
+ value = value << 1;
+ size = instance.varintSize(value);
+ buffer.rewind();
+ instance.putVarint(buffer, value);
+ assertEquals(i, size);
+ assertEquals(buffer.position(), size);
+ } else {
+ value = value << 7;
+ }
+ }
+ }
+
+ @Test
+ void longSize() {
+ long value = 0L;
+ int size = instance.longSize(value);
+ assertEquals(1, size);
+
+ value = 0x80;
+ for (int i = 1; i < 9; i++) {
+ size = instance.longSize(value);
+ assertEquals(i, size);
+
+ value = value << 8;
+ }
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/LongSequenceTest.java b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/LongSequenceTest.java
new file mode 100644
index 00000000000..42e90866f0f
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/LongSequenceTest.java
@@ -0,0 +1,96 @@
+package com.datadog.profiling.context;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.datadog.profiling.context.allocator.Allocators;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class LongSequenceTest {
+ private LongSequence instance;
+
+ @BeforeEach
+ void setup() throws Exception {
+ instance = new LongSequence(Allocators.heapAllocator(512, 32));
+ }
+
+ @Test
+ void testCapacity() {
+ int items = 64;
+ for (int i = 0; i < items; i++) {
+ try {
+ assertTrue(instance.add(i) > 0);
+ } catch (Throwable t) {
+ System.out.println("===> " + i);
+ throw t;
+ }
+ }
+ assertFalse(instance.add(65) > 0);
+ assertEquals(items, instance.size());
+ LongIterator iterator = instance.iterator();
+ long value = 0;
+ while (iterator.hasNext()) {
+ long retrieved = iterator.next();
+ assertEquals(value++, retrieved);
+ }
+ }
+
+ @Test
+ void testDoubleRelease() {
+ instance.release();
+ instance.release();
+ }
+
+ @Test
+ void testPositionalSetGet() {
+ for (int i = 1; i <= 3; i++) {
+ instance.add(i);
+ }
+
+ for (int i = 1; i <= 3; i++) {
+ assertEquals(i, instance.get(i - 1));
+ instance.set(i - 1, 2 * i);
+ assertEquals(2 * i, instance.get(i - 1));
+ }
+ }
+
+ @Test
+ void testGetInvalidIndex() {
+ assertEquals(Long.MIN_VALUE, instance.get(5));
+ assertEquals(Long.MIN_VALUE, instance.get(65));
+ }
+
+ @Test
+ void testSetInvalidIndex() {
+ assertFalse(instance.set(65, 0L));
+ }
+
+ @Test
+ void testIteratorInvalid() {
+ LongIterator iterator = instance.iterator();
+ assertThrows(IllegalStateException.class, iterator::next);
+ }
+
+ @Test
+ void testIteratorEmpty() {
+ LongIterator iterator = instance.iterator();
+ assertFalse(iterator.hasNext());
+ }
+
+ @Test
+ void testIterator() {
+ // fill in data
+ for (int i = 0; i < 64; i++) {
+ instance.add(i);
+ }
+
+ long expected = 0;
+ LongIterator iterator = instance.iterator();
+ while (iterator.hasNext()) {
+ assertTrue(iterator.hasNext());
+ assertEquals(expected++, iterator.next());
+ }
+ assertEquals(expected, instance.size());
+ assertFalse(iterator.hasNext());
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/PositionDecoderTest.java b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/PositionDecoderTest.java
new file mode 100644
index 00000000000..2972a1a6e0c
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/PositionDecoderTest.java
@@ -0,0 +1,69 @@
+package com.datadog.profiling.context;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class PositionDecoderTest {
+ private PositionDecoder instance;
+
+ @BeforeEach
+ void setup() {
+ instance = PositionDecoder.getInstance();
+ }
+
+ @Test
+ void decodeInvalidCapacityMap() {
+ int[] map = new int[0];
+
+ assertNull(instance.decode(0, map, -1));
+ assertNull(instance.decode(0, map, 0));
+ assertNull(instance.decode(0, map, 1));
+ }
+
+ @Test
+ void decodeInvalidCapacityMapSize() {
+ int[] map = new int[] {3, 10, 25};
+
+ assertNull(instance.decode(0, map, -1));
+ assertNull(instance.decode(0, map, map.length + 1));
+ }
+
+ @Test
+ void decodeValid() {
+ int maxBoundary = 12;
+ int[] map = new int[] {2, 4, 6, 8, 10, maxBoundary};
+ PositionDecoder.Coordinates[] expected =
+ new PositionDecoder.Coordinates[] {
+ new PositionDecoder.Coordinates(0, 0),
+ new PositionDecoder.Coordinates(0, 1),
+ new PositionDecoder.Coordinates(0, 2),
+ new PositionDecoder.Coordinates(1, 0),
+ new PositionDecoder.Coordinates(1, 1),
+ new PositionDecoder.Coordinates(2, 0),
+ new PositionDecoder.Coordinates(2, 1),
+ new PositionDecoder.Coordinates(3, 0),
+ new PositionDecoder.Coordinates(3, 1),
+ new PositionDecoder.Coordinates(4, 0),
+ new PositionDecoder.Coordinates(4, 1),
+ new PositionDecoder.Coordinates(5, 0),
+ new PositionDecoder.Coordinates(5, 1)
+ };
+
+ for (int mapIndex = map.length - 1; mapIndex >= 0; mapIndex--) {
+ int limit = map[mapIndex];
+ for (int pos = 0; pos <= limit; pos++) {
+ assertEquals(
+ expected[pos],
+ instance.decode(pos, map, mapIndex + 1),
+ "Failed to decode position "
+ + pos
+ + " with section boundary limit set to "
+ + (mapIndex + 1));
+ }
+ }
+
+ assertNull(instance.decode(map[map.length - 1] + 1, map));
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/ProfilerTracingContextTrackerFactoryTest.java b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/ProfilerTracingContextTrackerFactoryTest.java
new file mode 100644
index 00000000000..2f25394f957
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/ProfilerTracingContextTrackerFactoryTest.java
@@ -0,0 +1,68 @@
+package com.datadog.profiling.context;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import datadog.trace.api.profiling.TracingContextTracker;
+import datadog.trace.api.profiling.TracingContextTrackerFactory;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+
+class ProfilerTracingContextTrackerFactoryTest {
+ @Test
+ void testTracingAvailable() {
+ assertFalse(TracingContextTrackerFactory.isTrackingAvailable());
+
+ TracingContextTrackerFactory.registerImplementation(
+ new ProfilerTracingContextTrackerFactory(
+ TimeUnit.NANOSECONDS.convert(100, TimeUnit.MILLISECONDS), 10L, 512));
+ assertTrue(TracingContextTrackerFactory.isTrackingAvailable());
+ }
+
+ @Test
+ void testReleasedAfterInactivity() throws Exception {
+ long inactivityMs = 100;
+ ProfilerTracingContextTrackerFactory instance =
+ new ProfilerTracingContextTrackerFactory(
+ TimeUnit.NANOSECONDS.convert(inactivityMs, TimeUnit.MILLISECONDS), 10L, 512);
+ TracingContextTracker tracker1 = instance.instance(null);
+ TracingContextTracker tracker2 = instance.instance(null);
+
+ Thread.sleep((long) (inactivityMs * 1.5d));
+ assertFalse(
+ tracker1.release(), "Tracker resources were not released within " + inactivityMs + "ms");
+ assertFalse(
+ tracker2.release(), "Tracker resources were not released within " + inactivityMs + "ms");
+ }
+
+ @Test
+ void testNotReleasedBeforeInactivity() throws Exception {
+ long inactivityMs = 100;
+ ProfilerTracingContextTrackerFactory instance =
+ new ProfilerTracingContextTrackerFactory(
+ TimeUnit.NANOSECONDS.convert(inactivityMs, TimeUnit.MILLISECONDS), 10L, 512);
+ TracingContextTracker tracker1 = instance.instance(null);
+ TracingContextTracker tracker2 = instance.instance(null);
+
+ Thread.sleep((long) (inactivityMs * 0.5d));
+ assertTrue(
+ tracker1.release(),
+ "Tracker resources were erroneously released before " + inactivityMs + "ms");
+ assertTrue(
+ tracker2.release(),
+ "Tracker resources were erroneously released before " + inactivityMs + "ms");
+ }
+
+ @Test
+ void testNoInactivityRelease() throws Exception {
+ long inactivityMs = 0;
+ ProfilerTracingContextTrackerFactory instance =
+ new ProfilerTracingContextTrackerFactory(
+ TimeUnit.NANOSECONDS.convert(inactivityMs, TimeUnit.MILLISECONDS), 10L, 512);
+ TracingContextTracker tracker1 = instance.instance(null);
+ TracingContextTracker tracker2 = instance.instance(null);
+
+ Thread.sleep((long) (inactivityMs * 1.5d));
+ assertTrue(tracker1.release(), "Tracker resources are erroneously released");
+ assertTrue(tracker2.release(), "Tracker resources are erroneously released");
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/ProfilerTracingContextTrackerTest.java b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/ProfilerTracingContextTrackerTest.java
new file mode 100644
index 00000000000..0c3f836f559
--- /dev/null
+++ b/dd-java-agent/agent-profiling/profiling-context/src/test/java/com/datadog/profiling/context/ProfilerTracingContextTrackerTest.java
@@ -0,0 +1,212 @@
+package com.datadog.profiling.context;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import com.datadog.profiling.context.allocator.Allocators;
+import datadog.trace.api.profiling.TracingContextTracker;
+import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.Base64;
+import java.util.List;
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+
+class ProfilerTracingContextTrackerTest {
+ private static final class TestTimeTickProvider
+ implements ProfilerTracingContextTracker.TimeTicksProvider {
+ private final long step;
+ private final long frequency;
+ private final AtomicLong ts = new AtomicLong(0);
+
+ public TestTimeTickProvider(long initial, long step, long frequency) {
+ this.ts.set(initial);
+ this.step = step;
+ this.frequency = frequency;
+ }
+
+ @Override
+ public long ticks() {
+ return ts.getAndAdd(step);
+ }
+
+ @Override
+ public long frequency() {
+ return frequency;
+ }
+
+ public void set(long ticks) {
+ ts.set(ticks);
+ }
+ }
+
+ private final IntervalSequencePruner sequencePruner = new IntervalSequencePruner();
+ private ProfilerTracingContextTracker instance;
+
+ @BeforeEach
+ void setup() throws Exception {
+ ProfilerTracingContextTracker.TimeTicksProvider ticker =
+ new TestTimeTickProvider(100_000L, 1000L, 1_000_000_000L);
+ instance =
+ new ProfilerTracingContextTracker(
+ Allocators.heapAllocator(32, 16), null, ticker, sequencePruner);
+ }
+
+ @Test
+ void deactivateContext() {}
+
+ @Test
+ void persist() {
+ TestTimeTickProvider tickProvider = new TestTimeTickProvider(100_000L, 3, 1_000_000_000L);
+ instance =
+ new ProfilerTracingContextTracker(
+ Allocators.directAllocator(8192, 64), null, tickProvider, sequencePruner);
+ for (int i = 0; i < 40; i += 4) {
+ instance.activateContext(1L, (i + 1) * 1_000_000L);
+ instance.deactivateContext(1L, (i + 2) * 1_000_000L, false);
+ instance.activateContext(2L, (i + 3) * 1_000_000L);
+ instance.deactivateContext(2L, (i + 4) * 1_000_000L, true);
+ }
+ // set the timestamp after the last transition's timestamp
+ tickProvider.set((44 * 1_000_000L) + 1);
+
+ byte[] persisted = instance.persist();
+ assertNotNull(persisted);
+
+ List intervals = new IntervalParser().parseIntervals(persisted);
+ assertEquals(11, intervals.size());
+
+ byte[] encoded = Base64.getEncoder().encode(persisted);
+ System.out.println("===> encoded: " + encoded.length);
+ System.err.println("===> " + new String(encoded, StandardCharsets.UTF_8));
+ }
+
+ @Test
+ void testSanity() {
+ ProfilerTracingContextTrackerFactory instance =
+ new ProfilerTracingContextTrackerFactory(-1, 10L, 512);
+
+ ProfilerTracingContextTracker tracker =
+ (ProfilerTracingContextTracker) instance.instance(AgentTracer.NoopAgentSpan.INSTANCE);
+ AtomicInteger blobCounter = new AtomicInteger();
+
+ tracker.activateContext();
+ tracker.maybeDeactivateContext();
+ tracker.deactivateContext();
+ byte[] data1 = tracker.persist();
+ assertNotNull(data1);
+ byte[] data2 = tracker.persist();
+ assertNotNull(data2);
+ assertArrayEquals(data1, data2);
+
+ assertTrue(tracker.release());
+ assertArrayEquals(data1, tracker.persist()); // previously persisted data survives release
+ assertFalse(tracker.release()); // double release returns 'false'
+ assertArrayEquals(
+ data1, tracker.persist()); // previously persisted data survives double-release
+ }
+
+ @Test
+ void delayedCompareToTest() {
+ ProfilerTracingContextTrackerFactory instance =
+ new ProfilerTracingContextTrackerFactory(-1, 10L, 512);
+
+ ProfilerTracingContextTracker tracker1 =
+ (ProfilerTracingContextTracker) instance.instance(AgentTracer.NoopAgentSpan.INSTANCE);
+ ProfilerTracingContextTracker tracker2 =
+ (ProfilerTracingContextTracker) instance.instance(AgentTracer.NoopAgentSpan.INSTANCE);
+
+ TracingContextTracker.DelayedTracker delayed1 = tracker1.asDelayed();
+ TracingContextTracker.DelayedTracker delayed2 = tracker2.asDelayed();
+
+ Delayed other =
+ new Delayed() {
+ @Override
+ public long getDelay(TimeUnit unit) {
+ return 0;
+ }
+
+ @Override
+ public int compareTo(Delayed o) {
+ return 0;
+ }
+ };
+
+ assertEquals(0, delayed1.compareTo(other));
+ assertEquals(0, delayed1.compareTo(delayed1));
+
+ tracker2.activateContext();
+ assertEquals(-1, delayed1.compareTo(delayed2));
+ assertEquals(1, delayed2.compareTo(delayed1));
+ }
+
+ @Test
+ void delayedCachingTest() {
+ ProfilerTracingContextTrackerFactory instance =
+ new ProfilerTracingContextTrackerFactory(-1, 10L, 512);
+
+ ProfilerTracingContextTracker tracker1 =
+ (ProfilerTracingContextTracker) instance.instance(AgentTracer.NoopAgentSpan.INSTANCE);
+ ProfilerTracingContextTracker tracker2 =
+ (ProfilerTracingContextTracker) instance.instance(AgentTracer.NoopAgentSpan.INSTANCE);
+
+ TracingContextTracker.DelayedTracker delayed1 = tracker1.asDelayed();
+ TracingContextTracker.DelayedTracker delayed2 = tracker2.asDelayed();
+
+ assertEquals(delayed1, tracker1.asDelayed());
+ assertEquals(delayed2, tracker2.asDelayed());
+ }
+
+ @Test
+ void delayedReleaseTest() {
+ ProfilerTracingContextTrackerFactory instance =
+ new ProfilerTracingContextTrackerFactory(-1, 10L, 512);
+
+ ProfilerTracingContextTracker tracker1 =
+ (ProfilerTracingContextTracker) instance.instance(AgentTracer.NoopAgentSpan.INSTANCE);
+ ProfilerTracingContextTracker tracker2 =
+ (ProfilerTracingContextTracker) instance.instance(AgentTracer.NoopAgentSpan.INSTANCE);
+
+ TracingContextTracker.DelayedTracker delayed1 = tracker1.asDelayed();
+ TracingContextTracker.DelayedTracker delayed2 = tracker2.asDelayed();
+
+ // make sure the tracker release will remove the reference from delayed to the tracker
+ tracker1.release();
+ assertNull(((ProfilerTracingContextTracker.DelayedTrackerImpl) delayed1).ref);
+ assertNotEquals(delayed1, tracker1.asDelayed());
+
+ // make sure that the delayed cleanup will also trigger the tracker release
+ delayed2.cleanup();
+ assertNull(((ProfilerTracingContextTracker.DelayedTrackerImpl) delayed2).ref);
+ assertNotEquals(delayed2, tracker2.asDelayed());
+ }
+
+ @Test
+ @Disabled
+ // a handy test to try deserializing data from spans
+ void testPersisted() {
+ String[] encodedData =
+ new String[] {
+ "AAAAI6SK7u6LAcG57+//L7gXB4IDGEYITQFPAVEBUwHgAQEAAACvCi3WjrSyLhomSh0EAQl0Lpo45iXmARamNA4qQB9qOIQw/iiKJQ5shA7WC/4dMh2mO2wXaBfGKVAXWDwQFF4eSismAQeqAjJMMl4kIg3wGSxaUBRERkjb8CSsJC48VCJqFuAMolksb/TEAVe8JXuksSoCrFqEErM8BmcapgmfBDbYWgAf/ix8C4IM7AkKFbyEhigRaODq3X69tgLxcRxQLgMaybJTKgLqZGg+ZJKJKJJJJJJJJJJKRJJJJJJJSRaaJJJJSRZZKAAAAA"
+ };
+ for (String encoded : encodedData) {
+ System.out.println("====");
+ byte[] decoded = Base64.getDecoder().decode(encoded);
+
+ for (IntervalParser.Interval i : new IntervalParser().parseIntervals(decoded)) {
+ System.out.println(
+ "===> "
+ + Instant.ofEpochMilli(i.from / 1000000)
+ + " - "
+ + Instant.ofEpochMilli(i.till / 1000000)
+ + " :: "
+ + (i.till - i.from));
+ }
+ }
+ }
+}
diff --git a/dd-java-agent/agent-profiling/profiling-controller-jfr/profiling-controller-jfr.gradle b/dd-java-agent/agent-profiling/profiling-controller-jfr/profiling-controller-jfr.gradle
index b8b7af23425..3fb6419644a 100644
--- a/dd-java-agent/agent-profiling/profiling-controller-jfr/profiling-controller-jfr.gradle
+++ b/dd-java-agent/agent-profiling/profiling-controller-jfr/profiling-controller-jfr.gradle
@@ -6,7 +6,6 @@ apply from: "$rootDir/gradle/java.gradle"
dependencies {
api project(':dd-java-agent:agent-profiling:profiling-controller')
- implementation 'tools.profiler:async-profiler:1.8.3'
implementation deps.slf4j
diff --git a/dd-java-agent/agent-profiling/profiling-controller-jfr/src/main/resources/jfr/dd.jfp b/dd-java-agent/agent-profiling/profiling-controller-jfr/src/main/resources/jfr/dd.jfp
index 52c4b0df085..be0fcdd893c 100644
--- a/dd-java-agent/agent-profiling/profiling-controller-jfr/src/main/resources/jfr/dd.jfp
+++ b/dd-java-agent/agent-profiling/profiling-controller-jfr/src/main/resources/jfr/dd.jfp
@@ -246,3 +246,5 @@ datadog.Scope#threshold=10 ms
datadog.ExceptionSample#enabled=true
datadog.ExceptionCount#enabled=true
datadog.ProfilerSetting#enabled=true
+datadog.ContextInterval#enabled=true
+datadog.Checkpoint#enabled=true
diff --git a/dd-java-agent/agent-profiling/src/main/java/com/datadog/profiling/agent/ProfilingAgent.java b/dd-java-agent/agent-profiling/src/main/java/com/datadog/profiling/agent/ProfilingAgent.java
index 279af730078..dc984cb6f79 100644
--- a/dd-java-agent/agent-profiling/src/main/java/com/datadog/profiling/agent/ProfilingAgent.java
+++ b/dd-java-agent/agent-profiling/src/main/java/com/datadog/profiling/agent/ProfilingAgent.java
@@ -2,6 +2,8 @@
import static datadog.trace.util.AgentThreadFactory.AGENT_THREAD_GROUP;
+import com.datadog.profiling.context.JfrTimestampPatch;
+import com.datadog.profiling.context.ProfilerTracingContextTrackerFactory;
import com.datadog.profiling.controller.ConfigurationException;
import com.datadog.profiling.controller.Controller;
import com.datadog.profiling.controller.ControllerFactory;
@@ -9,6 +11,7 @@
import com.datadog.profiling.controller.UnsupportedEnvironmentException;
import com.datadog.profiling.uploader.ProfileUploader;
import datadog.trace.api.Config;
+import datadog.trace.api.Platform;
import datadog.trace.bootstrap.config.provider.ConfigProvider;
import java.io.IOException;
import java.lang.ref.WeakReference;
@@ -32,7 +35,7 @@ public class ProfilingAgent {
* Main entry point into profiling Note: this must be reentrant because we may want to start
* profiling before any other tool, and then attempt to start it again at normal time
*/
- public static synchronized void run(final boolean isStartingFirst)
+ public static synchronized void run(final boolean isStartingFirst, ClassLoader agentClasLoader)
throws IllegalArgumentException, IOException {
if (profiler == null) {
final Config config = Config.get();
@@ -50,11 +53,16 @@ public static synchronized void run(final boolean isStartingFirst)
"Profiling: API key doesn't match expected format, expected to get a 32 character hex string. Profiling is disabled.");
return;
}
+ if (Platform.isJavaVersionAtLeast(9)) {
+ JfrTimestampPatch.execute(agentClasLoader);
+ }
try {
final ConfigProvider configProvider = ConfigProvider.getInstance();
final Controller controller = ControllerFactory.createController(configProvider);
+ ProfilerTracingContextTrackerFactory.register(configProvider);
+
final ProfileUploader uploader = new ProfileUploader(config, configProvider);
final Duration startupDelay = Duration.ofSeconds(config.getProfilingStartDelay());
diff --git a/dd-java-agent/dd-java-agent.gradle b/dd-java-agent/dd-java-agent.gradle
index f745a9ab673..9a32f1dd426 100644
--- a/dd-java-agent/dd-java-agent.gradle
+++ b/dd-java-agent/dd-java-agent.gradle
@@ -194,7 +194,7 @@ tasks.register('checkAgentJarSize').configure {
doLast {
// Arbitrary limit to prevent unintentional increases to the agent jar size
// Raise or lower as required
- assert shadowJar.archiveFile.get().getAsFile().length() < 19 * 1024 * 1024
+ assert shadowJar.archiveFile.get().getAsFile().length() < 20 * 1024 * 1024
}
dependsOn "shadowJar"
diff --git a/dd-trace-api/src/main/java/datadog/trace/api/config/ProfilingConfig.java b/dd-trace-api/src/main/java/datadog/trace/api/config/ProfilingConfig.java
index 41ac137b033..b65eb7dc9e0 100644
--- a/dd-trace-api/src/main/java/datadog/trace/api/config/ProfilingConfig.java
+++ b/dd-trace-api/src/main/java/datadog/trace/api/config/ProfilingConfig.java
@@ -80,6 +80,21 @@ public final class ProfilingConfig {
public static final String PROFILING_ASYNC_MEMLEAK_CAPACITY = "profiling.async.memleak.capacity";
public static final int PROFILING_ASYNC_MEMLEAK_CAPACITY_DEFAULT = 1024;
+ public static final String PROFILING_TRACING_CONTEXT_ENABLED =
+ "profiling.tracing_context.enabled";
+ public static final boolean PROFILING_TRACING_CONTEXT_ENABLED_DEFAULT = false;
+ public static final String PROFILING_TRACING_CONTEXT_TRACKER_INACTIVE_SEC =
+ "profiling.tracing_context.tracker.inactive.seconds";
+ public static final int PROFILING_TRACING_CONTEXT_TRACKER_INACTIVE_DEFAULT = 90;
+ public static final String PROFILING_TRACING_CONTEXT_RESERVED_MEMORY_SIZE =
+ "profiling.tracing_context.memory.bytes";
+ public static final int PROFILING_TRACING_CONTEXT_RESERVED_MEMORY_SIZE_DEFAULT =
+ 32 * 1024 * 1024; // 32MB
+
+ public static final String PROFILING_TRACING_CONTEXT_RESERVED_MEMORY_TYPE =
+ "profiling.tracing_context.memory.type";
+ public static final String PROFILING_TRACING_CONTEXT_RESERVED_MEMORY_TYPE_DEFAULT = "heap";
+
public static final String PROFILING_LEGACY_TRACING_INTEGRATION =
"profiling.legacy.tracing.integration";
public static final boolean PROFILING_LEGACY_TRACING_INTEGRATION_DEFAULT = true;
diff --git a/dd-trace-core/dd-trace-core.gradle b/dd-trace-core/dd-trace-core.gradle
index 0a38802cb21..c8c5d62aaba 100644
--- a/dd-trace-core/dd-trace-core.gradle
+++ b/dd-trace-core/dd-trace-core.gradle
@@ -20,7 +20,8 @@ excludedClassesCoverage += [
'datadog.trace.core.PendingTraceBuffer.DelayingPendingTraceBuffer.FlushElement',
'datadog.trace.core.StatusLogger',
'datadog.trace.core.scopemanager.ContinuableScopeManager.SingleContinuation',
- 'datadog.trace.core.SpanCorrelationImpl'
+ 'datadog.trace.core.SpanCorrelationImpl',
+ 'datadog.trace.core.Base64Encoder'
]
apply plugin: 'org.unbroken-dome.test-sets'
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/Base64Encoder.java b/dd-trace-core/src/main/java/datadog/trace/core/Base64Encoder.java
new file mode 100644
index 00000000000..1beb24a40d7
--- /dev/null
+++ b/dd-trace-core/src/main/java/datadog/trace/core/Base64Encoder.java
@@ -0,0 +1,99 @@
+package datadog.trace.core;
+
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+
+// TODO - can be removed when JDK7 is dropped (adapted from JDK source)
+final class Base64Encoder {
+
+ private static final char[] BASE_64 = new char[65];
+
+ static {
+ int i = 0;
+ for (char c = 'A'; c <= 'Z'; ++c) {
+ BASE_64[i++] = c;
+ }
+ for (char c = 'a'; c <= 'z'; ++c) {
+ BASE_64[i++] = c;
+ }
+ for (char c = '0'; c <= '9'; ++c) {
+ BASE_64[i++] = c;
+ }
+ BASE_64[i++] = '+';
+ BASE_64[i] = '/';
+ }
+
+ public static final Base64Encoder INSTANCE = new Base64Encoder();
+
+ private Base64Encoder() {}
+
+ private int outLength(int srclen) {
+ int n = srclen % 3;
+ return 4 * (srclen / 3) + (n == 0 ? 0 : n + 1);
+ }
+
+ public byte[] encode(byte[] src) {
+ int len = outLength(src.length); // dst array size
+ byte[] dst = new byte[len];
+ int ret = encode0(src, 0, src.length, dst);
+ if (ret != dst.length) return Arrays.copyOf(dst, ret);
+ return dst;
+ }
+
+ public ByteBuffer encode(ByteBuffer buffer) {
+ int len = outLength(buffer.remaining());
+ byte[] dst = new byte[len];
+ int ret = 0;
+ if (buffer.hasArray()) {
+ ret =
+ encode0(
+ buffer.array(),
+ buffer.arrayOffset() + buffer.position(),
+ buffer.arrayOffset() + buffer.limit(),
+ dst);
+ buffer.position(buffer.limit());
+ } else {
+ byte[] src = new byte[buffer.remaining()];
+ buffer.get(src);
+ ret = encode0(src, 0, src.length, dst);
+ }
+ if (ret != dst.length) dst = Arrays.copyOf(dst, ret);
+ return ByteBuffer.wrap(dst);
+ }
+
+ private void encodeBlock(byte[] src, int sp, int sl, byte[] dst, int dp) {
+ for (int sp0 = sp, dp0 = dp; sp0 < sl; ) {
+ int bits = (src[sp0++] & 0xff) << 16 | (src[sp0++] & 0xff) << 8 | (src[sp0++] & 0xff);
+ dst[dp0++] = (byte) BASE_64[(bits >>> 18) & 0x3f];
+ dst[dp0++] = (byte) BASE_64[(bits >>> 12) & 0x3f];
+ dst[dp0++] = (byte) BASE_64[(bits >>> 6) & 0x3f];
+ dst[dp0++] = (byte) BASE_64[bits & 0x3f];
+ }
+ }
+
+ private int encode0(byte[] src, int off, int end, byte[] dst) {
+ int sp = off;
+ int slen = (end - off) / 3 * 3;
+ int sl = off + slen;
+ int dp = 0;
+ while (sp < sl) {
+ int sl0 = Math.min(sp + slen, sl);
+ encodeBlock(src, sp, sl0, dst, dp);
+ int dlen = (sl0 - sp) / 3 * 4;
+ dp += dlen;
+ sp = sl0;
+ }
+ if (sp < end) { // 1 or 2 leftover bytes
+ int b0 = src[sp++] & 0xff;
+ dst[dp++] = (byte) BASE_64[b0 >> 2];
+ if (sp == end) {
+ dst[dp++] = (byte) BASE_64[(b0 << 4) & 0x3f];
+ } else {
+ int b1 = src[sp++] & 0xff;
+ dst[dp++] = (byte) BASE_64[(b0 << 4) & 0x3f | (b1 >> 4)];
+ dst[dp++] = (byte) BASE_64[(b1 << 2) & 0x3f];
+ }
+ }
+ return dp;
+ }
+}
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java b/dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java
index e71d2eaa0bb..cb045458495 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/CoreTracer.java
@@ -26,6 +26,7 @@
import datadog.trace.api.gateway.RequestContext;
import datadog.trace.api.interceptor.MutableSpan;
import datadog.trace.api.interceptor.TraceInterceptor;
+import datadog.trace.api.profiling.TracingContextTrackerFactory;
import datadog.trace.api.sampling.PrioritySampling;
import datadog.trace.api.sampling.SamplingMechanism;
import datadog.trace.api.time.SystemTimeSource;
@@ -780,6 +781,18 @@ void write(final List trace) {
spanToSample.forceKeep(forceKeep);
boolean published = forceKeep || sampler.sample(spanToSample);
if (published) {
+ if (TracingContextTrackerFactory.isTrackingAvailable()) {
+ for (DDSpan span : writtenTrace) {
+ int stored = span.storeContextToTag();
+ if (stored > -1) {
+ log.trace(
+ "Sending statsd metric 'tracing.context.size'={} (client={})",
+ stored,
+ statsDClient);
+ statsDClient.histogram("tracing.context.size", stored);
+ }
+ }
+ }
writer.write(writtenTrace);
} else {
// with span streaming this won't work - it needs to be changed
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java b/dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java
index 0fd5732435d..7f4de132a6f 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/DDSpan.java
@@ -13,13 +13,18 @@
import datadog.trace.api.DDId;
import datadog.trace.api.DDTags;
import datadog.trace.api.gateway.RequestContext;
+import datadog.trace.api.profiling.TracingContextTracker;
+import datadog.trace.api.profiling.TracingContextTrackerFactory;
+import datadog.trace.api.profiling.TransientProfilingContextHolder;
import datadog.trace.api.sampling.PrioritySampling;
import datadog.trace.api.sampling.SamplingMechanism;
import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
import datadog.trace.bootstrap.instrumentation.api.AttachableWrapper;
import datadog.trace.bootstrap.instrumentation.api.ResourceNamePriorities;
+import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
import java.io.PrintWriter;
import java.io.StringWriter;
+import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
@@ -35,7 +40,8 @@
* Spans are created by the {@link CoreTracer#buildSpan}. This implementation adds some features
* according to the DD agent.
*/
-public class DDSpan implements AgentSpan, CoreSpan, AttachableWrapper {
+public class DDSpan
+ implements AgentSpan, CoreSpan, TransientProfilingContextHolder, AttachableWrapper {
private static final Logger log = LoggerFactory.getLogger(DDSpan.class);
public static final String CHECKPOINTED_TAG = "checkpointed";
@@ -49,6 +55,7 @@ static DDSpan create(
final DDSpan span = new DDSpan(timestampMicro, context, emitCheckpoints);
log.debug("Started span: {}", span);
context.getTrace().registerSpan(span);
+ span.tracingContextTracker = TracingContextTrackerFactory.instance(span);
return span;
}
@@ -101,6 +108,8 @@ private DDSpan(final long timestampMicro, @Nonnull DDSpanContext context) {
this(timestampMicro, context, true);
}
+ private volatile TracingContextTracker tracingContextTracker;
+
private DDSpan(
final long timestampMicro, @Nonnull DDSpanContext context, boolean emitLocalCheckpoints) {
this.context = context;
@@ -125,6 +134,7 @@ private void finishAndAddToTrace(final long durationNano) {
// ensure a min duration of 1
if (DURATION_NANO_UPDATER.compareAndSet(this, 0, Math.max(1, durationNano))) {
context.getTrace().onFinish(this);
+ tracingContextTracker.deactivateContext();
PendingTrace.PublishState publishState = context.getTrace().onPublish(this);
log.debug("Finished span ({}): {}", publishState, this);
} else {
@@ -237,6 +247,30 @@ public final void publish() {
}
}
+ public int storeContextToTag() {
+ try {
+ byte[] contextContent = tracingContextTracker.persist();
+ if (contextContent != null) {
+ String contextTagName = "_dd_tracing_context_" + tracingContextTracker.getVersion();
+ byte[] encoded = Base64Encoder.INSTANCE.encode(contextContent);
+ if (log.isTraceEnabled()) {
+ log.trace(
+ "Tracing context data for s_id:{}, tag:{}={}",
+ getSpanId(),
+ contextTagName,
+ new String(encoded, StandardCharsets.UTF_8));
+ }
+ context.setTag(contextTagName, UTF8BytesString.create(encoded));
+ return encoded.length;
+ }
+ } catch (Throwable t) {
+ log.error("", t);
+ } finally {
+ tracingContextTracker.release();
+ }
+ return -1;
+ }
+
@Override
public DDSpan setError(final boolean error) {
context.setErrorFlag(error);
@@ -521,6 +555,7 @@ public boolean eligibleForDropping() {
@Override
public void startThreadMigration() {
+ tracingContextTracker.maybeDeactivateContext();
if (hasCheckpoints()) {
context.getTracer().onStartThreadMigration(this);
}
@@ -528,6 +563,7 @@ public void startThreadMigration() {
@Override
public void finishThreadMigration() {
+ tracingContextTracker.activateContext();
if (hasCheckpoints()) {
context.getTracer().onFinishThreadMigration(this);
}
@@ -535,6 +571,7 @@ public void finishThreadMigration() {
@Override
public void finishWork() {
+ tracingContextTracker.deactivateContext();
if (hasCheckpoints()) {
context.getTracer().onFinishWork(this);
}
diff --git a/gradle/dependencies.gradle b/gradle/dependencies.gradle
index b3bfe071c44..efad4f3d4eb 100644
--- a/gradle/dependencies.gradle
+++ b/gradle/dependencies.gradle
@@ -29,7 +29,7 @@ final class CachedData {
testcontainers: '1.15.0-rc2',
jmc : "8.1.0-SNAPSHOT",
autoservice : "1.0-rc7",
- asyncprofiler : "2.7-DD-20220303"
+ asyncprofiler : "2.7-DD-20220411"
]
static deps = [
diff --git a/internal-api/src/main/java/datadog/trace/api/profiling/TracingContextTracker.java b/internal-api/src/main/java/datadog/trace/api/profiling/TracingContextTracker.java
new file mode 100644
index 00000000000..cbc39b3e4d9
--- /dev/null
+++ b/internal-api/src/main/java/datadog/trace/api/profiling/TracingContextTracker.java
@@ -0,0 +1,104 @@
+package datadog.trace.api.profiling;
+
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+
+/** A tracing context tracker */
+public interface TracingContextTracker {
+ /** Provides support for a timed tracker cleanup */
+ interface DelayedTracker extends Delayed {
+ DelayedTracker EMPTY =
+ new DelayedTracker() {
+ @Override
+ public void cleanup() {}
+
+ @Override
+ public long getDelay(TimeUnit unit) {
+ return -1;
+ }
+
+ @Override
+ public int compareTo(Delayed o) {
+ return -1;
+ }
+ };
+
+ void cleanup();
+ }
+
+ /** A no-op implementation */
+ TracingContextTracker EMPTY =
+ new TracingContextTracker() {
+ @Override
+ public boolean release() {
+ return false;
+ }
+
+ @Override
+ public void activateContext() {}
+
+ @Override
+ public void deactivateContext() {}
+
+ @Override
+ public void maybeDeactivateContext() {}
+
+ @Override
+ public byte[] persist() {
+ return null;
+ }
+
+ @Override
+ public int getVersion() {
+ return 0;
+ }
+
+ @Override
+ public DelayedTracker asDelayed() {
+ return DelayedTracker.EMPTY;
+ }
+ };
+
+ /**
+ * Release any resources held by the tracker
+ *
+ * @return {@literal false} if already released; {@literal true} otherwise
+ */
+ boolean release();
+
+ /** Notify of the context activation */
+ void activateContext();
+
+ /** Notify of the context deactivation */
+ void deactivateContext();
+
+ /**
+ * Notify of the eventual context deactivation
+ *
+ * @param the deactivation is conditional - if it is followed by an activation the deactivation
+ * should be disregarded
+ */
+ void maybeDeactivateContext();
+
+ /**
+ * Convert the sparse 'on-line' representation into a compressed binary blob
+ *
+ * @return the binary blob of the context tracking data or {@literal null}
+ */
+ byte[] persist();
+
+ /**
+ * The tracker version - should be in sync with the binary blob format
+ *
+ * @return the tracker version
+ */
+ int getVersion();
+
+ /**
+ * Turn this instance into a {@linkplain Delayed} one, suitable to be used in {@linkplain
+ * java.util.concurrent.DelayQueue}
+ *
+ * @return the delayed version of this instance
+ */
+ DelayedTracker asDelayed();
+}
diff --git a/internal-api/src/main/java/datadog/trace/api/profiling/TracingContextTrackerFactory.java b/internal-api/src/main/java/datadog/trace/api/profiling/TracingContextTrackerFactory.java
new file mode 100644
index 00000000000..daa2be2ceb7
--- /dev/null
+++ b/internal-api/src/main/java/datadog/trace/api/profiling/TracingContextTrackerFactory.java
@@ -0,0 +1,64 @@
+package datadog.trace.api.profiling;
+
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A factory class for {@linkplain TracingContextTracker} instances.
+ * This class is actually just a shared scaffolding and the actual implementation needs to register
+ * itself by calling {@linkplain
+ * TracingContextTrackerFactory#registerImplementation(Implementation)}
+ */
+public final class TracingContextTrackerFactory {
+ private static final Logger log = LoggerFactory.getLogger(TracingContextTrackerFactory.class);
+
+ public interface Implementation {
+ Implementation EMPTY =
+ new Implementation() {
+ public TracingContextTracker instance(AgentSpan span) {
+ return TracingContextTracker.EMPTY;
+ }
+ };
+
+ TracingContextTracker instance(AgentSpan span);
+ }
+
+ private static final TracingContextTrackerFactory INSTANCE = new TracingContextTrackerFactory();
+
+ private volatile Implementation implementation = Implementation.EMPTY;
+ private final AtomicReferenceFieldUpdater
+ implFieldUpdater =
+ AtomicReferenceFieldUpdater.newUpdater(
+ TracingContextTrackerFactory.class, Implementation.class, "implementation");
+
+ public static boolean registerImplementation(Implementation factoryImplementation) {
+ try {
+ return INSTANCE.implFieldUpdater.compareAndSet(
+ INSTANCE, Implementation.EMPTY, factoryImplementation);
+ } catch (Throwable t) {
+ log.warn("Failed to register a profiling context implementation", t);
+ }
+ return false;
+ }
+
+ static void removeImplementation(Implementation implementation) {
+ INSTANCE.implFieldUpdater.compareAndSet(INSTANCE, implementation, Implementation.EMPTY);
+ }
+
+ public static boolean isTrackingAvailable() {
+ // return true if a non-dummy implementatin has been registered
+ return INSTANCE.implementation != Implementation.EMPTY;
+ }
+
+ /**
+ * Create a new tracing context tracker associated with the given span
+ *
+ * @param span the span to associate the tracker with
+ * @return a new {@linkplain TracingContextTracker} instance
+ */
+ public static TracingContextTracker instance(AgentSpan span) {
+ return INSTANCE.implementation.instance(span);
+ }
+}
diff --git a/internal-api/src/main/java/datadog/trace/api/profiling/TransientProfilingContextHolder.java b/internal-api/src/main/java/datadog/trace/api/profiling/TransientProfilingContextHolder.java
new file mode 100644
index 00000000000..78e5da98033
--- /dev/null
+++ b/internal-api/src/main/java/datadog/trace/api/profiling/TransientProfilingContextHolder.java
@@ -0,0 +1,5 @@
+package datadog.trace.api.profiling;
+
+public interface TransientProfilingContextHolder {
+ int storeContextToTag();
+}
diff --git a/internal-api/src/test/groovy/datadog/trace/api/profiling/TracingContextTrackerFactoryTest.groovy b/internal-api/src/test/groovy/datadog/trace/api/profiling/TracingContextTrackerFactoryTest.groovy
new file mode 100644
index 00000000000..8c3c2c9106b
--- /dev/null
+++ b/internal-api/src/test/groovy/datadog/trace/api/profiling/TracingContextTrackerFactoryTest.groovy
@@ -0,0 +1,63 @@
+package datadog.trace.api.profiling
+
+import datadog.trace.test.util.DDSpecification
+
+class TracingContextTrackerFactoryTest extends DDSpecification {
+ def factory
+
+ def cleanup() {
+ TracingContextTrackerFactory.removeImplementation(factory)
+ }
+
+ def "sanity default"() {
+ expect:
+ TracingContextTrackerFactory.isTrackingAvailable() == false
+ TracingContextTrackerFactory.instance(null) == TracingContextTracker.EMPTY
+ }
+
+ def "sanity empty tracker"() {
+ when:
+ def tracker = TracingContextTrackerFactory.instance(null)
+
+ then:
+ tracker.version == 0
+ tracker.activateContext()
+ tracker.deactivateContext()
+ tracker.maybeDeactivateContext()
+ tracker.persist() == null
+ tracker.release() == false
+
+ when:
+ def delayed = tracker.asDelayed()
+
+ then:
+ delayed == TracingContextTracker.DelayedTracker.EMPTY
+ delayed.getDelay(null) == -1
+ delayed.compareTo(null) == -1
+ delayed.cleanup()
+ }
+
+ def "custom implementation registration"() {
+ setup:
+ factory = new TestTracingContextTrackerFactory()
+
+ expect:
+ TracingContextTrackerFactory.registerImplementation(factory) == true
+ TracingContextTrackerFactory.registerImplementation(factory) == false
+ TracingContextTrackerFactory.isTrackingAvailable() == true
+ }
+
+ def "custom factory usage"() {
+ setup:
+ factory = new TestTracingContextTrackerFactory()
+ TracingContextTrackerFactory.registerImplementation(factory)
+
+ when:
+ def tracker = TracingContextTrackerFactory.instance(null)
+
+ then:
+ tracker != null
+ tracker != TracingContextTracker.EMPTY
+ tracker instanceof TestTracingContextTracker
+ }
+}
diff --git a/internal-api/src/test/java/datadog/trace/api/profiling/TestTracingContextTracker.java b/internal-api/src/test/java/datadog/trace/api/profiling/TestTracingContextTracker.java
new file mode 100644
index 00000000000..23864750a20
--- /dev/null
+++ b/internal-api/src/test/java/datadog/trace/api/profiling/TestTracingContextTracker.java
@@ -0,0 +1,32 @@
+package datadog.trace.api.profiling;
+
+final class TestTracingContextTracker implements TracingContextTracker {
+ @Override
+ public boolean release() {
+ return false;
+ }
+
+ @Override
+ public void activateContext() {}
+
+ @Override
+ public void deactivateContext() {}
+
+ @Override
+ public void maybeDeactivateContext() {}
+
+ @Override
+ public byte[] persist() {
+ return new byte[0];
+ }
+
+ @Override
+ public int getVersion() {
+ return 0;
+ }
+
+ @Override
+ public DelayedTracker asDelayed() {
+ return null;
+ }
+}
diff --git a/internal-api/src/test/java/datadog/trace/api/profiling/TestTracingContextTrackerFactory.java b/internal-api/src/test/java/datadog/trace/api/profiling/TestTracingContextTrackerFactory.java
new file mode 100644
index 00000000000..146c0fcc896
--- /dev/null
+++ b/internal-api/src/test/java/datadog/trace/api/profiling/TestTracingContextTrackerFactory.java
@@ -0,0 +1,11 @@
+package datadog.trace.api.profiling;
+
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
+
+final class TestTracingContextTrackerFactory
+ implements TracingContextTrackerFactory.Implementation {
+ @Override
+ public TracingContextTracker instance(AgentSpan span) {
+ return new TestTracingContextTracker();
+ }
+}
diff --git a/settings.gradle b/settings.gradle
index cca6b3d8063..5425c74e5b8 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -45,9 +45,11 @@ include ':dd-java-agent:agent-jmxfetch'
include ':dd-java-agent:agent-logging'
include ':dd-java-agent:load-generator'
+// profiling
include ':dd-java-agent:agent-profiling'
include ':dd-java-agent:agent-profiling:profiling-auxiliary'
include ':dd-java-agent:agent-profiling:profiling-auxiliary-async'
+include ':dd-java-agent:agent-profiling:profiling-context'
include ':dd-java-agent:agent-profiling:profiling-controller'
include ':dd-java-agent:agent-profiling:profiling-controller-jfr'
include ':dd-java-agent:agent-profiling:profiling-controller-openjdk'