* 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
+ * 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