Optimize ByteString.grouped(size) (#25153)

* Add ByteString.grouped() benchmark & tests.

* Implement ByteString.grouped().

* PR comments.

akka-actor-tests/src/test/scala/akka/util/ByteStringSpec.scala

@@ -52,6 +52,15 @@ class ByteStringSpec extends WordSpec with Matchers with Checkers {
     } yield (xs, from, until)
   }
 
+  case class ByteStringGrouped(bs: ByteString, size: Int)
+
+  implicit val arbitraryByteStringGrouped = Arbitrary {
+    for {
+      xs ← arbitraryByteString.arbitrary
+      size ← Gen.choose(1, 1 max xs.length)
+    } yield ByteStringGrouped(xs, size)
+  }
+
   type ArraySlice[A] = (Array[A], Int, Int)
 
   def arbSlice[A](arbArray: Arbitrary[Array[A]]): Arbitrary[ArraySlice[A]] = Arbitrary {
@@ -730,6 +739,14 @@ class ByteStringSpec extends WordSpec with Matchers with Checkers {
         }
       }
 
+      "calling grouped" in {
+        check { grouped: ByteStringGrouped ⇒
+          likeVector(grouped.bs) {
+            _.grouped(grouped.size).toIndexedSeq
+          }
+        }
+      }
+
       "calling copyToArray" in {
         check { slice: ByteStringSlice ⇒
           slice match {

akka-actor/src/main/scala/akka/util/ByteString.scala

@@ -701,6 +701,16 @@ sealed abstract class ByteString extends IndexedSeq[Byte] with IndexedSeqOptimiz
   // optimized in subclasses
   override def indexOf[B >: Byte](elem: B): Int = indexOf(elem, 0)
 
+  override def grouped(size: Int): Iterator[ByteString] = {
+    if (size <= 0) {
+      throw new IllegalArgumentException(s"size=$size must be positive")
+    }
+
+    Iterator.iterate(this)(_.drop(size))
+      .takeWhile(_.nonEmpty)
+      .map(_.take(size))
+  }
+
   override def toString(): String = {
     val maxSize = 100
     if (size > maxSize)

akka-bench-jmh/src/main/scala/akka/util/ByteString_grouped_Benchmark.scala

@@ -0,0 +1,31 @@
+/**
+ * Copyright (C) 2014-2018 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.util
+
+import java.util.concurrent.TimeUnit
+
+import org.openjdk.jmh.annotations._
+import org.openjdk.jmh.infra.Blackhole
+
+@State(Scope.Benchmark)
+@BenchmarkMode(Array(Mode.AverageTime))
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 10, time = 1, timeUnit = TimeUnit.SECONDS)
+class ByteString_grouped_Benchmark {
+
+  private val bsLarge = ByteString(Array.ofDim[Byte](1000 * 1000))
+
+  /*
+    > akka-bench-jmh/jmh:run -f1 .*ByteString_grouped_Benchmark
+    [info] Benchmark                             Mode  Cnt      Score      Error  Units
+    [info] ByteString_grouped_Benchmark.grouped  avgt   10  59386.328 ± 1466.045  ns/op
+   */
+
+  @Benchmark
+  def grouped(bh: Blackhole): Unit = {
+    bh.consume(bsLarge.grouped(1000).foreach(bh.consume))
+  }
+}