Skip to content

perf: optimise UnsynchronizedByteArrayInputStream bulk operations #3541

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pjfanning wants to merge 2 commits into
base: main
Choose a base branch
from
Open

perf: optimise UnsynchronizedByteArrayInputStream bulk operations #3541

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2bfd685
perf: optimise UnsynchronizedByteArrayInputStream bulk operations
pjfanning Sep 13, 2026
7339b14
test: add differential test against ByteArrayInputStream
pjfanning Sep 22, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
209 changes: 206 additions & 3 deletions actor-tests/src/test/scala/org/apache/pekko/io/UnsynchronizedByteArrayInputStreamSpec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,18 +17,29 @@

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

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)

@He-Pin He-Pin Sep 22, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Consider adding a differential test that runs randomized operation sequences against both this stream and a ByteArrayInputStream, asserting identical outputs except the two documented deviations (zero-length read at EOF returns 0, not -1; negative skip throws IllegalArgumentException). It is the strongest guard for a serialization-path class.

@pjfanning pjfanning Sep 22, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added a randomized differential test against java.io.ByteArrayInputStream in 7339b14 (rebased on main as well). It compares return values, exceptions, buffer contents and available() after every operation, special-casing only the two documented deviations.


/** 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
Expand All @@ -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()
}
}
}
}
Loading
Loading