From fee78f99a7c8956e7e05b3f22ebad6780767f4df Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Sun, 13 Sep 2026 17:05:48 +0100 Subject: [PATCH] perf: optimise UnsynchronizedByteArrayInputStream bulk operations Motivation: UnsynchronizedByteArrayInputStream only overrode the basic read/skip methods, so readAllBytes, readNBytes, skipNBytes and transferTo fell through to the generic InputStream defaults. Those allocate 16 KiB scratch buffers in a loop and then concatenate, which is wasteful when the whole payload already sits in a byte[]. skip also advanced the offset unconditionally with Math.addExact/toIntExact, so it could throw ArithmeticException for large values and leave offset past eod. Modification: - Override readAllBytes, readNBytes(int), readNBytes(byte[],int,int), skipNBytes and transferTo to operate directly on the backing array (one Arrays.copyOfRange / OutputStream.write each). - Clamp skip to the remaining bytes so offset never exceeds eod and large skips cannot overflow; simplify available() and readLocal() accordingly. - Simplify the (data, offset, length) constructor so offset/eod are clamped to the array with long arithmetic (no int overflow), and use Objects.checkFromIndexSize for the read(byte[],int,int) bounds check. - Expand UnsynchronizedByteArrayInputStreamSpec to cover every method and edge case (EOF, zero-length reads, clamping, overflow). - Add readAllBytes and transferTo cases to ByteString_asInputStream_Benchmark. Result: readAllBytes/readNBytes/transferTo on ByteString.asInputStream copy the data once instead of chunking through temporary buffers. Local JMH (-f 1 -wi 2 -i 3) for single_bs_as_input_stream_read_all_bytes: 10 KB 201k -> 503k ops/s, 1000 KB 3.3k -> 5.5k ops/s. Tests: - sbt "actor-tests/testOnly org.apache.pekko.util.UnsynchronizedByteArrayInputStreamSpec org.apache.pekko.util.ByteStringSpec" (226 passed) - sbt "actor/mimaReportBinaryIssues" (no issues) - sbt "bench-jmh/Jmh/compile" and Jmh/run of the new benchmarks - scalafmt on changed Scala files, sbt actor/javafmtAll (JDK 17) - git diff --check References: None - follow-up to #2300 which introduced this class --- ...synchronizedByteArrayInputStreamSpec.scala | 120 +++++++++++++++++- .../UnsynchronizedByteArrayInputStream.java | 104 +++++++++------ .../ByteString_asInputStream_Benchmark.scala | 14 +- 3 files changed, 198 insertions(+), 40 deletions(-) diff --git a/actor-tests/src/test/scala/org/apache/pekko/io/UnsynchronizedByteArrayInputStreamSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/io/UnsynchronizedByteArrayInputStreamSpec.scala index 0fdff98c5a9..12a38c2ea48 100644 --- a/actor-tests/src/test/scala/org/apache/pekko/io/UnsynchronizedByteArrayInputStreamSpec.scala +++ b/actor-tests/src/test/scala/org/apache/pekko/io/UnsynchronizedByteArrayInputStreamSpec.scala @@ -17,6 +17,7 @@ package org.apache.pekko.util +import java.io.{ ByteArrayOutputStream, EOFException } import java.nio.charset.StandardCharsets import org.apache.pekko @@ -26,9 +27,13 @@ import org.scalatest.matchers.should.Matchers import org.scalatest.wordspec.AnyWordSpec class UnsynchronizedByteArrayInputStreamSpec extends AnyWordSpec with Matchers { + + private def bytes(s: String): Array[Byte] = s.getBytes(StandardCharsets.UTF_8) + private def str(b: Array[Byte]): String = new String(b, StandardCharsets.UTF_8) + "UnsynchronizedByteArrayInputStream" must { "support mark and reset" in { - val stream = new UnsynchronizedByteArrayInputStream("abc".getBytes(StandardCharsets.UTF_8)) + val stream = new UnsynchronizedByteArrayInputStream(bytes("abc")) stream.markSupported() should ===(true) stream.read() should ===('a') stream.mark(1) // the parameter value (a readAheadLimit) is ignored as it is in ByteArrayInputStream too @@ -38,15 +43,124 @@ class UnsynchronizedByteArrayInputStreamSpec extends AnyWordSpec with Matchers { stream.close() } "support skip" in { - val stream = new UnsynchronizedByteArrayInputStream("abc".getBytes(StandardCharsets.UTF_8)) + val stream = new UnsynchronizedByteArrayInputStream(bytes("abc")) stream.skip(1) should ===(1) stream.read() should ===('b') stream.close() } "support skip with large value" in { - val stream = new UnsynchronizedByteArrayInputStream("abc".getBytes(StandardCharsets.UTF_8)) + val stream = new UnsynchronizedByteArrayInputStream(bytes("abc")) stream.skip(50) should ===(3) // only 3 bytes to skip + stream.available() should ===(0) + stream.read() should ===(-1) + stream.skip(Long.MaxValue) should ===(0) // must not overflow + stream.available() should ===(0) + stream.close() + } + "reject negative skip" in { + val stream = new UnsynchronizedByteArrayInputStream(bytes("abc")) + an[IllegalArgumentException] should be thrownBy stream.skip(-1) + stream.close() + } + "support skipNBytes" in { + val stream = new UnsynchronizedByteArrayInputStream(bytes("abcdef")) + stream.skipNBytes(0) + stream.skipNBytes(-1) // no-op, as in InputStream + stream.read() should ===('a') + stream.skipNBytes(2) + stream.read() should ===('d') + an[EOFException] should be thrownBy stream.skipNBytes(3) + stream.available() should ===(0) + stream.close() + } + "support read into array" in { + val stream = new UnsynchronizedByteArrayInputStream(bytes("abcdef")) + val buf = new Array[Byte](4) + stream.read(buf) should ===(4) + str(buf) should ===("abcd") + stream.read(buf, 1, 0) should ===(0) + stream.read(buf, 1, 3) should ===(2) + str(buf) should ===("aefd") + stream.read(buf) should ===(-1) + stream.read(buf, 0, 0) should ===(0) // zero-length read at EOF returns 0, not -1 + an[IndexOutOfBoundsException] should be thrownBy stream.read(buf, 2, 3) + an[IndexOutOfBoundsException] should be thrownBy stream.read(buf, -1, 1) + stream.close() + } + "support readAllBytes" in { + val stream = new UnsynchronizedByteArrayInputStream(bytes("abcdef")) + stream.read() should ===('a') + str(stream.readAllBytes()) should ===("bcdef") + stream.available() should ===(0) + stream.readAllBytes() should ===(Array.emptyByteArray) + stream.close() + } + "support readNBytes(int)" in { + val stream = new UnsynchronizedByteArrayInputStream(bytes("abcdef")) + str(stream.readNBytes(2)) should ===("ab") + stream.readNBytes(0) should ===(Array.emptyByteArray) + str(stream.readNBytes(Int.MaxValue)) should ===("cdef") + stream.readNBytes(1) should ===(Array.emptyByteArray) + an[IllegalArgumentException] should be thrownBy stream.readNBytes(-1) + stream.close() + } + "support readNBytes(byte[], int, int)" in { + val stream = new UnsynchronizedByteArrayInputStream(bytes("abcdef")) + val buf = new Array[Byte](4) + stream.readNBytes(buf, 1, 3) should ===(3) + str(buf.slice(1, 4)) should ===("abc") + stream.readNBytes(buf, 0, 4) should ===(3) + str(buf.slice(0, 3)) should ===("def") + stream.readNBytes(buf, 0, 4) should ===(0) // 0 at EOF, not -1 + an[IndexOutOfBoundsException] should be thrownBy stream.readNBytes(buf, 2, 3) + stream.close() + } + "support transferTo" in { + val stream = new UnsynchronizedByteArrayInputStream(bytes("abcdef")) + stream.read() should ===('a') + val out = new ByteArrayOutputStream() + stream.transferTo(out) should ===(5) + out.toString(StandardCharsets.UTF_8) should ===("bcdef") + stream.transferTo(out) should ===(0) + out.size() should ===(5) + stream.close() + } + "support offset and length constructor" in { + val stream = new UnsynchronizedByteArrayInputStream(bytes("abcdef"), 1, 3) + stream.available() should ===(3) + stream.read() should ===('b') + stream.mark(0) + str(stream.readAllBytes()) should ===("cd") stream.read() should ===(-1) + stream.reset() + str(stream.readNBytes(10)) should ===("cd") + stream.close() + } + "clamp offset and length to the array bounds" in { + val overLength = new UnsynchronizedByteArrayInputStream(bytes("abc"), 1, Int.MaxValue) + overLength.available() should ===(2) + str(overLength.readAllBytes()) should ===("bc") + overLength.close() + + val overOffset = new UnsynchronizedByteArrayInputStream(bytes("abc"), 10, 2) + overOffset.available() should ===(0) + overOffset.read() should ===(-1) + overOffset.readAllBytes() should ===(Array.emptyByteArray) + overOffset.close() + + val empty = new UnsynchronizedByteArrayInputStream(Array.emptyByteArray, 5, 5) + empty.available() should ===(0) + empty.read() should ===(-1) + empty.close() + + an[IllegalArgumentException] should be thrownBy new UnsynchronizedByteArrayInputStream(bytes("abc"), -1, 1) + an[IllegalArgumentException] should be thrownBy new UnsynchronizedByteArrayInputStream(bytes("abc"), 0, -1) + } + "not copy the backing array" in { + val arr = bytes("abc") + val stream = new UnsynchronizedByteArrayInputStream(arr) + arr(0) = 'z'.toByte + stream.read() should ===('z') stream.close() } } diff --git a/actor/src/main/java/org/apache/pekko/io/UnsynchronizedByteArrayInputStream.java b/actor/src/main/java/org/apache/pekko/io/UnsynchronizedByteArrayInputStream.java index 943cf3ddfb1..86f0a77b9d3 100644 --- a/actor/src/main/java/org/apache/pekko/io/UnsynchronizedByteArrayInputStream.java +++ b/actor/src/main/java/org/apache/pekko/io/UnsynchronizedByteArrayInputStream.java @@ -18,7 +18,11 @@ package org.apache.pekko.io; import java.io.ByteArrayInputStream; +import java.io.EOFException; +import java.io.IOException; import java.io.InputStream; +import java.io.OutputStream; +import java.util.Arrays; import java.util.Objects; import org.apache.pekko.annotation.InternalApi; @@ -26,11 +30,15 @@ * Internal API: An unsynchronized byte array input stream. This class does not copy the provided * byte array, and it is not thread safe. * + *

All bulk operations ({@link #readAllBytes()}, {@link #readNBytes(int)}, {@link + * #transferTo(OutputStream)}, ...) are overridden so that they operate directly on the backing + * array instead of going through the chunked, allocation-heavy defaults in {@link InputStream}. + * * @see ByteArrayInputStream * @since 2.0.0 */ // @NotThreadSafe -// copied from +// originally copied from // https://github.com/apache/commons-io/blob/26e5aa9661a72bfd9697fb384ca72f58e5d672e9/src/main/java/org/apache/commons/io/input/UnsynchronizedByteArrayInputStream.java @InternalApi public class UnsynchronizedByteArrayInputStream extends InputStream { @@ -38,11 +46,6 @@ public class UnsynchronizedByteArrayInputStream extends InputStream { /** The end of stream marker. */ private static final int END_OF_STREAM = -1; - private static int minPosLen(final byte[] data, final int defaultValue) { - requireNonNegative(defaultValue, "defaultValue"); - return Math.min(defaultValue, data.length > 0 ? data.length : defaultValue); - } - private static int requireNonNegative(final int value, final String name) { if (value < 0) { throw new IllegalArgumentException(name + " cannot be negative"); @@ -50,21 +53,14 @@ private static int requireNonNegative(final int value, final String name) { return value; } - private static void checkFromIndexSize(final byte[] array, final int off, final int len) { - final int arrayLength = Objects.requireNonNull(array, "byte array").length; - if ((off | len | arrayLength) < 0 || arrayLength - len < off) { - throw new IndexOutOfBoundsException( - "Range [%s, %Similar to data.length, which is the last readable offset + 1. + *

Similar to data.length, which is the last readable offset + 1. Invariant: {@code offset <= + * eod <= data.length}. */ private final int eod; @@ -98,14 +94,16 @@ public UnsynchronizedByteArrayInputStream(final byte[] data, final int offset, f requireNonNegative(offset, "offset"); requireNonNegative(length, "length"); this.data = Objects.requireNonNull(data, "data"); - this.eod = Math.min(minPosLen(data, offset) + length, data.length); - this.offset = minPosLen(data, offset); - this.markedOffset = minPosLen(data, offset); + final int start = Math.min(offset, data.length); + this.offset = start; + this.markedOffset = start; + // long arithmetic avoids int overflow for large offset + length + this.eod = (int) Math.min((long) start + length, data.length); } @Override public int available() { - return offset < eod ? eod - offset : 0; + return eod - offset; } @SuppressWarnings("sync-override") @@ -132,31 +130,48 @@ public int read(final byte[] dest) { @Override public int read(final byte[] dest, final int off, final int len) { - checkFromIndexSize(dest, off, len); + Objects.checkFromIndexSize(off, len, Objects.requireNonNull(dest, "dest").length); return readLocal(dest, off, len); } - private final int readLocal(final byte[] dest, final int off, final int len) { + private int readLocal(final byte[] dest, final int off, final int len) { if (len == 0) { return 0; } - - if (offset >= eod) { - return END_OF_STREAM; - } - - int actualLen = eod - offset; - if (len < actualLen) { - actualLen = len; - } + final int actualLen = Math.min(len, eod - offset); if (actualLen <= 0) { - return 0; + return END_OF_STREAM; } System.arraycopy(data, offset, dest, off, actualLen); offset += actualLen; return actualLen; } + @Override + public byte[] readAllBytes() { + final byte[] result = Arrays.copyOfRange(data, offset, eod); + offset = eod; + return result; + } + + @Override + public byte[] readNBytes(final int len) { + if (len < 0) { + // same exception type/message as InputStream.readNBytes(int) + throw new IllegalArgumentException("len < 0"); + } + final int actualLen = Math.min(len, eod - offset); + final byte[] result = Arrays.copyOfRange(data, offset, offset + actualLen); + offset += actualLen; + return result; + } + + @Override + public int readNBytes(final byte[] dest, final int off, final int len) { + final int n = read(dest, off, len); + return n == END_OF_STREAM ? 0 : n; + } + @SuppressWarnings("sync-override") @Override public void reset() { @@ -168,13 +183,30 @@ public long skip(final long n) { if (n < 0) { throw new IllegalArgumentException("Skipping backward is not supported"); } + final int actualSkip = (int) Math.min(n, eod - offset); + offset += actualSkip; + return actualSkip; + } - long actualSkip = eod - offset; - if (n < actualSkip) { - actualSkip = n; + @Override + public void skipNBytes(final long n) throws IOException { + if (n > 0) { + if (n > eod - offset) { + offset = eod; + throw new EOFException(); + } + offset += (int) n; } + } - offset = Math.addExact(offset, Math.toIntExact(n)); - return actualSkip; + @Override + public long transferTo(final OutputStream out) throws IOException { + Objects.requireNonNull(out, "out"); + final int len = eod - offset; + if (len > 0) { + out.write(data, offset, len); + offset = eod; + } + return len; } } diff --git a/bench-jmh/src/main/scala/org/apache/pekko/util/ByteString_asInputStream_Benchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/util/ByteString_asInputStream_Benchmark.scala index e6b845226e8..fb6edbe0191 100644 --- a/bench-jmh/src/main/scala/org/apache/pekko/util/ByteString_asInputStream_Benchmark.scala +++ b/bench-jmh/src/main/scala/org/apache/pekko/util/ByteString_asInputStream_Benchmark.scala @@ -17,7 +17,7 @@ package org.apache.pekko.util -import java.io.InputStream +import java.io.{ ByteArrayOutputStream, InputStream } import java.util.concurrent.TimeUnit import org.openjdk.jmh.annotations._ @@ -88,6 +88,18 @@ class ByteString_asInputStream_Benchmark { blackhole.consume(countBytes(composed.asInputStream)) } + @Benchmark + def single_bs_as_input_stream_read_all_bytes(blackhole: Blackhole): Unit = { + blackhole.consume(bs.asInputStream.readAllBytes()) + } + + @Benchmark + def single_bs_as_input_stream_transfer_to(blackhole: Blackhole): Unit = { + val out = new ByteArrayOutputStream(bs.length) + bs.asInputStream.transferTo(out) + blackhole.consume(out.toByteArray) + } + private def countBytes(stream: InputStream): Int = { val buffer = new Array[Byte](1024) var count = 0