natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Check for splittable ranges 

Patch by Alex Petrov; reviewed by Ariel Weisberg for CASSANDRA-20032

Accord Deps tests have incorrect range semantics

patch by David Capwell; reviewed by Ariel Weisberg for CASSANDRA-20029
This commit is contained in:
Alex Petrov 2024-10-23 14:31:12 +02:00 committed by David Capwell
parent e94d86b1e5
commit 4e9110e881
14 changed files with 195 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 25f23ffec439a921387ca249908798b9cc7d4620
Subproject commit a63cac24a2198a5893874cdf72946073854a8d4d

7
src/java/org/apache/cassandra/dht/AccordSplitter.java
View File

@ -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()
{

15
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -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<TxnId, Command, AccordSafeCommand> commandCache;
private final AccordStateCache.Instance<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey> 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)

8
src/java/org/apache/cassandra/tcm/AtomicLongBackedProcessor.java
View File

@ -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<Entry> iter = state.iterator();
ImmutableList.Builder<Entry> 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);
}

17
src/java/org/apache/cassandra/tcm/log/LogReader.java
View File

@ -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<Entry> 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<Entry> 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));
}
}

5
src/java/org/apache/cassandra/tcm/log/LogState.java
View File

@ -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<Entry> 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;
}

2
src/java/org/apache/cassandra/tcm/transformations/BeginConsensusMigrationForTableAndRange.java
View File

@ -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;

13
test/distributed/org/apache/cassandra/distributed/test/log/SystemKeyspaceStorageTest.java
View File

@ -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);
}
}
}

11
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
View File

@ -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());
}

95
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java
View File

@ -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<Range> 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<RoutingKey, List<TxnId>> keyConflicts = new HashMap<>();
private final RangeTree<RoutingKey, Range, TxnId> rangeConflicts = RTree.create(RangeTreeRangeAccessor.instance);
private final Ranges storeRanges;
public DepsModel(Ranges storeRanges)
{
this.storeRanges = storeRanges;
}
public Map<RoutingKey, List<TxnId>> keyConflicts(Seekables<?, ?> keysOrRanges)
{
keysOrRanges = keysOrRanges.slice(storeRanges, Routables.Slice.Minimal);
switch (keysOrRanges.domain())
{
case Key:
{
Keys keys = (Keys) keysOrRanges;
Map<RoutingKey, List<TxnId>> 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<Range, List<TxnId>> 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<Range, List<TxnId>> 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();
}
}
}
}
}

16
test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java
View File

@ -186,13 +186,12 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
FullRangeRoute rangeRoute = ranges.toRoute(pk.toUnseekable());
Txn rangeTxn = createTxn(Txn.Kind.ExclusiveSyncPoint, ranges);
List<TxnId> keyConflicts = new ArrayList<>(numSamples);
List<TxnId> 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<TxnId> keyConflicts = new ArrayList<>(numSamples);
Map<Range, List<TxnId>> 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)
{

32
test/unit/org/apache/cassandra/service/accord/SimulatedMultiKeyAndRangeTest.java
View File

@ -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<Domain> domainGen = domainDistribution.next(rs);
Gen<DepsMessage> msgGen = msgDistribution.next(rs);
Map<RoutingKey, List<TxnId>> keyConflicts = new HashMap<>();
RangeTree<RoutingKey, Range, TxnId> 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<RoutingKey> route = keys.toRoute(keys.get(0).toUnseekable());
Map<RoutingKey, List<TxnId>> 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<RoutingKey, List<TxnId>> expectedKeyConflicts = keyConflicts.entrySet().stream()
.filter(e -> ranges.contains(e.getKey()))
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
Map<Range, List<TxnId>> 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:

26
test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java
View File

@ -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<State> 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<RoutingKey> 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<RoutingKey, List<TxnId>> keyConflicts = new HashMap<>();
final List<TxnId> 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

30
test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
View File

@ -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;