diff --git a/modules/accord b/modules/accord index 25f23ffec4..a63cac24a2 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 25f23ffec439a921387ca249908798b9cc7d4620 +Subproject commit a63cac24a2198a5893874cdf72946073854a8d4d diff --git a/src/java/org/apache/cassandra/dht/AccordSplitter.java b/src/java/org/apache/cassandra/dht/AccordSplitter.java index 467ac2a105..916e479679 100644 --- a/src/java/org/apache/cassandra/dht/AccordSplitter.java +++ b/src/java/org/apache/cassandra/dht/AccordSplitter.java @@ -21,6 +21,7 @@ package org.apache.cassandra.dht; import java.math.BigInteger; import accord.local.ShardDistributor; +import accord.primitives.Range; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.accord.TokenRange; import org.apache.cassandra.service.accord.api.AccordRoutingKey; @@ -60,6 +61,12 @@ public abstract class AccordSplitter implements ShardDistributor.EvenSplit.Split endOffset.compareTo(sizeOfRange) >= 0 ? endBound : new TokenKey(tableId, tokenForValue(start.add(endOffset)))); } + @Override + public boolean splittable(Range range, int numSplits) + { + return sizeOf(range).compareTo(BigInteger.valueOf(numSplits)) >= 0; + } + @Override public BigInteger zero() { diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 3157fb8355..c46c953a61 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -89,6 +89,7 @@ import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; import static accord.api.ConfigurationService.EpochReady.DONE; import static accord.local.KeyHistory.COMMANDS; +import static accord.primitives.SaveStatus.Applying; import static accord.primitives.Status.Committed; import static accord.primitives.Status.Invalidated; import static accord.primitives.Status.Truncated; @@ -102,6 +103,7 @@ public class AccordCommandStore extends CommandStore public final String loggingId; private final IJournal journal; + private final CommandStoreExecutor executor; private final AccordStateCache.Instance commandCache; private final AccordStateCache.Instance timestampsForKeyCache; @@ -295,6 +297,13 @@ public class AccordCommandStore extends CommandStore return commandsForKeyCache; } + @VisibleForTesting + @Override + protected void unsafeSetRangesForEpoch(CommandStores.RangesForEpoch newRangesForEpoch) + { + super.unsafeSetRangesForEpoch(newRangesForEpoch); + } + @Nullable @VisibleForTesting public Runnable appendToKeyspace(Command after) @@ -670,7 +679,11 @@ public class AccordCommandStore extends CommandStore safeStore -> { SafeCommand safeCommand = safeStore.unsafeGet(txnId); Command local = safeCommand.current(); - Commands.maybeExecute(safeStore, safeCommand, local, true, true); + if (local.hasBeen(Truncated)) + return; + + if (local.saveStatus().compareTo(Applying) >= 0) Commands.applyWrites(safeStore, context, local).begin(agent); + else Commands.maybeExecute(safeStore, safeCommand, local, true, true); }) .begin((unused, throwable) -> { if (throwable != null) diff --git a/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java b/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java index 1bf81b6048..55f4d96406 100644 --- a/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java +++ b/src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java @@ -32,6 +32,7 @@ import com.google.common.collect.ImmutableList; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import accord.utils.Invariants; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.tcm.log.Entry; import org.apache.cassandra.tcm.log.LocalLog; @@ -177,7 +178,7 @@ public class AtomicLongBackedProcessor extends AbstractLocalProcessor public LogState getLogState(Epoch start, Epoch end) { EntryHolder state = getEntries(Epoch.EMPTY); - ClusterMetadata metadata = new ClusterMetadata(DatabaseDescriptor.getPartitioner());; + ClusterMetadata metadata = new ClusterMetadata(DatabaseDescriptor.getPartitioner()); Iterator iter = state.iterator(); ImmutableList.Builder rest = new ImmutableList.Builder<>(); while (iter.hasNext()) @@ -186,8 +187,11 @@ public class AtomicLongBackedProcessor extends AbstractLocalProcessor if (current.epoch.isAfter(end)) break; if (current.epoch.isEqualOrBefore(start)) + { + Invariants.checkState(current.epoch.isDirectlyAfter(metadata.epoch)); metadata = current.transform.execute(metadata).success().metadata; - else + } + else if (current.epoch.isAfter(start)) rest.add(current); } diff --git a/src/java/org/apache/cassandra/tcm/log/LogReader.java b/src/java/org/apache/cassandra/tcm/log/LogReader.java index effc4d7561..7f2b80602a 100644 --- a/src/java/org/apache/cassandra/tcm/log/LogReader.java +++ b/src/java/org/apache/cassandra/tcm/log/LogReader.java @@ -28,6 +28,7 @@ import java.util.TreeSet; import com.google.common.collect.ImmutableList; import com.google.common.collect.Ordering; +import accord.utils.Invariants; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.Epoch; @@ -124,12 +125,15 @@ public interface LogReader { try { - ClusterMetadata closestSnapshot = snapshots().getSnapshotBefore(start); + ClusterMetadata closestSnapshot = null; + if (includeSnapshot) + closestSnapshot = snapshots().getSnapshotBefore(start); // Snapshot could not be found, fetch enough epochs to reconstruct the start metadata if (closestSnapshot == null) { - closestSnapshot = new ClusterMetadata(DatabaseDescriptor.getPartitioner()); + if (includeSnapshot) + closestSnapshot = new ClusterMetadata(DatabaseDescriptor.getPartitioner()); ImmutableList.Builder entries = new ImmutableList.Builder<>(); EntryHolder entryHolder = getEntries(Epoch.EMPTY, end); for (Entry entry : entryHolder.entries) @@ -144,20 +148,21 @@ public interface LogReader else if (closestSnapshot.epoch.isBefore(start)) { ImmutableList.Builder entries = new ImmutableList.Builder<>(); - EntryHolder entryHolder = getEntries(closestSnapshot.epoch, end); + EntryHolder entryHolder = getEntries(closestSnapshot.epoch.nextEpoch(), end); for (Entry entry : entryHolder.entries) { if (entry.epoch.isAfter(start)) entries.add(entry); - else + else if (includeSnapshot) closestSnapshot = entry.transform.execute(closestSnapshot).success().metadata; } return new LogState(closestSnapshot, entries.build()); } else { - assert closestSnapshot.epoch.isEqualOrAfter(start) : String.format("Got %s, but requested snapshot of %s", closestSnapshot.epoch, start); - EntryHolder entryHolder = getEntries(closestSnapshot.epoch.nextEpoch(), end); + Invariants.checkState(closestSnapshot.epoch.isEqualOrAfter(start), + "Got %s, but requested snapshot of %s", closestSnapshot.epoch, start); + EntryHolder entryHolder = getEntries(closestSnapshot.epoch, end); return new LogState(closestSnapshot, ImmutableList.copyOf(entryHolder.entries)); } } diff --git a/src/java/org/apache/cassandra/tcm/log/LogState.java b/src/java/org/apache/cassandra/tcm/log/LogState.java index 03294e9ffb..fcb1a05a51 100644 --- a/src/java/org/apache/cassandra/tcm/log/LogState.java +++ b/src/java/org/apache/cassandra/tcm/log/LogState.java @@ -27,6 +27,7 @@ import com.google.common.collect.ImmutableList; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import accord.utils.Invariants; import org.apache.cassandra.concurrent.ScheduledExecutors; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.SystemKeyspace; @@ -72,6 +73,10 @@ public class LogState // Uses Replication rather than an just a list of entries primarily to avoid duplicating the existing serializer public LogState(ClusterMetadata baseState, ImmutableList entries) { + Invariants.checkState(baseState == null || + entries.isEmpty() || + entries.get(0).epoch.isDirectlyAfter(baseState.epoch), + "Base state: %s, first entry: %s", baseState == null ? null : baseState.epoch, entries.isEmpty() ? null : entries.get(0).epoch); this.baseState = baseState; this.entries = entries; } diff --git a/src/java/org/apache/cassandra/tcm/transformations/BeginConsensusMigrationForTableAndRange.java b/src/java/org/apache/cassandra/tcm/transformations/BeginConsensusMigrationForTableAndRange.java index 59379852cc..be7e9a763f 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/BeginConsensusMigrationForTableAndRange.java +++ b/src/java/org/apache/cassandra/tcm/transformations/BeginConsensusMigrationForTableAndRange.java @@ -31,9 +31,9 @@ import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationTarget; import org.apache.cassandra.service.consensus.migration.ConsensusTableMigration; -import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.Transformation; import org.apache.cassandra.tcm.sequences.LockedRanges; diff --git a/test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java b/test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java index 2e131bcf49..28ef23bf62 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java @@ -98,14 +98,13 @@ public class SystemKeyspaceStorageTest extends CoordinatorPathTestBase cluster.get(1).runOnInstance(() -> deleteSnapshot(toRemoveSnapshot.getEpoch())); } } - Epoch latestSnapshot = remainingSnapshots.get(remainingSnapshots.size() - 1); Epoch lastEpoch = allEpochs.stream().max(Comparator.naturalOrder()).get(); repeat(10, () -> { repeat(100, () -> { Epoch since = allEpochs.get(rng.nextInt(allEpochs.size())); - for (boolean consistentReplay : new boolean[]{ true, false }) + for (boolean consistentFetch : new boolean[]{ true, false }) { - LogState logState = simulatedCluster.node(2).requestResponse(new FetchCMSLog(since, consistentReplay)); + LogState logState = simulatedCluster.node(2).requestResponse(new FetchCMSLog(since, consistentFetch)); // if we return a snapshot it is always the most recent one // we don't return a snapshot if there is only 1 snapshot after `since` Epoch start = since; @@ -119,12 +118,16 @@ public class SystemKeyspaceStorageTest extends CoordinatorPathTestBase } else { - assertEquals(latestSnapshot, logState.baseState.epoch); + assertEquals(since, logState.baseState.epoch); start = logState.baseState.epoch; if (logState.entries.isEmpty()) // no entries, snapshot should have the same epoch as since assertEquals(since, start); else // first epoch in entries should be snapshot epoch + 1 + { + if (!start.nextEpoch().equals(logState.entries.get(0).epoch)) + System.out.println(1); assertEquals(start.nextEpoch(), logState.entries.get(0).epoch); + } } for (Entry entry : logState.entries) @@ -174,7 +177,7 @@ public class SystemKeyspaceStorageTest extends CoordinatorPathTestBase } catch (Throwable throwable) { - throw new AssertionError(throwable); + throw new AssertionError(String.format("Failed on %dth/%d repetition", i, num), throwable); } } } diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java index 7da1a51e13..f96c3b438e 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java @@ -109,8 +109,8 @@ public class SimulatedAccordCommandStore implements AutoCloseable globalExecutor = new SimulatedExecutorFactory(rs.fork(), fromQT(Generators.TIMESTAMP_GEN.map(java.sql.Timestamp::getTime)).mapToLong(TimeUnit.MILLISECONDS::toNanos).next(rs), failures::add); this.unorderedScheduled = globalExecutor.scheduled("ignored"); ExecutorFactory.Global.unsafeSet(globalExecutor); - Stage.READ.unsafeSetExecutor(unorderedScheduled); - Stage.MUTATION.unsafeSetExecutor(unorderedScheduled); + for (Stage stage : Arrays.asList(Stage.READ, Stage.MUTATION, Stage.ACCORD_RANGE_LOADER)) + stage.unsafeSetExecutor(unorderedScheduled); for (Stage stage : Arrays.asList(Stage.MISC, Stage.ACCORD_MIGRATION, Stage.READ, Stage.MUTATION)) stage.unsafeSetExecutor(globalExecutor.configureSequential("ignore").build()); @@ -216,11 +216,16 @@ public class SimulatedAccordCommandStore implements AutoCloseable this.topology = AccordTopology.createAccordTopology(ClusterMetadata.current()); this.topologies = new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, topology); var rangesForEpoch = new CommandStores.RangesForEpoch(topology.epoch(), topology.ranges(), store); + store.unsafeSetRangesForEpoch(rangesForEpoch); updateHolder.add(topology.epoch(), rangesForEpoch, topology.ranges()); updateHolder.updateGlobal(topology.ranges()); shouldEvict = boolSource(rs.fork()); - shouldFlush = boolSource(rs.fork()); + { + // tests used to take 1m but after many changes in accord they now take many minutes and its due to flush... so lower the frequency of flushing + var fork = rs.fork(); + shouldFlush = () -> fork.decide(.01); + } shouldCompact = boolSource(rs.fork()); } diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java index 74ee74a4d1..05f4dffe94 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java @@ -20,8 +20,10 @@ package org.apache.cassandra.service.accord; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.TreeSet; import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; import java.util.stream.IntStream; @@ -44,7 +46,9 @@ import accord.primitives.LatestDeps; import accord.primitives.Range; import accord.primitives.Ranges; import accord.primitives.Routable; +import accord.primitives.Routables; import accord.primitives.RoutingKeys; +import accord.primitives.Seekables; import accord.primitives.Txn; import accord.primitives.TxnId; import accord.primitives.Unseekables; @@ -67,6 +71,8 @@ import org.apache.cassandra.service.accord.api.AccordRoutingKey; import org.apache.cassandra.service.accord.api.PartitionKey; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.utils.Pair; +import org.apache.cassandra.utils.RTree; +import org.apache.cassandra.utils.RangeTree; import org.assertj.core.api.Assertions; import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME; @@ -168,6 +174,17 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester return kc; } + protected static void assertDepsMessage(SimulatedAccordCommandStore instance, + DepsMessage messageType, + Txn txn, FullRoute route, + DepsModel model) throws ExecutionException, InterruptedException + { + TxnId id = assertDepsMessage(instance, messageType, txn, route, + model.keyConflicts(txn.keys()), + model.rangeConflicts(txn.keys())); + model.register(id, txn); + } + protected static TxnId assertDepsMessage(SimulatedAccordCommandStore instance, DepsMessage messageType, Txn txn, FullRoute route, @@ -294,7 +311,6 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester else { List actualRanges = IntStream.range(0, deps.rangeDeps.rangeCount()).mapToObj(deps.rangeDeps::range).collect(Collectors.toList()); -// Assertions.assertThat(deps.rangeDeps.rangeCount()).describedAs("Txn %s Expected ranges size; %s", txnId, deps.rangeDeps).isEqualTo(rangeConflicts.size()); Assertions.assertThat(Ranges.of(actualRanges.toArray(Range[]::new))) .describedAs("Txn %s had different ranges than expected", txnId) .isEqualTo(Ranges.of(rangeConflicts.keySet().toArray(Range[]::new))); @@ -380,4 +396,81 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester } }; } + + public static class DepsModel + { + private final Map> keyConflicts = new HashMap<>(); + private final RangeTree rangeConflicts = RTree.create(RangeTreeRangeAccessor.instance); + private final Ranges storeRanges; + + public DepsModel(Ranges storeRanges) + { + this.storeRanges = storeRanges; + } + + public Map> keyConflicts(Seekables keysOrRanges) + { + keysOrRanges = keysOrRanges.slice(storeRanges, Routables.Slice.Minimal); + switch (keysOrRanges.domain()) + { + case Key: + { + Keys keys = (Keys) keysOrRanges; + Map> expectedConflicts = new HashMap<>(); + keys.forEach(k -> expectedConflicts.put(k.toUnseekable(), keyConflicts.getOrDefault(k.toUnseekable(), Collections.emptyList()))); + return expectedConflicts; + } + case Range: + { + Ranges ranges = (Ranges) keysOrRanges; + return keyConflicts.entrySet().stream() + .filter(e -> ranges.contains(e.getKey())) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + } + + default: + throw new UnsupportedOperationException(); + } + } + + public Map> rangeConflicts(Seekables keysOrRanges) + { + // there is a patch pending to add range support for keys... that isn't here yet so not handled + if (keysOrRanges.domain() != Routable.Domain.Range) + return Collections.emptyMap(); + keysOrRanges = keysOrRanges.slice(storeRanges, Routables.Slice.Minimal); + + Ranges ranges = (Ranges) keysOrRanges; + Map> conflicts = new HashMap<>(); + ranges.forEach(r -> rangeConflicts.search(r, e -> { + for (Range range : Ranges.single(e.getKey()).slice(ranges, Routables.Slice.Minimal)) + conflicts.computeIfAbsent(range, ignore -> new ArrayList<>()).add(e.getValue()); + })); + // need to dedup/sort txns + conflicts.values().forEach(l -> { + var sortedDedup = new ArrayList<>(new TreeSet<>(l)); + l.clear(); + l.addAll(sortedDedup); + }); + return conflicts; + } + + public void register(TxnId txnId, Txn txn) + { + for (var s : txn.keys()) + { + switch (s.domain()) + { + case Key: + keyConflicts.computeIfAbsent(s.asKey().toUnseekable(), i -> new ArrayList<>()).add(txnId); + break; + case Range: + rangeConflicts.add(s.asRange(), txnId); + break; + default: + throw new UnsupportedOperationException(); + } + } + } + } } diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java index 961a1dfdb3..1d9935084c 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java @@ -186,13 +186,12 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase FullRangeRoute rangeRoute = ranges.toRoute(pk.toUnseekable()); Txn rangeTxn = createTxn(Txn.Kind.ExclusiveSyncPoint, ranges); - List keyConflicts = new ArrayList<>(numSamples); - List rangeConflicts = new ArrayList<>(numSamples); + DepsModel model = new DepsModel(instance.store.unsafeRangesForEpoch().currentRanges()); for (int i = 0; i < numSamples; i++) { instance.maybeCacheEvict(keyRoute, ranges); - keyConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, keyConflicts(keyConflicts, keyRoute))); - rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts(keyConflicts, keyRoute), rangeConflicts(rangeConflicts, ranges))); + assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, model); + assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, model); } } }); @@ -259,21 +258,18 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase Range left = tokenRange(tbl.id, token - 10, token + 5); Range right = tokenRange(tbl.id, token - 5, token + 10); - List keyConflicts = new ArrayList<>(numSamples); - Map> rangeConflicts = new HashMap<>(); - rangeConflicts.put(left, new ArrayList<>()); - rangeConflicts.put(right, new ArrayList<>()); + DepsModel model = new DepsModel(instance.store.unsafeRangesForEpoch().currentRanges()); for (int i = 0; i < numSamples; i++) { Ranges partialRange = Ranges.of(rs.nextBoolean() ? left : right); try { instance.maybeCacheEvict(keyRoute, partialRange); - keyConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, keyConflicts(keyConflicts, keyRoute))); + assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, model); FullRangeRoute rangeRoute = partialRange.toRoute(pk.toUnseekable()); Txn rangeTxn = createTxn(Txn.Kind.ExclusiveSyncPoint, partialRange); - rangeConflicts.get(partialRange.get(0)).add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts(keyConflicts, keyRoute), rangeConflicts)); + assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, model); } catch (Throwable t) { diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedMultiKeyAndRangeTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedMultiKeyAndRangeTest.java index feaddeff8c..16a4d83065 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedMultiKeyAndRangeTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedMultiKeyAndRangeTest.java @@ -19,11 +19,8 @@ package org.apache.cassandra.service.accord; import java.util.ArrayList; -import java.util.Collections; -import java.util.HashMap; import java.util.HashSet; import java.util.List; -import java.util.Map; import java.util.Set; import java.util.TreeSet; import java.util.stream.Collectors; @@ -40,12 +37,9 @@ import accord.primitives.Range; import accord.primitives.Ranges; import accord.primitives.Routable.Domain; import accord.primitives.Txn; -import accord.primitives.TxnId; import accord.utils.Gen; import accord.utils.Gens; import org.apache.cassandra.service.accord.api.PartitionKey; -import org.apache.cassandra.utils.RTree; -import org.apache.cassandra.utils.RangeTree; import static accord.utils.Property.qt; import static org.apache.cassandra.dht.Murmur3Partitioner.LongToken.keyForToken; @@ -73,12 +67,12 @@ public class SimulatedMultiKeyAndRangeTest extends SimulatedAccordCommandStoreTe Gen.LongGen tokenGen = tokenDistribution.next(rs); Gen domainGen = domainDistribution.next(rs); Gen msgGen = msgDistribution.next(rs); - Map> keyConflicts = new HashMap<>(); - RangeTree rangeConflicts = RTree.create(RangeTreeRangeAccessor.instance); Gen.IntGen keyCountGen = keyDistribution.next(rs); Gen.IntGen rangeCountGen = rangeDistribution.next(rs); + DepsModel model = new DepsModel(instance.store.unsafeRangesForEpoch().currentRanges()); + for (int i = 0; i < numSamples; i++) { switch (domainGen.next(rs)) @@ -99,11 +93,7 @@ public class SimulatedMultiKeyAndRangeTest extends SimulatedAccordCommandStoreTe Txn txn = createTxn(wrapInTxn(inserts), binds); FullRoute route = keys.toRoute(keys.get(0).toUnseekable()); - Map> expectedConflicts = new HashMap<>(); - route.forEach(k -> expectedConflicts.put(k, keyConflicts.computeIfAbsent(k, ignore -> new ArrayList<>()))); - - TxnId id = assertDepsMessage(instance, msgGen.next(rs), txn, route, expectedConflicts, Collections.emptyMap()); - route.forEach(k -> keyConflicts.get(k).add(id)); + assertDepsMessage(instance, msgGen.next(rs), txn, route, model); } break; case Range: @@ -133,21 +123,7 @@ public class SimulatedMultiKeyAndRangeTest extends SimulatedAccordCommandStoreTe FullRangeRoute route = ranges.toRoute(ranges.get(0).end()); Txn txn = createTxn(Txn.Kind.ExclusiveSyncPoint, ranges); - Map> expectedKeyConflicts = keyConflicts.entrySet().stream() - .filter(e -> ranges.contains(e.getKey())) - .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); - Map> expectedRangeConflicts = new HashMap<>(); - ranges.forEach(r -> - rangeConflicts.search(r, e -> - expectedRangeConflicts.computeIfAbsent(e.getKey(), ignore -> new ArrayList<>()).add(e.getValue()))); - // need to dedup/sort txns - expectedRangeConflicts.values().forEach(l -> { - var sortedDedup = new ArrayList<>(new TreeSet<>(l)); - l.clear(); - l.addAll(sortedDedup); - }); - TxnId id = assertDepsMessage(instance, msgGen.next(rs), txn, route, expectedKeyConflicts, expectedRangeConflicts); - ranges.forEach(r -> rangeConflicts.add(r, id)); + assertDepsMessage(instance, msgGen.next(rs), txn, route, model); } break; default: diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java index 30f0f0aa7f..d1edbb4455 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java @@ -25,21 +25,13 @@ import accord.primitives.Keys; import accord.primitives.Ranges; import accord.primitives.RoutingKeys; import accord.primitives.Txn; -import accord.primitives.TxnId; import accord.utils.Property; import accord.utils.RandomSource; -import org.apache.cassandra.dht.Murmur3Partitioner.LongToken; import org.apache.cassandra.schema.TableMetadata; -import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey; import org.apache.cassandra.utils.FailingConsumer; import org.junit.Test; -import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; import static accord.utils.Property.commands; import static accord.utils.Property.stateful; @@ -51,16 +43,14 @@ public class SimulatedRandomKeysWithRangeConflictTest extends SimulatedAccordCom private static Property.SimpleCommand insertKey(RandomSource rs, State state) { long token = rs.nextLong(Long.MIN_VALUE + 1, Long.MAX_VALUE); - RoutingKey key = new TokenKey(state.tbl.id, new LongToken(token)); Txn keyTxn = createTxn(wrapInTxn("INSERT INTO " + state.tbl + "(pk, value) VALUES (?, ?)"), - Arrays.asList(keyForToken(token), 42)); + Arrays.asList(keyForToken(token), 42)); Keys keys = (Keys) keyTxn.keys(); FullRoute keyRoute = keys.toRoute(keys.get(0).toUnseekable()); return new Property.SimpleCommand<>("Write Txn: " + keys, FailingConsumer.orFail(s -> { s.instance.maybeCacheEvict(keyRoute, s.wholeRange); - var k = assertDepsMessage(s.instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, Map.of(key, s.keyConflicts.computeIfAbsent(key, ignore -> new ArrayList<>())), Collections.emptyMap()); - s.keyConflicts.get(key).add(k); + assertDepsMessage(s.instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, s.model); })); } @@ -68,7 +58,7 @@ public class SimulatedRandomKeysWithRangeConflictTest extends SimulatedAccordCom { return new Property.SimpleCommand<>("Range Txn: " + state.wholeRange, FailingConsumer.orFail(s -> { s.instance.maybeCacheEvict(RoutingKeys.EMPTY, s.wholeRange); - s.rangeConflicts.add(assertDepsMessage(s.instance, rs.pick(DepsMessage.values()), s.rangeTxn, s.rangeRoute, s.keyConflicts, rangeConflicts(s.rangeConflicts, s.wholeRange))); + assertDepsMessage(s.instance, rs.pick(DepsMessage.values()), s.rangeTxn, s.rangeRoute, s.model); })); } @@ -77,27 +67,27 @@ public class SimulatedRandomKeysWithRangeConflictTest extends SimulatedAccordCom public void keysAllOverConflictingWithRange() { stateful().withSteps(State.steps).check(commands(() -> State::new) - .add(SimulatedRandomKeysWithRangeConflictTest::insertKey) - .add(SimulatedRandomKeysWithRangeConflictTest::insertRange) - .build()); + .add(SimulatedRandomKeysWithRangeConflictTest::insertKey) + .add(SimulatedRandomKeysWithRangeConflictTest::insertRange) + .build()); } public static class State { static final int steps = 300; final SimulatedAccordCommandStore instance; - final Map> keyConflicts = new HashMap<>(); - final List rangeConflicts = new ArrayList<>(steps); final TableMetadata tbl = reverseTokenTbl; final Ranges wholeRange = Ranges.of(fullRange(tbl.id)); final FullRangeRoute rangeRoute = wholeRange.toRoute(wholeRange.get(0).end()); final Txn rangeTxn = createTxn(Txn.Kind.ExclusiveSyncPoint, wholeRange); + final DepsModel model; public State(RandomSource rs) { AccordKeyspace.unsafeClear(); this.instance = new SimulatedAccordCommandStore(rs); + this.model = new DepsModel(instance.store.unsafeRangesForEpoch().currentRanges()); } @Override diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java index ae11f3fa59..052736200c 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java @@ -34,29 +34,31 @@ import java.util.function.LongUnaryOperator; import java.util.function.Supplier; import org.apache.commons.lang3.ArrayUtils; +import org.junit.After; import org.junit.Assert; +import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; import accord.api.Key; import accord.api.RoutingKey; -import accord.local.StoreParticipants; -import accord.local.cfk.CommandsForKey; -import accord.local.cfk.CommandsForKey.InternalStatus; import accord.local.Command; -import accord.local.cfk.CommandsForKey.TxnInfo; -import accord.local.cfk.CommandsForKey.Unmanaged; import accord.local.CommonAttributes; import accord.local.CommonAttributes.Mutable; import accord.local.Node; -import accord.primitives.SaveStatus; -import accord.primitives.Status; +import accord.local.StoreParticipants; +import accord.local.cfk.CommandsForKey; +import accord.local.cfk.CommandsForKey.InternalStatus; +import accord.local.cfk.CommandsForKey.TxnInfo; +import accord.local.cfk.CommandsForKey.Unmanaged; import accord.primitives.Ballot; import accord.primitives.KeyDeps; import accord.primitives.PartialDeps; import accord.primitives.PartialTxn; import accord.primitives.RangeDeps; import accord.primitives.Routable; +import accord.primitives.SaveStatus; +import accord.primitives.Status; import accord.primitives.Timestamp; import accord.primitives.Txn; import accord.primitives.TxnId; @@ -82,9 +84,9 @@ import org.apache.cassandra.utils.AccordGenerators; import org.apache.cassandra.utils.CassandraGenerators; import static accord.local.cfk.CommandsForKey.NO_BOUNDS_INFO; -import static accord.primitives.Status.Durability.NotDurable; import static accord.primitives.Known.KnownExecuteAt.ExecuteAtErased; import static accord.primitives.Known.KnownExecuteAt.ExecuteAtUnknown; +import static accord.primitives.Status.Durability.NotDurable; import static accord.utils.Property.qt; import static accord.utils.SortedArrays.Search.FAST; import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse; @@ -103,6 +105,18 @@ public class CommandsForKeySerializerTest StorageService.instance.initServer(); } + @Before + public void before() throws Throwable + { + CommandsForKey.disableLinearizabilityViolationsReporting(); + } + + @After + public void after() throws Throwable + { + CommandsForKey.enableLinearizabilityViolationsReporting(); + } + static class Cmd { final TxnId txnId;