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,20 +1540,25 @@ 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
-    // the index of one fragment with the start of another, which separate int fields would allow.
-    // The end offset is not stored; it is recomputed as start + fragment.length, which is a
-    // cheap array read. The field is deliberately not volatile: it is only a hint, so a reader
+    // Packed into a single long -- the fragment index in the high 32 bits, its start offset in the
+    // low 32 -- and volatile so that the two always come from the same write. JLS 17.7 lets a
+    // non-volatile 64-bit field be read as two 32-bit halves, which would pair the index of one
+    // fragment with the start of another; the fast path below tests that pair against the wrong
+    // fragment's length and can pass, returning a byte from the wrong fragment.
+    //
+    // Volatile is for that atomicity, not for ordering: the value remains only a hint, so a reader
     // that misses another thread's update simply rescans, and ByteString is immutable, so a
-    // resolved mapping never becomes wrong.
-    private[this] var fragmentHint: Long = ByteStrings.NoHint
+    // resolved mapping never becomes wrong. Reads are the hot path; the write happens only when a
+    // lookup misses, which is once per fragment crossed rather than once per byte.
+    //
+    // The end offset is not stored; it is recomputed as start + fragment.length, a cheap array read.
+    @volatile private[this] var fragmentHint: Long = ByteStrings.NoHint
 
     private[pekko] override def byteAtUnchecked(offset: Int): Byte = {
-      val hint = fragmentHint // single read: index and start below are mutually consistent
+      val hint = fragmentHint // single read: index and start below come from the same write
       val hintIdx = (hint >>> 32).toInt
       val hintStart = hint.toInt
       if (hintIdx >= 0) {
@@ -1571,7 +1576,7 @@ object ByteString {
      * the hint uses. `offset` must be within this ByteString.
      */
     private def locateFragment(offset: Int): Long = {
-      val hint = fragmentHint // single read: index and start below are mutually consistent
+      val hint = fragmentHint // single read: index and start below come from the same write
       val hintIdx = (hint >>> 32).toInt
       val hintStart = hint.toInt
       if (hintIdx >= 0 && offset >= hintStart && offset - hintStart < bytestrings(hintIdx).length) hint
@@ -1581,17 +1586,38 @@ 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) {
-        val hintEnd = hintStart + bytestrings(hintIdx).length
-        if (offset >= hintEnd && hintIdx + 1 < bytestrings.length) {
-          // moving forward past the remembered fragment: resume the scan from it
-          pos = hintIdx + 1
-          seen = hintEnd
+        if (offset < hintStart) {
+          // Moving backward before the remembered fragment: walk back from it, keeping `seen` at the
+          // start of fragment `pos`. The hint pair is atomic (see above), so `seen` is a real
+          // fragment start; fragment 0 starts at 0 and `offset >= 0`, so the walk stops at fragment
+          // 0 at the latest. At most `hintIdx` steps -- the same O(fragments) bound per call as the
+          // from-zero scan it replaces, but not always fewer steps than it, since a far-end hint
+          // with a near-zero offset walks the whole way back. That case is random access, which
+          // neither strategy serves; backward sequential access, the case this is for, costs one
+          // step. Hence no distance heuristic.
+          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
+        } else {
+          val hintEnd = hintStart + bytestrings(hintIdx).length
+          if (offset >= hintEnd && hintIdx + 1 < bytestrings.length) {
+            // moving forward past the remembered fragment: resume the scan from it
+            pos = hintIdx + 1
+            seen = hintEnd
+          }
         }
       }
       var frag = bytestrings(pos)

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

@@ -64,8 +64,37 @@ 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.
+
+  Making the hint volatile, so that its index and start cannot be read as two halves of
+  different writes (JLS 17.7), costs nothing measurable outside the one-byte-fragment
+  sequential case. Paired run toggling only the `@volatile` keyword, -f2 -wi 5 -i 5, with
+  manyFragments_map -- which walks the fragments directly and never consults the hint -- as
+  an in-run control:
+
+                            plain long                    volatile
+  manyFragments_map          143209.374 ± 8645.210         139763.871 ± 9327.740  ops/s
+  manyFragments_random          911.739 ±   76.206            901.787 ±  154.965  ops/s
+  manyFragments_reverse       43961.948 ± 9780.313          42858.330 ±  949.351  ops/s
+  manyFragments_sequential    48989.773 ± 6972.720          41100.395 ± 3682.974  ops/s
+
+  Random and reverse move by about as much as the control (~2%); only sequential shows a
+  possible cost, and its intervals barely separate. These 1024 one-byte fragments are the
+  worst case for it, since the hint is then written on every single access -- at realistic
+  fragment sizes the write is amortised over the whole fragment.
    */
 
   private val randomIndices: Array[Int] = {