perf: resume the ByteStrings fragment hint backward as well as forward

actor-tests/src/test/scala/org/apache/pekko/util/ByteStringSpec.scala

@@ -2088,6 +2088,22 @@ class ByteStringSpec extends AnyWordSpec with Matchers with Checkers {
       rope.iterator.toArray should ===(expected)
     }
 
+    "return the right byte when stepping backward onto fragment boundaries" in {
+      // the backward resume recomputes fragment starts by subtracting lengths, so land exactly
+      // on the first and last byte of every fragment, from a hint planted at the far end
+      val freshRope = fragmentLengths
+        .foldLeft((ByteString.empty, 0)) { case ((acc, offset), len) =>
+          (acc ++ ByteString(expected.slice(offset, offset + len)), offset + len)
+        }
+        ._1
+      val starts = fragmentLengths.scanLeft(0)(_ + _).init
+      val lasts = fragmentLengths.scanLeft(0)(_ + _).tail.map(_ - 1)
+      freshRope(expected.length - 1) should ===(expected(expected.length - 1))
+      for (s <- starts.reverse) withClue(s"fragment start $s: ")(freshRope(s) should ===(expected(s)))
+      freshRope(expected.length - 1) should ===(expected(expected.length - 1))
+      for (e <- lasts.reverse) withClue(s"fragment last byte $e: ")(freshRope(e) should ===(expected(e)))
+    }
+
     "still reject out of range indices" in {
       an[IndexOutOfBoundsException] should be thrownBy rope(-1)
       an[IndexOutOfBoundsException] should be thrownBy rope(expected.length)

actor/src/main/scala/org/apache/pekko/util/ByteString.scala

@@ -1540,8 +1540,8 @@ object ByteString {
       else throw new IndexOutOfBoundsException(idx.toString)
 
     // Remembers the fragment resolved by the last byteAtUnchecked call, so sequential access --
-    // the dominant pattern -- stays on the same fragment or steps to the next one instead of
-    // rescanning the fragment vector from index 0 for every byte.
+    // the dominant pattern -- stays on the same fragment or steps to the neighbouring one, in
+    // either direction, instead of rescanning the fragment vector from index 0 for every byte.
     //
     // Packed into a single long so the triple is read and written atomically: the fragment index
     // in the high 32 bits and its start offset in the low 32. A reader can therefore never pair
@@ -1581,12 +1581,27 @@ object ByteString {
     /**
      * Scans for the fragment containing `offset` and records it as the hint. `hintIdx` and
      * `hintStart` are a previously read hint that missed, used to resume the scan from that
-     * fragment rather than from the start.
+     * fragment -- forward or backward, whichever side of it `offset` is on -- rather than
+     * from the start.
      */
     private def resolveFragment(offset: Int, hintIdx: Int, hintStart: Int): Long = {
       var pos = 0
       var seen = 0
       if (hintIdx >= 0) {
+        if (offset < hintStart) {
+          // moving backward before the remembered fragment: walk back from it. `offset >= 0`
+          // and fragment 0 starts at 0, so the walk stops at fragment 0 at the latest, and it
+          // is never longer than the scan from fragment 0 it replaces.
+          pos = hintIdx
+          seen = hintStart
+          while (offset < seen) {
+            pos -= 1
+            seen -= bytestrings(pos).length
+          }
+          val located = (pos.toLong << 32) | (seen.toLong & 0xFFFFFFFFL)
+          fragmentHint = located
+          return located
+        }
         val hintEnd = hintStart + bytestrings(hintIdx).length
         if (offset >= hintEnd && hintIdx + 1 < bytestrings.length) {
           // moving forward past the remembered fragment: resume the scan from it

bench-jmh/src/main/scala/org/apache/pekko/util/ByteString_byteAtUnchecked_Benchmark.scala

@@ -64,8 +64,20 @@ class ByteString_byteAtUnchecked_Benchmark {
   manyFragments_sequential  thrpt    3   54838.759 ±   29275.476  ops/s
 
   Sequential access is roughly 84x faster. Random access is unchanged (the hint never hits) and
-  reverse access does not benefit either, since each step lands before the remembered fragment
-  and falls back to a scan from the start -- both stay within the noise of the previous numbers.
+  reverse access did not benefit at that point, since each step landed before the remembered
+  fragment and fell back to a scan from the start -- both stayed within the noise of the
+  previous numbers.
+
+  After resolveFragment also resumes backward from the remembered fragment (same short run,
+  same wide error bars):
+
+  manyFragments_reverse     thrpt    3    386.052 ±  550.638  ops/s   (before, on this machine)
+  manyFragments_reverse     thrpt    3  39969.362 ± 49806.718 ops/s   (after)
+  manyFragments_sequential  thrpt    3  43415.035 ± 45223.244 ops/s   (before, on this machine)
+  manyFragments_sequential  thrpt    3  44624.286 ± 42770.609 ops/s   (after)
+
+  Reverse access is roughly 100x faster and on par with sequential; sequential is unchanged
+  within the noise.
    */
 
   private val randomIndices: Array[Int] = {