diff --git a/communication/src/main/java/datadog/communication/serialization/ChunkedStreamingBuffer.java b/communication/src/main/java/datadog/communication/serialization/ChunkedStreamingBuffer.java index 6a82a77b49f..f95a7895936 100644 --- a/communication/src/main/java/datadog/communication/serialization/ChunkedStreamingBuffer.java +++ b/communication/src/main/java/datadog/communication/serialization/ChunkedStreamingBuffer.java @@ -24,6 +24,14 @@ protected void onCapacityExceeded() { throw new BufferOverflowException(); } + /** + * A buffer whose chunks start at {@code firstChunkSize} and double, for usually-small payloads. + */ + public ChunkedStreamingBuffer(int maxBytes, int firstChunkSize, ChunkedBufferConsumer consumer) { + super(maxBytes, firstChunkSize); + this.consumer = consumer; + } + /** * Hands the completed messages to the consumer, then resets, which also releases the chunks the * payload didn't need. diff --git a/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/InstrumentationSpecification.groovy b/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/InstrumentationSpecification.groovy index 38904cdd717..9e5b9246d23 100644 --- a/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/InstrumentationSpecification.groovy +++ b/dd-java-agent/instrumentation-testing/src/main/groovy/datadog/trace/agent/test/InstrumentationSpecification.groovy @@ -375,7 +375,7 @@ abstract class InstrumentationSpecification extends DDSpecification implements A DDAgentFeaturesDiscovery features = new MockFeaturesDiscovery(true) Sink sink = new Sink() { - void accept(int messageCount, ByteBuffer buffer) {} + void accept(int messageCount, List chunks) {} void register(EventListener listener) {} } diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java index 2ab9e0b7763..4091a17dd2f 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-0.11/src/latestDepTest/java/KafkaProducerFailureDataStreamsForkedTest.java @@ -176,7 +176,7 @@ private static String repeat(char c, int count) { private static final class NoopSink implements Sink { @Override - public void accept(int messageCount, ByteBuffer buffer) {} + public void accept(int messageCount, List chunks) {} @Override public void register(EventListener listener) {} diff --git a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java index b56ed2328fa..867b65cf7a8 100644 --- a/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java +++ b/dd-java-agent/instrumentation/kafka/kafka-clients-3.8/src/test/java/KafkaProducerFailureDataStreamsForkedTest.java @@ -177,7 +177,7 @@ private static String repeat(char c, int count) { private static final class NoopSink implements Sink { @Override - public void accept(int messageCount, ByteBuffer buffer) {} + public void accept(int messageCount, List chunks) {} @Override public void register(EventListener listener) {} diff --git a/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorBenchmark.java b/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorBenchmark.java index 08449d7a9ac..5c9420283e9 100644 --- a/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorBenchmark.java +++ b/dd-trace-core/src/jmh/java/datadog/trace/common/metrics/ClientStatsAggregatorBenchmark.java @@ -68,7 +68,7 @@ static class NullSink implements Sink { public void register(EventListener listener) {} @Override - public void accept(int messageCount, ByteBuffer buffer) {} + public void accept(int messageCount, List chunks) {} } static class FixedAgentFeaturesDiscovery extends DDAgentFeaturesDiscovery { diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java index 75236791324..eaca6935f32 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java @@ -1,6 +1,7 @@ package datadog.trace.common.metrics; import java.nio.ByteBuffer; +import java.util.List; /** A {@link Sink} that discards everything. */ public final class NoOpSink implements Sink { @@ -10,7 +11,7 @@ public final class NoOpSink implements Sink { private NoOpSink() {} @Override - public void accept(int messageCount, ByteBuffer buffer) {} + public void accept(int messageCount, List chunks) {} @Override public void register(EventListener listener) {} diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OkHttpSink.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OkHttpSink.java index 193a2fb0436..d1c93eefc0c 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OkHttpSink.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OkHttpSink.java @@ -67,13 +67,13 @@ public OkHttpSink( } @Override - public void accept(int messageCount, ByteBuffer buffer) { + public void accept(int messageCount, List chunks) { // if the agent is healthy, then we can send on this thread, - // without copying the buffer, otherwise this needs to be async, + // without copying the chunks, otherwise this needs to be async, // so need to copy and buffer the request, and let it be executed // on the main task scheduler as a last resort if (!bufferingEnabled || lastRequestTime.get() < ASYNC_THRESHOLD_LATENCY) { - send(prepareRequest(metricsUrl, headers).post(makeRequestBody(buffer)).build()); + send(prepareRequest(metricsUrl, headers).post(makeRequestBody(chunks)).build()); AgentTaskScheduler.Scheduled future = this.future; if (future != null && enqueuedRequests.isEmpty()) { // async mode has been started but request latency is normal, @@ -87,29 +87,46 @@ public void accept(int messageCount, ByteBuffer buffer) { AgentTaskScheduler.get() .scheduleAtFixedRate(new Sender(enqueuedRequests), this, 1, 1, SECONDS); } - sendAsync(messageCount, buffer); + sendAsync(messageCount, chunks); } } - private RequestBody makeRequestBody(ByteBuffer buffer) { + private RequestBody makeRequestBody(List chunks) { if (compressionEnabled) { - return gzippedMsgpackRequestBodyOf(Collections.singletonList(buffer)); + return gzippedMsgpackRequestBodyOf(chunks); } else { - return msgpackRequestBodyOf(Collections.singletonList(buffer)); + return msgpackRequestBodyOf(chunks); } } - private void sendAsync(int messageCount, ByteBuffer buffer) { + private void sendAsync(int messageCount, List chunks) { asyncRequestCounter.getAndIncrement(); + // the chunks are reused for the next payload as soon as accept returns, so copy them + ByteBuffer copy = copyOf(chunks); if (!enqueuedRequests.offer( - prepareRequest(metricsUrl, headers).post(makeRequestBody(buffer.duplicate())).build())) { + prepareRequest(metricsUrl, headers) + .post(makeRequestBody(Collections.singletonList(copy))) + .build())) { log.debug( "dropping payload of {} and {}B because sending queue was full", messageCount, - buffer.limit()); + copy.limit()); } } + private static ByteBuffer copyOf(List chunks) { + int size = 0; + for (ByteBuffer chunk : chunks) { + size += chunk.remaining(); + } + ByteBuffer copy = ByteBuffer.allocate(size); + for (ByteBuffer chunk : chunks) { + copy.put(chunk.duplicate()); + } + copy.flip(); + return copy; + } + public boolean isInDegradedMode() { return asyncTaskStarted.get(); } diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java index 3eacde8bd45..db303233b30 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/SerializingMetricWriter.java @@ -3,7 +3,7 @@ import static datadog.trace.bootstrap.instrumentation.api.UTF8BytesString.EMPTY; import static java.nio.charset.StandardCharsets.ISO_8859_1; -import datadog.communication.serialization.GrowableBuffer; +import datadog.communication.serialization.ChunkedStreamingBuffer; import datadog.communication.serialization.WritableFormatter; import datadog.communication.serialization.msgpack.MsgPackWriter; import datadog.metrics.api.Histogram; @@ -14,12 +14,20 @@ import datadog.trace.api.git.GitInfo; import datadog.trace.api.git.GitInfoProvider; import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString; +import datadog.trace.util.ChunkedBuffer; +import java.nio.BufferOverflowException; import java.nio.ByteBuffer; import java.util.List; import java.util.function.Function; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public final class SerializingMetricWriter implements MetricWriter { + private static final Logger log = LoggerFactory.getLogger(SerializingMetricWriter.class); + + private static final int MAX_PAYLOAD_SIZE = 5 << 20; + private static final byte[] SEQUENCE = "Sequence".getBytes(ISO_8859_1); private static final byte[] RUNTIME_ID = "RuntimeID".getBytes(ISO_8859_1); private static final byte[] HOSTNAME = "Hostname".getBytes(ISO_8859_1); @@ -73,7 +81,7 @@ public final class SerializingMetricWriter implements MetricWriter { */ private final boolean additionalTagsConfigured; - private final GrowableBuffer buffer; + private final ChunkedStreamingBuffer buffer; private final DDCache gitInfoCache = DDCaches.newFixedSizeWeakKeyCache(4); private long sequence = 0; @@ -91,37 +99,69 @@ public SerializingMetricWriter(WellKnownTags wellKnownTags, Sink sink) { public SerializingMetricWriter( WellKnownTags wellKnownTags, Sink sink, boolean additionalTagsConfigured) { - this(wellKnownTags, sink, 512 * 1024, GitInfoProvider.INSTANCE, additionalTagsConfigured); + this( + wellKnownTags, + sink, + ChunkedBuffer.DEFAULT_FIRST_CHUNK_SIZE, + GitInfoProvider.INSTANCE, + additionalTagsConfigured); } - public SerializingMetricWriter(WellKnownTags wellKnownTags, Sink sink, int initialCapacity) { - this(wellKnownTags, sink, initialCapacity, GitInfoProvider.INSTANCE); + public SerializingMetricWriter(WellKnownTags wellKnownTags, Sink sink, int firstChunkSize) { + this(wellKnownTags, sink, firstChunkSize, GitInfoProvider.INSTANCE); } public SerializingMetricWriter( WellKnownTags wellKnownTags, Sink sink, - int initialCapacity, + int firstChunkSize, final GitInfoProvider gitInfoProvider) { - this(wellKnownTags, sink, initialCapacity, gitInfoProvider, false); + this(wellKnownTags, sink, firstChunkSize, gitInfoProvider, false); } public SerializingMetricWriter( WellKnownTags wellKnownTags, Sink sink, - int initialCapacity, + int firstChunkSize, + final GitInfoProvider gitInfoProvider, + boolean additionalTagsConfigured) { + this( + wellKnownTags, + sink, + firstChunkSize, + MAX_PAYLOAD_SIZE, + gitInfoProvider, + additionalTagsConfigured); + } + + SerializingMetricWriter( + WellKnownTags wellKnownTags, + Sink sink, + int firstChunkSize, + int maxPayloadSize, final GitInfoProvider gitInfoProvider, boolean additionalTagsConfigured) { this.wellKnownTags = wellKnownTags; - this.buffer = new GrowableBuffer(initialCapacity); + this.buffer = new ChunkedStreamingBuffer(maxPayloadSize, firstChunkSize, sink); this.writer = new MsgPackWriter(buffer); this.sink = sink; this.gitInfoProvider = gitInfoProvider; this.additionalTagsConfigured = additionalTagsConfigured; } + // A payload spans startBucket, add and finishBucket. Only the first write past the cap throws; + // the buffer discards the rest, and mark() in finishBucket reports the payload as dropped. + @Override public void startBucket(int metricCount, long start, long duration) { + try { + writeBucketStart(metricCount, start, duration); + } catch (BufferOverflowException ignored) { + // reported by finishBucket + } + } + + private void writeBucketStart(int metricCount, long start, long duration) { final UTF8BytesString processTags = ProcessTags.getTagsForSerialization(); final boolean writeProcessTags = processTags != null; // the gitinfo can be rebuild in time and initialized after this class is created. So that a @@ -177,6 +217,14 @@ public void startBucket(int metricCount, long start, long duration) { @Override public void add(AggregateEntry entry) { + try { + writeEntry(entry); + } catch (BufferOverflowException ignored) { + // reported by finishBucket + } + } + + private void writeEntry(AggregateEntry entry) { // Dynamic map size based on optional fields; AggregateEntry encapsulates the EMPTY-as-absent // sentinel via its hasFoo() predicates so the serializer doesn't depend on the storage choice. final boolean hasHttpMethod = entry.hasHttpMethod(); @@ -306,9 +354,12 @@ private byte[] emptyErrorHistogramBytes() { @Override public void finishBucket() { - buffer.mark(); - sink.accept(buffer.messageCount(), buffer.slice()); - buffer.reset(); + if (buffer.mark()) { + buffer.flush(); + } else { + log.debug("dropping a stats payload larger than {} bytes", buffer.capacity()); + buffer.reset(); + } } @Override diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/Sink.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/Sink.java index 7a812f341f4..f8261339926 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/Sink.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/Sink.java @@ -1,8 +1,15 @@ package datadog.trace.common.metrics; -import datadog.communication.serialization.ByteBufferConsumer; +import datadog.communication.serialization.ChunkedBufferConsumer; +import java.nio.ByteBuffer; +import java.util.Collections; -public interface Sink extends ByteBufferConsumer { +public interface Sink extends ChunkedBufferConsumer { + + /** Accepts a payload held in a single buffer. */ + default void accept(int messageCount, ByteBuffer buffer) { + accept(messageCount, Collections.singletonList(buffer)); + } void register(EventListener listener); } diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/FootprintForkedTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/FootprintForkedTest.java index ac54d7d700f..78dbed26f3b 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/FootprintForkedTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/FootprintForkedTest.java @@ -16,6 +16,7 @@ import java.util.Collection; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Random; import java.util.UUID; @@ -177,7 +178,7 @@ private static class ValidatingSink implements Sink { public void register(EventListener listener) {} @Override - public void accept(int messageCount, ByteBuffer buffer) { + public void accept(int messageCount, List chunks) { latch.countDown(); } } diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OkHttpSinkTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OkHttpSinkTest.java index c864baeaf8c..044ac8d78cb 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OkHttpSinkTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OkHttpSinkTest.java @@ -3,6 +3,7 @@ import static datadog.communication.ddagent.DDAgentFeaturesDiscovery.V06_METRICS_ENDPOINT; import static datadog.trace.common.metrics.EventListener.EventType.OK; import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -17,6 +18,7 @@ import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; @@ -55,7 +57,7 @@ void httpStatusCodeResponseCodeYieldsEventType( doAnswer(invocation -> respond(invocation.getArgument(0), responseCode)) .when(client) .newCall(any()); - sink.accept(0, ByteBuffer.allocate(0)); + sink.accept(0, Collections.singletonList(ByteBuffer.allocate(0))); verify(client, times(1)).newCall(any()); verify(listener).onEvent(eq(eventType), any()); @@ -97,8 +99,8 @@ void degradeToAsyncModeWhenAgentSlowToRespond() throws Exception { .newCall(any()); // one slow response followed by a request - sink.accept(1, ByteBuffer.allocate(0)); - sink.accept(1, ByteBuffer.allocate(0)); + sink.accept(1, Collections.singletonList(ByteBuffer.allocate(0))); + sink.accept(1, Collections.singletonList(ByteBuffer.allocate(0))); latch.await(10, SECONDS); // the second request degrades to async mode @@ -112,7 +114,7 @@ void degradeToAsyncModeWhenAgentSlowToRespond() throws Exception { assertTrue(sink.isInDegradedMode()); // the agent has recovered and has responded quickly once - sink.accept(1, ByteBuffer.allocate(0)); + sink.accept(1, Collections.singletonList(ByteBuffer.allocate(0))); // the request was sent synchronously verify(client, times(3)).newCall(any()); @@ -120,6 +122,48 @@ void degradeToAsyncModeWhenAgentSlowToRespond() throws Exception { assertFalse(sink.isInDegradedMode()); } + @Test + void asyncModeSendsACopyOfThePayload() throws Exception { + // the writer reuses its chunks for the next payload as soon as accept returns, so a request + // queued in async mode must not share them + String agentUrl = "http://localhost:8126"; + CountDownLatch latch = new CountDownLatch(2); + BlockingListener listener = new BlockingListener(latch); + OkHttpClient client = mock(OkHttpClient.class); + OkHttpSink sink = + new OkHttpSink(client, agentUrl, V06_METRICS_ENDPOINT, true, false, Collections.emptyMap()); + sink.register(listener); + AtomicInteger callCount = new AtomicInteger(0); + List sentBodies = new CopyOnWriteArrayList<>(); + doAnswer( + invocation -> { + Request request = invocation.getArgument(0); + okio.Buffer body = new okio.Buffer(); + request.body().writeTo(body); + sentBodies.add(body.readByteArray()); + if (callCount.incrementAndGet() == 1) { + // a slow agent switches the sink to async mode + Thread.sleep(1001); + } + return respond(request, 200); + }) + .when(client) + .newCall(any()); + + sink.accept(1, Collections.singletonList(ByteBuffer.wrap(new byte[] {1}))); + byte[] first = {2, 3}; + byte[] second = {4, 5}; + sink.accept(1, Arrays.asList(ByteBuffer.wrap(first), ByteBuffer.wrap(second))); + assertTrue(sink.isInDegradedMode()); + // the writer starts its next payload in the same chunks + Arrays.fill(first, (byte) 9); + Arrays.fill(second, (byte) 9); + latch.await(10, SECONDS); + + assertEquals(2, sentBodies.size()); + assertArrayEquals(new byte[] {2, 3, 4, 5}, sentBodies.get(1)); + } + private static Call respond(Request request, int code) throws IOException { if (code == 0) { return error(); diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/SerializingMetricWriterTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/SerializingMetricWriterTest.java index bc05393cea3..84b8c5f93a0 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/SerializingMetricWriterTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/SerializingMetricWriterTest.java @@ -620,6 +620,57 @@ private static void serializeAndValidate(List content) { assertTrue(sink.validatedInput()); } + @Test + void payloadOverTheCapIsDroppedAndTheNextOneIsSent() { + long startTime = MILLISECONDS.toNanos(System.currentTimeMillis()); + long duration = SECONDS.toNanos(10); + WellKnownTags wellKnownTags = + new WellKnownTags("runtimeid", "hostname", "env", "service", "version", "language"); + AggregateEntry entry = + entry( + "resource", + "service", + "operation", + null, + "type", + 200, + false, + false, + "server", + Collections.emptyList(), + "GET", + "/api/users", + null, + 1); + List sent = new ArrayList<>(); + Sink sink = + new Sink() { + @Override + public void accept(int messageCount, List chunks) { + sent.add(messageCount); + } + + @Override + public void register(EventListener listener) {} + }; + SerializingMetricWriter writer = + new SerializingMetricWriter( + wellKnownTags, sink, 128, 4096, GitInfoProvider.INSTANCE, false); + + // a bucket well past the 4 KiB cap is dropped, without throwing + writer.startBucket(2000, startTime, duration); + for (int i = 0; i < 2000; ++i) { + writer.add(entry); + } + writer.finishBucket(); + assertTrue(sent.isEmpty()); + + writer.startBucket(1, startTime, duration); + writer.add(entry); + writer.finishBucket(); + assertEquals(Collections.singletonList(1), sent); + } + static final class ValidatingSink implements Sink { private final WellKnownTags wellKnownTags; @@ -666,16 +717,29 @@ static final class ValidatingSink implements Sink { public void register(EventListener listener) {} @Override - public void accept(int messageCount, ByteBuffer buffer) { + public void accept(int messageCount, List chunks) { // Sink.accept can't declare checked exceptions; surface any msgpack decode failure as an // assertion error so the test reports it. try { - validate(buffer); + validate(join(chunks)); } catch (IOException e) { throw new AssertionError("Failed to decode metric payload", e); } } + private static ByteBuffer join(List chunks) { + int size = 0; + for (ByteBuffer chunk : chunks) { + size += chunk.remaining(); + } + ByteBuffer joined = ByteBuffer.allocate(size); + for (ByteBuffer chunk : chunks) { + joined.put(chunk.duplicate()); + } + joined.flip(); + return joined; + } + private void validate(ByteBuffer buffer) throws IOException { MessageUnpacker unpacker = MessagePack.newDefaultUnpacker(buffer); int mapSize = unpacker.unpackMapHeader(); diff --git a/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java b/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java index c9e64e3d60e..dd5a1bde8d9 100644 --- a/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java +++ b/internal-api/src/main/java/datadog/trace/util/ChunkedBuffer.java @@ -13,8 +13,8 @@ * collector: G1 treats anything over half a region as humongous (1 MB regions on heaps up to about * 2 GB), Shenandoah anything over a region, and ZGC moves anything over 256 KiB off its small * pages. Humongous objects sit in old gen, are never compacted, and push the collector toward - * concurrent cycles. Each chunk here is an array of {@link #CHUNK_SIZE} bytes, a whole object of at - * most 256 KiB for any array header layout, which is regular-sized on all of them. + * concurrent cycles. Each chunk here is an array of at most {@link #CHUNK_SIZE} bytes, a whole + * object of at most 256 KiB for any array header layout, which is regular-sized on all of them. * *

Messages, marks and rewinding. Callers write a message, then call {@link #mark()} to * complete it. {@link #rewind()} drops everything written since the last mark, so a message that @@ -36,6 +36,10 @@ * so the buffer holds exactly as much as a single array of the same capacity would. The last chunk * is only as long as the remaining capacity, so a small buffer is a single small chunk. * + *

Small payloads stay small. Chunks start at {@link #DEFAULT_FIRST_CHUNK_SIZE} and double + * up to {@link #CHUNK_SIZE}, so a buffer whose payloads are usually a few KB holds a few KB rather + * than a whole chunk. A large payload reaches full-size chunks after about 250 KB. + * *

Retention. Chunks are allocated on first use. {@link #reset()} keeps the chunks that * held completed data plus half of any surplus kept from the previous reset, so after a burst the * extra chunks drain over a few resets instead of all at once. Payloads that vary in size don't @@ -54,10 +58,17 @@ public class ChunkedBuffer { */ public static final int CHUNK_SIZE = 256 * 1024 - 32; + /** The size of the first chunk unless another is given; later chunks double. */ + public static final int DEFAULT_FIRST_CHUNK_SIZE = 4 * 1024; + /** Where writes go once the buffer has failed; reused from the start whenever it fills. */ private static final int SCRATCH_SIZE = 64; + /** The smallest first chunk: large enough that any primitive fits in a fresh chunk. */ + private static final int MIN_FIRST_CHUNK_SIZE = 64; + private final int maxBytes; + private final int firstChunkSize; private final int maxChunks; private final byte[][] chunks; private final ByteBuffer[] views; @@ -88,9 +99,18 @@ public class ChunkedBuffer { * @param maxBytes the most data the buffer may hold */ public ChunkedBuffer(int maxBytes) { + this(maxBytes, DEFAULT_FIRST_CHUNK_SIZE); + } + + /** + * @param maxBytes the most data the buffer may hold + * @param firstChunkSize the size of the first chunk; each later chunk doubles, up to {@link + * #CHUNK_SIZE} + */ + public ChunkedBuffer(int maxBytes, int firstChunkSize) { this.maxBytes = Math.max(0, maxBytes); - // each chunk may leave up to 7 bytes unused, so allow for one more chunk than whole chunks - this.maxChunks = this.maxBytes == 0 ? 0 : this.maxBytes / (CHUNK_SIZE - 7) + 1; + this.firstChunkSize = Math.max(MIN_FIRST_CHUNK_SIZE, Math.min(CHUNK_SIZE, firstChunkSize)); + this.maxChunks = chunksNeeded(); this.chunks = new byte[maxChunks][]; this.views = new ByteBuffer[maxChunks]; this.fills = new int[maxChunks]; @@ -306,7 +326,7 @@ private int advance(int width) { fail(); return 0; } - int length = Math.min(CHUNK_SIZE, remaining); + int length = Math.min(chunkSize(next), remaining); if (next == allocatedChunks) { ++allocatedChunks; allocate(next, length); @@ -320,6 +340,26 @@ private int advance(int width) { return 0; } + /** The full size of the chunk at {@code index}: doubling from the first chunk, up to the max. */ + private int chunkSize(int index) { + long size = (long) firstChunkSize << Math.min(index, 31); + return (int) Math.min(CHUNK_SIZE, size); + } + + /** Enough chunks to hold {@code maxBytes} even if each leaves up to 7 bytes unused. */ + private int chunksNeeded() { + if (maxBytes == 0) { + return 0; + } + long covered = 0; + int count = 0; + while (covered < maxBytes) { + covered += chunkSize(count) - 7; + ++count; + } + return count + 1; + } + private void allocate(int index, int length) { byte[] chunk = new byte[length]; chunks[index] = chunk; diff --git a/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java b/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java index 4ee50d1475b..c718e6f454a 100644 --- a/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java +++ b/internal-api/src/test/java/datadog/trace/util/ChunkedBufferTest.java @@ -26,14 +26,14 @@ void chunkObjectFitsIn256KiBForEveryArrayHeaderLayout() { @Test void capacityIsExact() { - assertEquals(5 << 20, new ChunkedBuffer(5 << 20).capacity()); - assertEquals(1024, new ChunkedBuffer(1024).capacity()); - assertEquals(0, new ChunkedBuffer(0).capacity()); + assertEquals(5 << 20, new ChunkedBuffer(5 << 20, CHUNK_SIZE).capacity()); + assertEquals(1024, new ChunkedBuffer(1024, CHUNK_SIZE).capacity()); + assertEquals(0, new ChunkedBuffer(0, CHUNK_SIZE).capacity()); } @Test void lastChunkIsShortenedToFitCapacity() { - ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE + 10); + ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE + 10, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE + 2]); buffer.putLong(5L); assertTrue(buffer.mark()); @@ -46,7 +46,7 @@ void lastChunkIsShortenedToFitCapacity() { @Test void unusedChunkTailsDoNotReduceCapacity() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE - 3]); // leaves 3 unused bytes at the end of the first chunk buffer.putLong(1L); @@ -61,7 +61,7 @@ void unusedChunkTailsDoNotReduceCapacity() { @Test void keptShortChunkIsReplacedWhenAPayloadNeedsMoreOfIt() { - ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE + 10); + ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE + 10, CHUNK_SIZE); // no unused tail, so the second chunk only needs to be 10 bytes long buffer.put(new byte[CHUNK_SIZE + 10]); assertTrue(buffer.mark()); @@ -78,7 +78,7 @@ void keptShortChunkIsReplacedWhenAPayloadNeedsMoreOfIt() { @Test void smallBufferIsOneSmallChunk() { - ChunkedBuffer buffer = new ChunkedBuffer(16); + ChunkedBuffer buffer = new ChunkedBuffer(16, CHUNK_SIZE); buffer.putLong(1L); buffer.putLong(2L); assertTrue(buffer.mark()); @@ -91,7 +91,7 @@ void smallBufferIsOneSmallChunk() { @Test void zeroCapacityRejectsEveryMessage() { - ChunkedBuffer buffer = new ChunkedBuffer(0); + ChunkedBuffer buffer = new ChunkedBuffer(0, CHUNK_SIZE); buffer.putInt(1); buffer.put(new byte[10]); assertFalse(buffer.mark()); @@ -101,7 +101,7 @@ void zeroCapacityRejectsEveryMessage() { @Test void allocatesNothingBeforeFirstWrite() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); assertEquals(0, buffer.allocatedChunks()); buffer.mark(); buffer.reset(); @@ -113,7 +113,7 @@ void allocatesNothingBeforeFirstWrite() { @Test void typedPutsAreBigEndianAcrossChunkBoundaries() { // a mix of widths, repeated enough to cross several chunk boundaries at varying offsets - ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS, CHUNK_SIZE); ByteBuffer expected = ByteBuffer.allocate(THREE_CHUNKS); int i = 0; while (expected.position() < TWO_CHUNKS + CHUNK_SIZE / 2) { @@ -144,7 +144,7 @@ void typedPutsAreBigEndianAcrossChunkBoundaries() { @Test void primitiveThatDoesNotFitStartsTheNextChunk() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE - 3]); buffer.putLong(0x0102030405060708L); buffer.mark(); @@ -159,7 +159,7 @@ void primitiveThatDoesNotFitStartsTheNextChunk() { @Test void byteArraysSplitAcrossChunks() { - ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS, CHUNK_SIZE); byte[] data = sequence(CHUNK_SIZE + CHUNK_SIZE / 2); buffer.put((byte) -1); buffer.put(data, 0, data.length); @@ -177,7 +177,7 @@ void byteArraysSplitAcrossChunks() { @Test void byteBufferSplitsAcrossChunks() { - ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS, CHUNK_SIZE); byte[] data = sequence(CHUNK_SIZE + 100); ByteBuffer src = ByteBuffer.wrap(data); buffer.put(src); @@ -189,7 +189,7 @@ void byteBufferSplitsAcrossChunks() { @Test void emptyMarksAreNotCounted() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.mark(); buffer.mark(); assertEquals(0, buffer.messageCount()); @@ -202,7 +202,7 @@ void emptyMarksAreNotCounted() { @Test void rewindDiscardsUnmarkedBytesAcrossChunks() { - ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(THREE_CHUNKS, CHUNK_SIZE); buffer.putInt(42); buffer.mark(); // spill the unfinished message into a second chunk @@ -222,7 +222,7 @@ void rewindDiscardsUnmarkedBytesAcrossChunks() { @Test void rewindBeforeAnyMarkEmptiesTheBuffer() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE + 10]); buffer.rewind(); assertFalse(buffer.isDirty()); @@ -235,7 +235,7 @@ void rewindBeforeAnyMarkEmptiesTheBuffer() { @Test void overflowFailsWithoutThrowingAndMarkRewinds() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); byte[] first = sequence(CHUNK_SIZE + 10); buffer.put(first); assertTrue(buffer.mark()); @@ -267,7 +267,7 @@ void overflowFailsWithoutThrowingAndMarkRewinds() { @Test void messageLargerThanCapacityFailsFromEmpty() { - ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE + 1]); assertFalse(buffer.mark()); assertFalse(buffer.isDirty()); @@ -276,7 +276,7 @@ void messageLargerThanCapacityFailsFromEmpty() { @Test void writesFillCapacityExactly() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.put(new byte[TWO_CHUNKS - 8]); buffer.putLong(1L); assertTrue(buffer.mark()); @@ -289,7 +289,7 @@ void writesFillCapacityExactly() { @Test void rewindAlsoClearsFailure() { - ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(CHUNK_SIZE, CHUNK_SIZE); buffer.put(new byte[CHUNK_SIZE + 1]); buffer.rewind(); buffer.putInt(5); @@ -299,7 +299,7 @@ void rewindAlsoClearsFailure() { @Test void resetKeepsTheChunksTheCompletedMessagesUsed() { - ChunkedBuffer buffer = new ChunkedBuffer(5 * CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(5 * CHUNK_SIZE, CHUNK_SIZE); buffer.put(new byte[2 * CHUNK_SIZE + 1]); buffer.mark(); assertEquals(3, buffer.allocatedChunks()); @@ -312,7 +312,7 @@ void resetKeepsTheChunksTheCompletedMessagesUsed() { @Test void resetReleasesHalfTheSurplusEachTime() { - ChunkedBuffer buffer = new ChunkedBuffer(20 * CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(20 * CHUNK_SIZE, CHUNK_SIZE); buffer.put(new byte[18 * CHUNK_SIZE + 1]); buffer.mark(); buffer.reset(); @@ -330,7 +330,7 @@ void resetReleasesHalfTheSurplusEachTime() { @Test void growthIsNotHalved() { - ChunkedBuffer buffer = new ChunkedBuffer(5 * CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(5 * CHUNK_SIZE, CHUNK_SIZE); buffer.putInt(1); buffer.mark(); buffer.reset(); @@ -344,7 +344,7 @@ void growthIsNotHalved() { @Test void resetIgnoresUnfinishedAndFailedMessages() { - ChunkedBuffer buffer = new ChunkedBuffer(3 * CHUNK_SIZE); + ChunkedBuffer buffer = new ChunkedBuffer(3 * CHUNK_SIZE, CHUNK_SIZE); buffer.putInt(1); buffer.mark(); // an oversized message touches every chunk before failing @@ -355,7 +355,7 @@ void resetIgnoresUnfinishedAndFailedMessages() { @Test void resetReusesRetainedChunks() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.put(sequence(CHUNK_SIZE + 10)); buffer.mark(); ByteBuffer before = buffer.completedChunks().get(1); @@ -370,7 +370,7 @@ void resetReusesRetainedChunks() { @Test void completedChunksAreIndependentViews() { - ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS); + ChunkedBuffer buffer = new ChunkedBuffer(TWO_CHUNKS, CHUNK_SIZE); buffer.putLong(3L); buffer.mark(); @@ -404,6 +404,61 @@ protected void onCapacityExceeded() { assertEquals(8, buffer.completedBytes()); } + @Test + void defaultFirstChunkIsSmall() { + ChunkedBuffer buffer = new ChunkedBuffer(5 << 20); + buffer.putInt(1); + assertTrue(buffer.mark()); + assertEquals( + ChunkedBuffer.DEFAULT_FIRST_CHUNK_SIZE, buffer.completedChunks().get(0).array().length); + } + + @Test + void chunksStartSmallAndDoubleUpToChunkSize() { + ChunkedBuffer buffer = new ChunkedBuffer(4 * CHUNK_SIZE, 4096); + buffer.put(new byte[4096 + 8192 + 16384 + 32768 + 65536 + 131072 + CHUNK_SIZE + 1]); + assertTrue(buffer.mark()); + + List chunks = buffer.completedChunks(); + int[] expected = {4096, 8192, 16384, 32768, 65536, 131072, CHUNK_SIZE, 1}; + assertEquals(expected.length, chunks.size()); + for (int i = 0; i < expected.length; ++i) { + assertEquals(expected[i], chunks.get(i).remaining(), "chunk " + i); + } + } + + @Test + void smallPayloadKeepsOnlyASmallChunk() { + ChunkedBuffer buffer = new ChunkedBuffer(32 << 20, 4096); + buffer.put(new byte[100]); + assertTrue(buffer.mark()); + assertEquals(4096, buffer.completedChunks().get(0).array().length); + buffer.reset(); + assertEquals(1, buffer.allocatedChunks()); + } + + @Test + void capacityIsExactWithSmallFirstChunks() { + ChunkedBuffer buffer = new ChunkedBuffer(300_000, 4096); + // primitives leave unused tails in the small chunks + for (int i = 0; i < 300_000 / 8; ++i) { + buffer.putLong(i); + } + assertTrue(buffer.mark()); + assertEquals(300_000, buffer.completedBytes()); + + buffer.put((byte) 1); + assertFalse(buffer.mark()); + } + + @Test + void firstChunkSizeIsLargeEnoughForAnyPrimitive() { + ChunkedBuffer buffer = new ChunkedBuffer(1024, 1); + buffer.putLong(7L); + assertTrue(buffer.mark()); + assertEquals(7L, buffer.completedChunks().get(0).getLong()); + } + private static byte[] sequence(int length) { byte[] data = new byte[length]; for (int i = 0; i < length; ++i) {