diff --git a/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringSpec.scala index 6cb0118ab3..271d982c0e 100644 --- a/actor-tests/src/test/scala/org/apache/pekko/util/ByteStringSpec.scala +++ b/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) diff --git a/actor/src/main/scala/org/apache/pekko/util/ByteString.scala b/actor/src/main/scala/org/apache/pekko/util/ByteString.scala index 2b672aa7bb..b948a84706 100644 --- a/actor/src/main/scala/org/apache/pekko/util/ByteString.scala +++ b/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) diff --git a/bench-jmh/src/main/scala/org/apache/pekko/util/ByteString_byteAtUnchecked_Benchmark.scala b/bench-jmh/src/main/scala/org/apache/pekko/util/ByteString_byteAtUnchecked_Benchmark.scala index 95e2e89325..cbb787552c 100644 --- a/bench-jmh/src/main/scala/org/apache/pekko/util/ByteString_byteAtUnchecked_Benchmark.scala +++ b/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] = {