perf: optimise UnsynchronizedByteArrayInputStream bulk operations

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()
     }
   }

actor/src/main/java/org/apache/pekko/io/UnsynchronizedByteArrayInputStream.java

@@ -18,53 +18,49 @@
 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;
 
 /**
  * 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 {
 
   /** 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");
     }
     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 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;
   }
 }

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