diff --git a/server/src/main/java/org/elasticsearch/common/bytes/CompositeBytesReference.java b/server/src/main/java/org/elasticsearch/common/bytes/CompositeBytesReference.java index 537082fedd602..94c7a324894ae 100644 --- a/server/src/main/java/org/elasticsearch/common/bytes/CompositeBytesReference.java +++ b/server/src/main/java/org/elasticsearch/common/bytes/CompositeBytesReference.java @@ -274,4 +274,8 @@ public double getDoubleLE(int index) { } return super.getDoubleLE(index); } + + public BytesReference[] references() { + return references; + } } diff --git a/server/src/main/java/org/elasticsearch/common/bytes/ReleasableBytesReference.java b/server/src/main/java/org/elasticsearch/common/bytes/ReleasableBytesReference.java index fa3789eaedc3b..1a461c6686000 100644 --- a/server/src/main/java/org/elasticsearch/common/bytes/ReleasableBytesReference.java +++ b/server/src/main/java/org/elasticsearch/common/bytes/ReleasableBytesReference.java @@ -94,6 +94,10 @@ public ReleasableBytesReference retainedSlice(int from, int length) { return new ReleasableBytesReference(slice, refCounted); } + public BytesReference delegate() { + return delegate; + } + @Override public void close() { refCounted.decRef(); diff --git a/server/src/main/java/org/elasticsearch/common/io/stream/DelayableWriteable.java b/server/src/main/java/org/elasticsearch/common/io/stream/DelayableWriteable.java index e1161f03cb8a7..23590a07a7d48 100644 --- a/server/src/main/java/org/elasticsearch/common/io/stream/DelayableWriteable.java +++ b/server/src/main/java/org/elasticsearch/common/io/stream/DelayableWriteable.java @@ -11,12 +11,15 @@ import org.elasticsearch.TransportVersion; import org.elasticsearch.common.bytes.BytesReference; +import org.elasticsearch.common.bytes.CompositeBytesReference; import org.elasticsearch.common.bytes.ReleasableBytesReference; +import org.elasticsearch.common.util.PageCacheRecycler; import org.elasticsearch.core.Nullable; import org.elasticsearch.core.Releasable; import java.io.IOException; import java.io.UncheckedIOException; +import java.util.Arrays; import java.util.HashMap; import java.util.Map; @@ -203,8 +206,7 @@ public boolean isSerialized() { @Override public long getSerializedSize() { - // We're already serialized - return serialized.length(); + return pageAlignedRamUsedByReferenceBytes(serialized); } @Override @@ -213,6 +215,22 @@ public void close() { } } + /** + * Over-estimates retained RAM by rounding each component up to a + * {@link PageCacheRecycler#BYTE_PAGE_SIZE} page (composites sum per component). + * Matches how Netty retains page-sized buffers rather than exact payload lengths. + */ + static long pageAlignedRamUsedByReferenceBytes(BytesReference bytes) { + if (bytes instanceof ReleasableBytesReference r) { + return pageAlignedRamUsedByReferenceBytes(r.delegate()); + } + if (bytes instanceof CompositeBytesReference composited) { + return Arrays.stream(composited.references()).mapToLong(DelayableWriteable::pageAlignedRamUsedByReferenceBytes).sum(); + } + final long numPages = (bytes.length() + PageCacheRecycler.BYTE_PAGE_SIZE - 1L) / PageCacheRecycler.BYTE_PAGE_SIZE; + return numPages * PageCacheRecycler.BYTE_PAGE_SIZE; + } + /** * Returns the serialized size in bytes of the provided {@link Writeable}. */ diff --git a/server/src/test/java/org/elasticsearch/common/io/stream/DelayableWriteableTests.java b/server/src/test/java/org/elasticsearch/common/io/stream/DelayableWriteableTests.java index 5603ddf647db1..5327c693c3b97 100644 --- a/server/src/test/java/org/elasticsearch/common/io/stream/DelayableWriteableTests.java +++ b/server/src/test/java/org/elasticsearch/common/io/stream/DelayableWriteableTests.java @@ -10,6 +10,11 @@ package org.elasticsearch.common.io.stream; import org.elasticsearch.TransportVersion; +import org.elasticsearch.common.bytes.BytesArray; +import org.elasticsearch.common.bytes.BytesReference; +import org.elasticsearch.common.bytes.CompositeBytesReference; +import org.elasticsearch.common.bytes.ReleasableBytesReference; +import org.elasticsearch.common.util.PageCacheRecycler; import org.elasticsearch.test.ESTestCase; import org.elasticsearch.test.TransportVersionUtils; @@ -124,9 +129,12 @@ public void testRoundTripFromReferencingWithNamedWriteable() throws IOException } public void testRoundTripFromDelayed() throws IOException { - Example e = new Example(randomAlphaOfLength(5)); + Example e = new Example(randomAlphaOfLengthBetween(100, 1000)); DelayableWriteable original = DelayableWriteable.referencing(e).asSerialized(Example::new, writableRegistry()); assertTrue(original.isSerialized()); + long length = DelayableWriteable.getSerializedSize(e); + long page = PageCacheRecycler.BYTE_PAGE_SIZE; + assertThat(original.getSerializedSize(), equalTo(((length + page - 1) / page) * page)); roundTripTestCase(original, Example::new); } @@ -165,6 +173,26 @@ public void testAsSerializedIsNoopOnSerialized() throws IOException { assertSame(d, d.asSerialized(Example::new, writableRegistry())); } + public void testPageAlignedRamUsedByReferenceBytes() { + final int page = PageCacheRecycler.BYTE_PAGE_SIZE; + assertThat(DelayableWriteable.pageAlignedRamUsedByReferenceBytes(BytesArray.EMPTY), equalTo(0L)); + assertThat(DelayableWriteable.pageAlignedRamUsedByReferenceBytes(new BytesArray(new byte[1])), equalTo((long) page)); + assertThat(DelayableWriteable.pageAlignedRamUsedByReferenceBytes(new BytesArray(new byte[page])), equalTo((long) page)); + assertThat(DelayableWriteable.pageAlignedRamUsedByReferenceBytes(new BytesArray(new byte[page + 1])), equalTo(2L * page)); + + assertThat( + DelayableWriteable.pageAlignedRamUsedByReferenceBytes(ReleasableBytesReference.wrap(new BytesArray(new byte[1]))), + equalTo((long) page) + ); + + BytesReference composite = CompositeBytesReference.of( + new BytesArray(new byte[1]), + new BytesArray(new byte[page / 2]), + new BytesArray(new byte[page + 1]) + ); + assertThat(DelayableWriteable.pageAlignedRamUsedByReferenceBytes(composite), equalTo(4L * page)); + } + private void roundTripTestCase(DelayableWriteable original, Writeable.Reader reader) throws IOException { DelayableWriteable roundTripped = roundTrip(original, reader, TransportVersion.current()); assertThat(roundTripped.expand(), equalTo(original.expand()));