This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new a5cf13f26c perf: optimise UnsynchronizedByteArrayInputStream bulk
operations (#3541)
a5cf13f26c is described below
commit a5cf13f26c443961bf51e0bf0d98af523c46a514
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 9 08:36:33 2026 +0100
perf: optimise UnsynchronizedByteArrayInputStream bulk operations (#3541)
Motivation:
`UnsynchronizedByteArrayInputStream` (used by `ByteString.asInputStream`
and the Java/Jackson serializers) only overrode the basic `read`/`skip`
methods. `readAllBytes`, `readNBytes`, `skipNBytes` and `transferTo` fell
through to the generic `InputStream` defaults, which allocate 16 KiB scratch
buffers in a loop and then concatenate — wasteful when the whole payload
already sits in a `byte[]`. `skip` also advanced `offset` unconditionally via
`Math.addExact(offset, Math.toIntExact(n)) [...]
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) — the same set of overrides
`java.io.ByteArrayInputStream` has.
- Clamp `skip` to the remaining bytes so `offset <= eod` always holds and
large skips cannot overflow; simplify `available()` and `readLocal()`
accordingly.
- Simplify the `(data, offset, length)` constructor so `offset`/`eod` are
clamped to the array using 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, negative args).
- Add a randomized differential test that runs operation sequences against
both this stream and `java.io.ByteArrayInputStream`, asserting identical
results except for the two documented deviations (zero-length read at EOF
returns 0; negative `skip` throws `IllegalArgumentException`).
- 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. Observable
behaviour is otherwise unchanged (negative `skip` still throws, as before).
Local JMH (`-f 1 -wi 2 -i 3`, short run so error bars are wide):
| Benchmark | kb | before (ops/s) | after (ops/s) |
|---|---|---|---|
| `single_bs_as_input_stream_read_all_bytes` | 10 | 201,138 | 502,768 |
| `single_bs_as_input_stream_read_all_bytes` | 1000 | 3,307 | 5,460 |
| `single_bs_as_input_stream_transfer_to` | 10 | 238,211 | 267,642 |
| `single_bs_as_input_stream_transfer_to` | 1000 | 2,883 | 2,747 |
`transferTo` is dominated by the `ByteArrayOutputStream` copy on the
receiving side, so the difference there is within noise.
Tests:
- `sbt "actor-tests/testOnly
org.apache.pekko.util.UnsynchronizedByteArrayInputStreamSpec
org.apache.pekko.util.ByteStringSpec"` — 226 passed
- `sbt "actor-tests/testOnly
org.apache.pekko.util.UnsynchronizedByteArrayInputStreamSpec"` after adding the
differential test — 14 passed; injecting a bug into `readNBytes` makes the
differential test fail
- `sbt "actor/mimaReportBinaryIssues"` — no issues (class is
`@InternalApi`; only methods added)
- `sbt "bench-jmh/Jmh/compile"` and `Jmh/run` of the new benchmarks
- `scalafmt` on changed Scala files, `sbt actor/javafmtAll` on JDK 17
- `git diff --check`
References:
None - follow-up to #2300 which introduced this class
---
.../UnsynchronizedByteArrayInputStreamSpec.scala | 209 ++++++++++++++++++++-
.../io/UnsynchronizedByteArrayInputStream.java | 104 ++++++----
.../util/ByteString_asInputStream_Benchmark.scala | 14 +-
3 files changed, 287 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 0fdff98c5a..522d031d9d 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,8 +17,11 @@
package org.apache.pekko.util
+import java.io.{ ByteArrayInputStream, ByteArrayOutputStream, EOFException }
import java.nio.charset.StandardCharsets
+import scala.util.{ Random, Try }
+
import org.apache.pekko
import pekko.io.UnsynchronizedByteArrayInputStream
@@ -26,9 +29,17 @@ 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)
+
+ /** Result of an operation, with a failure reduced to its exception class,
so two streams can be compared. */
+ private def outcome[A](f: => A): Either[String, A] =
+ Try(f).toEither.left.map(_.getClass.getName)
+
"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,16 +49,208 @@ 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()
+ }
+ "behave like java.io.ByteArrayInputStream under randomized operation
sequences" in {
+ // Differential test: every operation is run against both this stream
and a ByteArrayInputStream
+ // constructed identically, and the results (return values, exceptions,
buffer contents, available())
+ // must match. The only allowed deviations are the two documented ones:
+ // - a zero-length read at EOF returns 0 here, ByteArrayInputStream
returns -1
+ // - a negative skip throws IllegalArgumentException here,
ByteArrayInputStream returns 0
+ val rnd = new Random(20260922L)
+ for (iteration <- 1 to 500) {
+ val data = new Array[Byte](rnd.nextInt(48))
+ rnd.nextBytes(data)
+ val (stream, reference) =
+ if (rnd.nextBoolean()) (new
UnsynchronizedByteArrayInputStream(data), new ByteArrayInputStream(data))
+ else {
+ val off = rnd.nextInt(data.length + 1)
+ val len = rnd.nextInt(data.length - off + 1)
+ (new UnsynchronizedByteArrayInputStream(data, off, len), new
ByteArrayInputStream(data, off, len))
+ }
+ for (step <- 1 to 40) {
+ val op = rnd.nextInt(13)
+ withClue(s"iteration $iteration, step $step, op $op: ") {
+ op match {
+ case 0 =>
+ stream.read() should ===(reference.read())
+ case 1 =>
+ val a = new Array[Byte](rnd.nextInt(6))
+ val b = a.clone()
+ val expected = reference.read(b)
+ if (a.length == 0 && expected == -1) stream.read(a) should
===(0) // documented deviation
+ else stream.read(a) should ===(expected)
+ a should ===(b)
+ case 2 =>
+ val a = new Array[Byte](rnd.nextInt(8))
+ val b = a.clone()
+ val off = rnd.nextInt(a.length + 2) - 1 // may be out of bounds
+ val len = rnd.nextInt(a.length + 2) - 1
+ val expected = outcome(reference.read(b, off, len))
+ if (len == 0 && expected == Right(-1)) stream.read(a, off,
len) should ===(0) // documented deviation
+ else outcome(stream.read(a, off, len)) should ===(expected)
+ a should ===(b)
+ case 3 =>
+ stream.readAllBytes() should ===(reference.readAllBytes())
+ case 4 =>
+ val len = rnd.nextInt(12) - 1 // may be negative
+ outcome(stream.readNBytes(len).toSeq) should
===(outcome(reference.readNBytes(len).toSeq))
+ case 5 =>
+ val a = new Array[Byte](rnd.nextInt(8))
+ val b = a.clone()
+ val off = rnd.nextInt(a.length + 2) - 1 // may be out of bounds
+ val len = rnd.nextInt(a.length + 2) - 1
+ outcome(stream.readNBytes(a, off, len)) should
===(outcome(reference.readNBytes(b, off, len)))
+ a should ===(b)
+ case 6 =>
+ val n = rnd.nextInt(12).toLong
+ stream.skip(n) should ===(reference.skip(n))
+ case 7 =>
+ val n = rnd.nextInt(6).toLong
+ reference.skip(-n) should ===(0L)
+ if (n == 0) stream.skip(0) should ===(0L)
+ else an[IllegalArgumentException] should be thrownBy
stream.skip(-n) // documented deviation
+ case 8 =>
+ val n = rnd.nextInt(12).toLong - 1 // may be negative
+ outcome(stream.skipNBytes(n)) should
===(outcome(reference.skipNBytes(n)))
+ case 9 =>
+ val a = new ByteArrayOutputStream()
+ val b = new ByteArrayOutputStream()
+ stream.transferTo(a) should ===(reference.transferTo(b))
+ a.toByteArray should ===(b.toByteArray)
+ case 10 =>
+ stream.available() should ===(reference.available())
+ case 11 =>
+ stream.mark(rnd.nextInt(4))
+ reference.mark(rnd.nextInt(4))
+ case 12 =>
+ stream.reset()
+ reference.reset()
+ }
+ stream.available() should ===(reference.available())
+ }
+ }
+ stream.close()
+ reference.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 943cf3ddfb..86f0a77b9d 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 @@ import org.apache.pekko.annotation.InternalApi;
* Internal API: An unsynchronized byte array input stream. This class does
not copy the provided
* byte array, and it is not thread safe.
*
+ * <p>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 @@ public class UnsynchronizedByteArrayInputStream extends
InputStream {
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, %<s + %s) out of bounds for length %s".formatted(off,
len, arrayLength));
- }
- }
-
/** The underlying data buffer. */
private final byte[] data;
/**
* End Of Data.
*
- * <p>Similar to data.length, which is the last readable offset + 1.
+ * <p>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 class UnsynchronizedByteArrayInputStream extends
InputStream {
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 class UnsynchronizedByteArrayInputStream extends
InputStream {
@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 class UnsynchronizedByteArrayInputStream extends
InputStream {
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 e6b845226e..fb6edbe019 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
