From 48eed6f62d177dcddc46ad9513a8fedb5c4a2d17 Mon Sep 17 00:00:00 2001 From: Perfloop Agent Date: Wed, 8 Jul 2026 20:37:32 +0000 Subject: [PATCH] Optimize RangeTombstoneList copying with copy-on-write array sharing (CASSANDRA-21492) --- CHANGES.txt | 1 + .../cassandra/db/RangeTombstoneList.java | 87 ++++++++++++------- .../microbench/RangeTombstoneListBench.java | 87 +++++++++++++++++++ .../RangeTombstoneListSize10Bench.java | 83 ++++++++++++++++++ .../cassandra/db/RangeTombstoneListTest.java | 67 ++++++++++++++ 5 files changed, 296 insertions(+), 29 deletions(-) create mode 100644 test/microbench/org/apache/cassandra/test/microbench/RangeTombstoneListBench.java create mode 100644 test/microbench/org/apache/cassandra/test/microbench/RangeTombstoneListSize10Bench.java diff --git a/CHANGES.txt b/CHANGES.txt index 04ab37acf0bf..699da685135a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,5 +1,6 @@ 7.0 * Implementation of CEP-49: Hardware-accelerated compression (CASSANDRA-20975) + * Optimize RangeTombstoneList copying with copy-on-write array sharing (CASSANDRA-21492) * Avoid using ObjectUtils.getFirstNonNull in Schema (CASSANDRA-21394) * Allow nodetool garbagecollect to take a user defined list of SSTables (CASSANDRA-16767) * Add a guardrail for misprepared statements (CASSANDRA-21139) diff --git a/src/java/org/apache/cassandra/db/RangeTombstoneList.java b/src/java/org/apache/cassandra/db/RangeTombstoneList.java index cf85d728cfb3..a3f7868615f3 100644 --- a/src/java/org/apache/cassandra/db/RangeTombstoneList.java +++ b/src/java/org/apache/cassandra/db/RangeTombstoneList.java @@ -48,6 +48,13 @@ *

* The only use of the local deletion time is to know when a given tombstone can * be purged, which will be done by the purge() method. + *

+ * Copy-on-Write (COW) Invariant: + * This class implements a copy-on-write optimization. The {@link #copy()} method creates a new instance + * that shares the four parallel backing arrays ({@code starts}, {@code ends}, {@code markedAts}, + * and {@code delTimesUnsignedIntegers}) with the original instance and marks both instances as shared ({@code shared = true}). + * To ensure safe copy independence, any operation that performs in-place mutation of these backing arrays + * MUST call {@link #isolate()} before writing to them for the first time. All backing-array writes are guarded this way. */ public class RangeTombstoneList implements Iterable, IMeasurableMemory { @@ -64,6 +71,19 @@ public class RangeTombstoneList implements Iterable, IMeasurable private long boundaryHeapSize; private int size; + private boolean shared; + + private void isolate() + { + if (shared) + { + starts = Arrays.copyOf(starts, starts.length); + ends = Arrays.copyOf(ends, ends.length); + markedAts = Arrays.copyOf(markedAts, markedAts.length); + delTimesUnsignedIntegers = Arrays.copyOf(delTimesUnsignedIntegers, delTimesUnsignedIntegers.length); + shared = false; + } + } private RangeTombstoneList(ClusteringComparator comparator, ClusteringBound[] starts, @@ -103,14 +123,28 @@ public ClusteringComparator comparator() return comparator; } + /** + * Creates a copy-on-write duplicate of this list. + *

+ * This method shares the four backing arrays ({@code starts}, {@code ends}, {@code markedAts}, and + * {@code delTimesUnsignedIntegers}) between the original list and the returned copy, marking both instances + * as shared ({@code shared = true}). To maintain independent states, any subsequent in-place mutation on + * either instance must first trigger {@link #isolate()} to copy and isolate the backing arrays before writing. + * + * @return a copy-on-write copy of this list. + */ public RangeTombstoneList copy() { - return new RangeTombstoneList(comparator, - Arrays.copyOf(starts, size), - Arrays.copyOf(ends, size), - Arrays.copyOf(markedAts, size), - Arrays.copyOf(delTimesUnsignedIntegers, size), - boundaryHeapSize, size); + this.shared = true; + RangeTombstoneList copy = new RangeTombstoneList(comparator, + starts, + ends, + markedAts, + delTimesUnsignedIntegers, + boundaryHeapSize, + size); + copy.shared = true; + return copy; } public RangeTombstoneList clone(ByteBufferCloner cloner) @@ -188,7 +222,14 @@ public void addAll(RangeTombstoneList tombstones) if (isEmpty()) { - copyArrays(tombstones, this); + tombstones.shared = true; + this.starts = tombstones.starts; + this.ends = tombstones.ends; + this.markedAts = tombstones.markedAts; + this.delTimesUnsignedIntegers = tombstones.delTimesUnsignedIntegers; + this.size = tombstones.size; + this.boundaryHeapSize = tombstones.boundaryHeapSize; + this.shared = true; return; } @@ -324,12 +365,14 @@ public void collectStats(EncodingStats.Collector collector) public void updateAllTimestamp(long timestamp) { + isolate(); for (int i = 0; i < size; i++) markedAts[i] = timestamp; } public void updateAllTimestampAndLocalDeletionTime(long timestamp, long localDeletionTime) { + isolate(); int unsignedLocalDeletionTime = Cell.deletionTimeLongToUnsignedInteger(localDeletionTime); for (int i = 0; i < size; i++) { @@ -527,17 +570,6 @@ public final int hashCode() return result; } - private static void copyArrays(RangeTombstoneList src, RangeTombstoneList dst) - { - dst.grow(src.size); - System.arraycopy(src.starts, 0, dst.starts, 0, src.size); - System.arraycopy(src.ends, 0, dst.ends, 0, src.size); - System.arraycopy(src.markedAts, 0, dst.markedAts, 0, src.size); - System.arraycopy(src.delTimesUnsignedIntegers, 0, dst.delTimesUnsignedIntegers, 0, src.size); - dst.size = src.size; - dst.boundaryHeapSize = src.boundaryHeapSize; - } - /* * Inserts a new element starting at index i. This method assumes that: * ends[i-1] <= start < ends[i] @@ -552,6 +584,7 @@ private static void copyArrays(RangeTombstoneList src, RangeTombstoneList dst) */ private void insertFrom(int i, ClusteringBound start, ClusteringBound end, long markedAt, int delTimeUnsignedInternal) { + isolate(); while (i < size) { assert start.isStart() && end.isEnd(); @@ -677,8 +710,12 @@ private void addInternal(int i, ClusteringBound start, ClusteringBound end if (size == capacity()) growToFree(i); - else if (i < size) - moveElements(i); + else + { + isolate(); + if (i < size) + moveElements(i); + } setInternal(i, start, end, markedAt, delTimeUnsignedInteger); size++; @@ -698,21 +735,13 @@ private void growToFree(int i) grow(i, newLength); } - /* - * Grow the arrays to match newLength capacity. - */ - private void grow(int newLength) - { - if (capacity() < newLength) - grow(-1, newLength); - } - private void grow(int i, int newLength) { starts = grow(starts, size, newLength, i); ends = grow(ends, size, newLength, i); markedAts = grow(markedAts, size, newLength, i); delTimesUnsignedIntegers = grow(delTimesUnsignedIntegers, size, newLength, i); + shared = false; } private static ClusteringBound[] grow(ClusteringBound[] a, int size, int newLength, int i) diff --git a/test/microbench/org/apache/cassandra/test/microbench/RangeTombstoneListBench.java b/test/microbench/org/apache/cassandra/test/microbench/RangeTombstoneListBench.java new file mode 100644 index 000000000000..3e25188d09c6 --- /dev/null +++ b/test/microbench/org/apache/cassandra/test/microbench/RangeTombstoneListBench.java @@ -0,0 +1,87 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.test.microbench; + +import java.util.concurrent.TimeUnit; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Param; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ClusteringComparator; +import org.apache.cassandra.db.DeletionTime; +import org.apache.cassandra.db.RangeTombstone; +import org.apache.cassandra.db.RangeTombstoneList; +import org.apache.cassandra.db.Slice; +import org.apache.cassandra.db.marshal.Int32Type; + +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 3, time = 1, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) +@Fork(value = 1, jvmArgsAppend = "-Xmx512M") +@Threads(1) +@State(Scope.Benchmark) +public class RangeTombstoneListBench +{ + private ClusteringComparator comparator; + private RangeTombstoneList existing; + private RangeTombstone tombstone; + + @Param({"10", "100", "1000"}) + private int size; + + @Setup(Level.Trial) + public void setup() + { + DatabaseDescriptor.daemonInitialization(); + comparator = new ClusteringComparator(Int32Type.instance); + existing = new RangeTombstoneList(comparator, size); + for (int i = 0; i < size; i++) + { + existing.add(new RangeTombstone(Slice.make(comparator.make(i * 2), comparator.make(i * 2 + 1)), DeletionTime.build(1, 1))); + } + tombstone = new RangeTombstone(Slice.make(comparator.make(100000), comparator.make(100001)), DeletionTime.build(1, 1)); + } + + @Benchmark + public RangeTombstoneList benchCopyOnly() + { + return existing.copy(); + } + + @Benchmark + public RangeTombstoneList benchCopyAndAdd() + { + RangeTombstoneList copy = existing.copy(); + copy.add(tombstone); + return copy; + } +} diff --git a/test/microbench/org/apache/cassandra/test/microbench/RangeTombstoneListSize10Bench.java b/test/microbench/org/apache/cassandra/test/microbench/RangeTombstoneListSize10Bench.java new file mode 100644 index 000000000000..b5f80e7a26e2 --- /dev/null +++ b/test/microbench/org/apache/cassandra/test/microbench/RangeTombstoneListSize10Bench.java @@ -0,0 +1,83 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.test.microbench; + +import java.util.concurrent.TimeUnit; + +import org.openjdk.jmh.annotations.Benchmark; +import org.openjdk.jmh.annotations.BenchmarkMode; +import org.openjdk.jmh.annotations.Fork; +import org.openjdk.jmh.annotations.Level; +import org.openjdk.jmh.annotations.Measurement; +import org.openjdk.jmh.annotations.Mode; +import org.openjdk.jmh.annotations.OutputTimeUnit; +import org.openjdk.jmh.annotations.Scope; +import org.openjdk.jmh.annotations.Setup; +import org.openjdk.jmh.annotations.State; +import org.openjdk.jmh.annotations.Threads; +import org.openjdk.jmh.annotations.Warmup; + +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ClusteringComparator; +import org.apache.cassandra.db.DeletionTime; +import org.apache.cassandra.db.RangeTombstone; +import org.apache.cassandra.db.RangeTombstoneList; +import org.apache.cassandra.db.Slice; +import org.apache.cassandra.db.marshal.Int32Type; + +@BenchmarkMode(Mode.AverageTime) +@OutputTimeUnit(TimeUnit.NANOSECONDS) +@Warmup(iterations = 3, time = 1, timeUnit = TimeUnit.SECONDS) +@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS) +@Fork(value = 1, jvmArgsAppend = "-Xmx512M") +@Threads(1) +@State(Scope.Benchmark) +public class RangeTombstoneListSize10Bench +{ + private ClusteringComparator comparator; + private RangeTombstoneList existing; + private RangeTombstone tombstone; + + @Setup(Level.Trial) + public void setup() + { + DatabaseDescriptor.daemonInitialization(); + comparator = new ClusteringComparator(Int32Type.instance); + existing = new RangeTombstoneList(comparator, 10); + for (int i = 0; i < 10; i++) + { + existing.add(new RangeTombstone(Slice.make(comparator.make(i * 2), comparator.make(i * 2 + 1)), DeletionTime.build(1, 1))); + } + tombstone = new RangeTombstone(Slice.make(comparator.make(100000), comparator.make(100001)), DeletionTime.build(1, 1)); + } + + @Benchmark + public RangeTombstoneList benchCopyOnly() + { + return existing.copy(); + } + + @Benchmark + public RangeTombstoneList benchCopyAndAdd() + { + RangeTombstoneList copy = existing.copy(); + copy.add(tombstone); + return copy; + } +} diff --git a/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java b/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java index e9f28c0591c2..4381c257e6bc 100644 --- a/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java +++ b/test/unit/org/apache/cassandra/db/RangeTombstoneListTest.java @@ -629,6 +629,73 @@ public void testSetInitialAllocationSize() } } + @Test + public void testCopyIndependence() + { + RangeTombstoneList original = new RangeTombstoneList(cmp, 5); + original.add(rt(1, 5, 10)); + original.add(rt(7, 10, 20)); + + RangeTombstoneList copy = original.copy(); + + original.add(rt(12, 15, 30)); + + assertEquals(3, original.size()); + assertEquals(2, copy.size()); + + Iterator origIter = original.iterator(); + assertRT(rt(1, 5, 10), origIter.next()); + assertRT(rt(7, 10, 20), origIter.next()); + assertRT(rt(12, 15, 30), origIter.next()); + assertFalse(origIter.hasNext()); + + Iterator copyIter = copy.iterator(); + assertRT(rt(1, 5, 10), copyIter.next()); + assertRT(rt(7, 10, 20), copyIter.next()); + assertFalse(copyIter.hasNext()); + + copy.add(rt(20, 25, 40)); + assertEquals(3, original.size()); + assertEquals(3, copy.size()); + + origIter = original.iterator(); + assertRT(rt(1, 5, 10), origIter.next()); + assertRT(rt(7, 10, 20), origIter.next()); + assertRT(rt(12, 15, 30), origIter.next()); + assertFalse(origIter.hasNext()); + + copyIter = copy.iterator(); + assertRT(rt(1, 5, 10), copyIter.next()); + assertRT(rt(7, 10, 20), copyIter.next()); + assertRT(rt(20, 25, 40), copyIter.next()); + assertFalse(copyIter.hasNext()); + + RangeTombstoneList other = new RangeTombstoneList(cmp, 1); + other.add(rt(30, 35, 50)); + copy.addAll(other); + assertEquals(3, original.size()); + assertEquals(4, copy.size()); + + copy.updateAllTimestamp(999); + origIter = original.iterator(); + assertRT(rt(1, 5, 10), origIter.next()); + assertRT(rt(7, 10, 20), origIter.next()); + assertRT(rt(12, 15, 30), origIter.next()); + + copyIter = copy.iterator(); + assertRT(rt(1, 5, 999), copyIter.next()); + assertRT(rt(7, 10, 999), copyIter.next()); + assertRT(rt(20, 25, 999), copyIter.next()); + assertRT(rt(30, 35, 999), copyIter.next()); + + copy.updateAllTimestampAndLocalDeletionTime(888, 888); + origIter = original.iterator(); + assertRT(rt(1, 5, 10), origIter.next()); + + copyIter = copy.iterator(); + assertRT(rt(1, 5, 888, 888), copyIter.next()); + } + private void assertHasException(ThrowingRunnable block, Consumer verifier) { try