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

CommandsForRanges does not support slice which cause over returned data being sent 

patch by David Capwell; reviewed by Alex Petrov for CASSANDRA-19857
This commit is contained in:
David Capwell 2024-08-23 09:49:54 -07:00
parent 8ab5003118
commit a7c2bcafcd
13 changed files with 226 additions and 181 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
View File

@ -179,6 +179,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
@Override
public void registerHistoricalTransactions(Deps deps)
{
if (deps.isEmpty()) return;
// used in places such as accord.local.CommandStore.fetchMajorityDeps
// We find a set of dependencies for a range then update CommandsFor to know about them
Ranges allRanges = ranges.all();
@ -221,28 +222,27 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
{
if (commandsForRanges == null)
return accumulate;
CommandsForRanges cfr = commandsForRanges.current().slice(slice);
switch (keysOrRanges.domain())
{
case Key:
{
AbstractKeys<Key> keys = (AbstractKeys<Key>) keysOrRanges.slice(slice, Routables.Slice.Minimal);
if (!commandsForRanges.ranges().intersects(keys))
if (!cfr.ranges.intersects(keys))
return accumulate;
accumulate = map.apply(commandsForRanges.current(), accumulate);
}
break;
case Range:
{
AbstractRanges ranges = (AbstractRanges) keysOrRanges.slice(slice, Routables.Slice.Minimal);
if (!commandsForRanges.ranges().intersects(ranges))
if (!cfr.ranges.intersects(ranges))
return accumulate;
accumulate = map.apply(commandsForRanges.current(), accumulate);
}
break;
default:
throw new AssertionError("Unknown domain: " + keysOrRanges.domain());
}
return accumulate;
return map.apply(cfr, accumulate);
}
private <O> O mapReduceForKey(Routables<?> keysOrRanges, Ranges slice, BiFunction<CommandsSummary, O, O> map, O accumulate)

2
src/java/org/apache/cassandra/service/accord/AccordSafeCommandsForRanges.java
View File

@ -49,7 +49,7 @@ public class AccordSafeCommandsForRanges extends ImmutableAccordSafeState<Ranges
Pair<CommandsForRangesLoader.Watcher, NavigableMap<TxnId, CommandsForRangesLoader.Summary>> pair = AsyncChains.getUnchecked(chain);
pair.left.close();
pair.left.get().entrySet().forEach(e -> pair.right.put(e.getKey(), e.getValue()));
original = new CommandsForRanges(key, pair.right);
original = CommandsForRanges.create(key, pair.right);
}
@Override

32
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
View File

@ -26,6 +26,8 @@ import java.util.TreeMap;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import accord.impl.CommandsSummary;
import accord.local.SafeCommandStore.CommandFunction;
import accord.local.SafeCommandStore.TestDep;
@ -44,16 +46,28 @@ import static accord.local.SafeCommandStore.TestStartedAt.STARTED_BEFORE;
import static accord.local.SafeCommandStore.TestStatus.ANY_STATUS;
import static accord.local.Status.Stable;
import static accord.local.Status.Truncated;
import static accord.primitives.Routables.Slice.Minimal;
public class CommandsForRanges implements CommandsSummary
{
private final Ranges ranges;
public final Ranges ranges;
private final NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map;
public CommandsForRanges(Ranges ranges, NavigableMap<TxnId, CommandsForRangesLoader.Summary> map)
private CommandsForRanges(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
{
this.ranges = ranges;
this.map = (NavigableMap<Timestamp, CommandsForRangesLoader.Summary>) (NavigableMap<?, ?>) map;
this.map = map;
}
public static CommandsForRanges create(Ranges ranges, NavigableMap<TxnId, CommandsForRangesLoader.Summary> map)
{
return new CommandsForRanges(ranges, (NavigableMap<Timestamp, CommandsForRangesLoader.Summary>) (NavigableMap<?, ?>) map);
}
@VisibleForTesting
public int size()
{
return map.size();
}
@Override
@ -160,4 +174,16 @@ public class CommandsForRanges implements CommandsSummary
return accumulate;
}
public CommandsForRanges slice(Ranges slice)
{
Ranges ranges = this.ranges.slice(slice, Minimal);
NavigableMap<Timestamp, CommandsForRangesLoader.Summary> copy = new TreeMap<>();
for (Map.Entry<Timestamp, CommandsForRangesLoader.Summary> e : map.entrySet())
{
if (!e.getValue().ranges.intersects(slice)) continue;
copy.put(e.getKey(), e.getValue().slice(slice));
}
return new CommandsForRanges(ranges, copy);
}
}

10
src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java
View File

@ -31,6 +31,7 @@ import java.util.TreeMap;
import java.util.function.BiFunction;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableMap;
import accord.local.Command;
@ -41,6 +42,7 @@ import accord.primitives.PartialDeps;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.Routables;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
@ -256,7 +258,8 @@ public class CommandsForRangesLoader
public final Ranges ranges;
public final List<TxnId> depsIds;
private Summary(TxnId txnId, @Nullable Timestamp executeAt, SaveStatus saveStatus, Ranges ranges, List<TxnId> depsIds)
@VisibleForTesting
Summary(TxnId txnId, @Nullable Timestamp executeAt, SaveStatus saveStatus, Ranges ranges, List<TxnId> depsIds)
{
this.txnId = txnId;
this.executeAt = executeAt;
@ -265,6 +268,11 @@ public class CommandsForRangesLoader
this.depsIds = depsIds;
}
public Summary slice(Ranges slice)
{
return new Summary(txnId, executeAt, saveStatus, ranges.slice(slice, Routables.Slice.Minimal), depsIds);
}
@Override
public String toString()
{

2
src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java
View File

@ -304,7 +304,7 @@ public abstract class AsyncOperation<R> extends AsyncChains.Head<R> implements R
}
catch (Throwable t)
{
logger.error(String.format("Operation %s failed", this), t);
logger.error("Operation {} failed", this, t);
fail(t);
}
finally

129
test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java Normal file
View File

@ -0,0 +1,129 @@
/*
* 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.service.accord;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.TreeMap;
import org.junit.Test;
import accord.api.RoutingKey;
import accord.impl.IntKey;
import accord.local.SaveStatus;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.TxnId;
import accord.utils.AccordGens;
import accord.utils.Gen;
import accord.utils.Gens;
import static accord.utils.Property.qt;
import static org.assertj.core.api.Assertions.assertThat;
public class CommandsForRangesTest
{
private static final AccordGens.RangeFactory<IntKey.Routing> ROUTING_RANGE_FACTORY = (i, a, b) -> IntKey.range(a, b);
private static final Gen.IntGen SMALL_INTS = Gens.ints().between(1, 10);
private static final Gen<Ranges> RANGES_GEN = AccordGens.ranges(SMALL_INTS, AccordGens.intRoutingKey(), ROUTING_RANGE_FACTORY);
private static final Gen<TxnId> TXN_ID_GEN = AccordGens.txnIds();
private static final Gen<CommandsForRanges> CFK_GEN = rs -> {
Ranges ranges = RANGES_GEN.next(rs);
int numTxn = 10;
TreeMap<TxnId, CommandsForRangesLoader.Summary> map = new TreeMap<>();
for (int i = 0; i < numTxn; i++)
{
TxnId id = TXN_ID_GEN.next(rs);
map.put(id, new CommandsForRangesLoader.Summary(id, id, SaveStatus.ReadyToExecute, ranges, Collections.emptyList()));
}
return CommandsForRanges.create(ranges, map);
};
private static final IntKey.Routing MIN = IntKey.routing(Integer.MIN_VALUE);
private static final IntKey.Routing MAX = IntKey.routing(Integer.MAX_VALUE);
@Test
public void sliceEmptyWhenOutside()
{
qt().check(rs -> {
CommandsForRanges cfr = CFK_GEN.next(rs);
for (Range range : allOutside(cfr.ranges))
{
Ranges slice = Ranges.single(range);
CommandsForRanges subset = cfr.slice(slice);
assertThat(subset.ranges).isEmpty();
assertThat(subset.size()).isEqualTo(0);
}
});
}
@Test
public void sliceSameNoop()
{
qt().check(rs -> {
CommandsForRanges cfr = CFK_GEN.next(rs);
CommandsForRanges subset = cfr.slice(cfr.ranges);
assertThat(subset.ranges).isEqualTo(cfr.ranges);
assertThat(subset.size()).isEqualTo(cfr.size());
});
}
private static List<Range> allOutside(Ranges ranges)
{
if (ranges.isEmpty()) return Collections.emptyList();
List<Range> matches = new ArrayList<>();
{
Range first = ranges.get(0);
if (!first.start().equals(MIN))
{
int start = Integer.MIN_VALUE;
int end = key(first.start());
matches.add(IntKey.range(start, end));
}
}
if (ranges.size() > 1)
{
{
Range last = ranges.get(ranges.size() - 1);
if (!last.end().equals(MAX))
{
int start = key(last.end());
int end = Integer.MAX_VALUE;
matches.add(IntKey.range(start, end));
}
}
for (int i = 1; i < ranges.size(); i++)
{
Range previous = ranges.get(i - 1);
Range next = ranges.get(i - 1);
int start = key(previous.end());
int end = key(next.start());
if (start < end)
matches.add(IntKey.range(start, end));
}
}
return matches;
}
private static int key(RoutingKey key)
{
return ((IntKey.Routing) key).key;
}
}

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

@ -46,6 +46,7 @@ import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.RoutableKey;
import accord.primitives.Routables;
import accord.primitives.Seekables;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@ -200,6 +201,11 @@ public class SimulatedAccordCommandStore implements AutoCloseable
shouldCompact = boolSource(rs.fork());
}
public Ranges slice(Ranges ranges)
{
return ranges.slice(topology.ranges(), Routables.Slice.Minimal);
}
private static BooleanSupplier boolSource(RandomSource rs)
{
var gen = Gens.bools().mixedDistribution().next(rs);

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

@ -133,30 +133,6 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester
ServerTestUtils.markCMS();
}
protected static void safeBlock(List<AsyncResult<?>> asyncs) throws InterruptedException, ExecutionException
{
int counter = 0;
for (var chain : asyncs)
{
Assertions.assertThat(chain.isDone())
.describedAs("The %dth async task is blocked!", counter++)
.isTrue();
AsyncChains.getBlocking(chain);
}
}
protected static void safeBlock(List<AsyncResult<?>> asyncs, List<?> details) throws InterruptedException, ExecutionException
{
int counter = 0;
for (var chain : asyncs)
{
Assertions.assertThat(chain.isDone())
.describedAs("The %dth async task %s is blocked!", counter, details.get(counter++))
.isTrue();
AsyncChains.getBlocking(chain);
}
}
protected static TokenRange fullRange(TableId id)
{
return new TokenRange(AccordRoutingKey.SentinelKey.min(id), AccordRoutingKey.SentinelKey.max(id));
@ -174,25 +150,19 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester
protected static Map<Key, List<TxnId>> keyConflicts(List<TxnId> list, Keys keys)
{
if (list.isEmpty()) return Collections.emptyMap();
Map<Key, List<TxnId>> kc = Maps.newHashMapWithExpectedSize(keys.size());
for (Key key : keys)
{
if (list.isEmpty())
continue;
kc.put(key, list);
}
return kc;
}
protected static Map<Range, List<TxnId>> rangeConflicts(List<TxnId> list, Ranges ranges)
{
if (list.isEmpty()) return Collections.emptyMap();
Map<Range, List<TxnId>> kc = Maps.newHashMapWithExpectedSize(ranges.size());
for (Range range : ranges)
{
if (list.isEmpty())
continue;
kc.put(range, list);
}
return kc;
}
@ -217,14 +187,6 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester
return pair.left;
}
protected static Pair<TxnId, AsyncResult<?>> assertDepsMessageAsync(SimulatedAccordCommandStore instance,
DepsMessage messageType,
Txn txn, FullRoute<?> route,
Map<Key, List<TxnId>> keyConflicts)
{
return assertDepsMessageAsync(instance, messageType, txn, route, keyConflicts, Collections.emptyMap());
}
protected static Pair<TxnId, AsyncResult<?>> assertDepsMessageAsync(SimulatedAccordCommandStore instance,
DepsMessage messageType,
Txn txn, FullRoute<?> route,

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

@ -26,7 +26,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.Ignore;
import org.junit.Test;
import accord.api.Key;
@ -38,7 +37,6 @@ import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.utils.async.AsyncResult;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
import org.apache.cassandra.schema.TableMetadata;
@ -47,7 +45,6 @@ import org.apache.cassandra.service.accord.api.PartitionKey;
import static accord.utils.Property.qt;
import static org.apache.cassandra.service.accord.AccordTestUtils.createTxn;
@Ignore // TODO (required): This class relies on removed ExecutionOrder for correctness, and needs to be adjusted
public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
{
@Test
@ -66,33 +63,17 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
try (var instance = new SimulatedAccordCommandStore(rs))
{
List<TxnId> conflicts = new ArrayList<>(numSamples);
boolean concurrent = rs.nextBoolean();
List<AsyncResult<?>> asyncs = !concurrent ? null : new ArrayList<>(numSamples);
for (int i = 0; i < numSamples; i++)
{
instance.maybeCacheEvict(keys, Ranges.EMPTY);
if (concurrent)
{
var pair = assertDepsMessageAsync(instance, rs.pick(DepsMessage.values()), txn, route, keyConflicts(conflicts, keys));
conflicts.add(pair.left);
asyncs.add(pair.right);
}
else
{
conflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), txn, route, keyConflicts(conflicts, keys)));
}
}
if (concurrent)
{
instance.processAll();
safeBlock(asyncs);
conflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), txn, route, keyConflicts(conflicts, keys)));
}
}
});
}
@Test
public void concurrentRangePartialKeyMatch()
public void rangePartialKeyMatch()
{
var tbl = reverseTokenTbl;
int numSamples = 250;
@ -104,6 +85,7 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
{
long token = rs.nextLong(Long.MIN_VALUE + 1, Long.MAX_VALUE);
Ranges partialRange = Ranges.of(tokenRange(tbl.id, token - 1, token));
Ranges partialRangeSliced = instance.slice(partialRange);
long outOfRangeToken = token - 10;
if (outOfRangeToken == Long.MIN_VALUE) // if this wraps around that is fine, just can't be min
outOfRangeToken++;
@ -121,38 +103,24 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
Keys conflictingKeys = (Keys) conflictingKeyTxn.keys();
FullRoute<?> conflictingRoute = conflictingKeys.toRoute(conflictingKeys.get(0).toUnseekable());
FullRangeRoute rangeRoute = partialRange.toRoute(keys.get(0).toUnseekable());
FullRangeRoute rangeRoute = partialRange.toRoute(key.toUnseekable());
Txn rangeTxn = createTxn(Txn.Kind.ExclusiveSyncPoint, partialRange);
List<TxnId> keyConflicts = new ArrayList<>(numSamples);
List<TxnId> outOfRangeKeyConflicts = new ArrayList<>(numSamples);
List<TxnId> rangeConflicts = new ArrayList<>(numSamples);
List<AsyncResult<?>> asyncs = new ArrayList<>(numSamples * 2 + numSamples * numConflictKeyTxns);
List<TxnId> asyncIds = new ArrayList<>(numSamples * 2 + numSamples * numConflictKeyTxns);
for (int i = 0; i < numSamples; i++)
{
instance.maybeCacheEvict((Keys) keyTxn.keys(), partialRange);
for (int j = 0; j < numConflictKeyTxns; j++)
{
var p = instance.enqueuePreAccept(conflictingKeyTxn, conflictingRoute);
outOfRangeKeyConflicts.add(p.left);
asyncs.add(p.right);
asyncIds.add(p.left);
}
outOfRangeKeyConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), conflictingKeyTxn, conflictingRoute, Map.of(outOfRangeKey, outOfRangeKeyConflicts)));
var k = assertDepsMessageAsync(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, Map.of(key, keyConflicts, outOfRangeKey, outOfRangeKeyConflicts), Collections.emptyMap());
keyConflicts.add(k.left);
outOfRangeKeyConflicts.add(k.left);
asyncs.add(k.right);
asyncIds.add(k.left);
TxnId id = assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, Map.of(key, keyConflicts, outOfRangeKey, outOfRangeKeyConflicts));
keyConflicts.add(id);
outOfRangeKeyConflicts.add(id);
var r = assertDepsMessageAsync(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, Map.of(key, keyConflicts), rangeConflicts(rangeConflicts, partialRange));
rangeConflicts.add(r.left);
asyncs.add(r.right);
asyncIds.add(r.left);
rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, Map.of(key, keyConflicts), rangeConflicts(rangeConflicts, partialRangeSliced)));
}
instance.processAll();
safeBlock(asyncs, asyncIds);
}
});
}
@ -183,30 +151,11 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
List<TxnId> keyConflicts = new ArrayList<>(numSamples);
List<TxnId> rangeConflicts = new ArrayList<>(numSamples);
boolean concurrent = rs.nextBoolean();
List<AsyncResult<?>> asyncs = !concurrent ? null : new ArrayList<>(numSamples * 2);
for (int i = 0; i < numSamples; i++)
{
instance.maybeCacheEvict(keys, ranges);
if (concurrent)
{
var k = assertDepsMessageAsync(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, keyConflicts(keyConflicts, keys));
keyConflicts.add(k.left);
asyncs.add(k.right);
var r = assertDepsMessageAsync(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts(keyConflicts, keys), rangeConflicts(rangeConflicts, ranges));
rangeConflicts.add(r.left);
asyncs.add(r.right);
}
else
{
keyConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, keyConflicts(keyConflicts, keys)));
rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts(keyConflicts, keys), rangeConflicts(rangeConflicts, ranges)));
}
}
if (concurrent)
{
instance.processAll();
safeBlock(asyncs);
keyConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, keyConflicts(keyConflicts, keys)));
rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts(keyConflicts, keys), rangeConflicts(rangeConflicts, instance.slice(ranges))));
}
}
});
@ -218,7 +167,7 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
var tbl = reverseTokenTbl;
int numSamples = 100;
qt().withSeed(6484101342775432632L).withExamples(10).check(rs -> {
qt().withExamples(10).check(rs -> {
AccordKeyspace.unsafeClear();
try (var instance = new SimulatedAccordCommandStore(rs))
{
@ -231,9 +180,6 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
List<TxnId> keyConflicts = new ArrayList<>(numSamples);
Map<Range, List<TxnId>> rangeConflicts = new HashMap<>();
boolean concurrent = rs.nextBoolean();
List<AsyncResult<?>> asyncs = !concurrent ? null : new ArrayList<>(numSamples);
List<TxnId> info = !concurrent ? null : new ArrayList<>(numSamples);
for (int i = 0; i < numSamples; i++)
{
Ranges partialRange = Ranges.of(tokenRange(tbl.id, token - i - 1, token + i));
@ -242,23 +188,8 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
try
{
instance.maybeCacheEvict(keys, partialRange);
if (concurrent)
{
var pair = assertDepsMessageAsync(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, keyConflicts(keyConflicts, keys));
info.add(pair.left);
keyConflicts.add(pair.left);
asyncs.add(pair.right);
pair = assertDepsMessageAsync(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts(keyConflicts, keys), rangeConflicts);
info.add(pair.left);
rangeConflicts.put(partialRange.get(0), Collections.singletonList(pair.left));
asyncs.add(pair.right);
}
else
{
keyConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, keyConflicts(keyConflicts, keys)));
rangeConflicts.put(partialRange.get(0), Collections.singletonList(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts(keyConflicts, keys), rangeConflicts)));
}
keyConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, keyConflicts(keyConflicts, keys)));
rangeConflicts.put(partialRange.get(0), Collections.singletonList(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts(keyConflicts, keys), rangeConflicts)));
}
catch (Throwable t)
{
@ -267,11 +198,6 @@ public class SimulatedDepsTest extends SimulatedAccordCommandStoreTestBase
throw t;
}
}
if (concurrent)
{
instance.processAll();
safeBlock(asyncs, info);
}
}
});
}

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

@ -29,7 +29,6 @@ import java.util.TreeSet;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.junit.Ignore;
import org.junit.Test;
import accord.api.Key;
@ -44,7 +43,6 @@ import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.utils.Gen;
import accord.utils.Gens;
import accord.utils.async.AsyncResult;
import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.utils.RTree;
import org.apache.cassandra.utils.RangeTree;
@ -53,7 +51,6 @@ import static accord.utils.Property.qt;
import static org.apache.cassandra.dht.Murmur3Partitioner.LongToken.keyForToken;
import static org.apache.cassandra.service.accord.AccordTestUtils.createTxn;
@Ignore // TODO (required): This class relies on removed ExecutionOrder for correctness, and needs to be adjusted
public class SimulatedMultiKeyAndRangeTest extends SimulatedAccordCommandStoreTestBase
{
@Test
@ -78,7 +75,6 @@ public class SimulatedMultiKeyAndRangeTest extends SimulatedAccordCommandStoreTe
Gen<DepsMessage> msgGen = msgDistribution.next(rs);
Map<Key, List<TxnId>> keyConflicts = new HashMap<>();
RangeTree<RoutingKey, Range, TxnId> rangeConflicts = RTree.create(RangeTreeRangeAccessor.instance);
List<AsyncResult<?>> asyncs = new ArrayList<>(numSamples);
Gen.IntGen keyCountGen = keyDistribution.next(rs);
Gen.IntGen rangeCountGen = rangeDistribution.next(rs);
@ -106,9 +102,8 @@ public class SimulatedMultiKeyAndRangeTest extends SimulatedAccordCommandStoreTe
Map<Key, List<TxnId>> expectedConflicts = new HashMap<>();
keys.forEach(k -> expectedConflicts.put(k, keyConflicts.computeIfAbsent(k, ignore -> new ArrayList<>())));
var p = assertDepsMessageAsync(instance, msgGen.next(rs), txn, route, expectedConflicts, Collections.emptyMap());
keys.forEach(k -> keyConflicts.get(k).add(p.left));
asyncs.add(p.right);
TxnId id = assertDepsMessage(instance, msgGen.next(rs), txn, route, expectedConflicts, Collections.emptyMap());
keys.forEach(k -> keyConflicts.get(k).add(id));
}
break;
case Range:
@ -151,17 +146,14 @@ public class SimulatedMultiKeyAndRangeTest extends SimulatedAccordCommandStoreTe
l.clear();
l.addAll(sortedDedup);
});
var p = assertDepsMessageAsync(instance, msgGen.next(rs), txn, route, expectedKeyConflicts, expectedRangeConflicts);
asyncs.add(p.right);
ranges.forEach(r -> rangeConflicts.add(r, p.left));
TxnId id = assertDepsMessage(instance, msgGen.next(rs), txn, route, expectedKeyConflicts, expectedRangeConflicts);
ranges.forEach(r -> rangeConflicts.add(r, id));
}
break;
default:
throw new AssertionError();
}
}
instance.processAll();
safeBlock(asyncs);
}
});
}

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

@ -25,7 +25,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.Ignore;
import org.junit.Test;
import accord.api.Key;
@ -35,14 +34,12 @@ import accord.primitives.Keys;
import accord.primitives.Ranges;
import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.utils.async.AsyncResult;
import org.apache.cassandra.service.accord.api.PartitionKey;
import static accord.utils.Property.qt;
import static org.apache.cassandra.dht.Murmur3Partitioner.LongToken.keyForToken;
import static org.apache.cassandra.service.accord.AccordTestUtils.createTxn;
@Ignore // TODO (required): This class relies on removed ExecutionOrder for correctness, and needs to be adjusted
public class SimulatedRandomKeysWithRangeConflictTest extends SimulatedAccordCommandStoreTestBase
{
@Test
@ -60,8 +57,6 @@ public class SimulatedRandomKeysWithRangeConflictTest extends SimulatedAccordCom
{
Map<Key, List<TxnId>> keyConflicts = new HashMap<>();
List<TxnId> rangeConflicts = new ArrayList<>(numSamples);
boolean concurrent = rs.nextBoolean();
List<AsyncResult<?>> asyncs = !concurrent ? null : new ArrayList<>(numSamples * 2);
for (int i = 0; i < numSamples; i++)
{
long token = rs.nextLong(Long.MIN_VALUE + 1, Long.MAX_VALUE);
@ -73,27 +68,11 @@ public class SimulatedRandomKeysWithRangeConflictTest extends SimulatedAccordCom
instance.maybeCacheEvict((Keys) keyTxn.keys(), wholeRange);
if (concurrent)
{
var k = assertDepsMessageAsync(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, Map.of(key, keyConflicts.computeIfAbsent(key, ignore -> new ArrayList<>())), Collections.emptyMap());
keyConflicts.get(key).add(k.left);
asyncs.add(k.right);
var r = assertDepsMessageAsync(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts, rangeConflicts(rangeConflicts, wholeRange));
rangeConflicts.add(r.left);
asyncs.add(r.right);
}
else
{
var k = assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, Map.of(key, keyConflicts.computeIfAbsent(key, ignore -> new ArrayList<>())), Collections.emptyMap());
keyConflicts.get(key).add(k);
rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts, rangeConflicts(rangeConflicts, wholeRange)));
}
}
if (concurrent)
{
instance.processAll();
safeBlock(asyncs);
// the full range is (-Inf, +Inf] but the store could be [(-Inf, Number], (Number, +Inf]], so need to slice to the store to get a matching range
Ranges wholeRangeSlicedShard = instance.slice(wholeRange);
var k = assertDepsMessage(instance, rs.pick(DepsMessage.values()), keyTxn, keyRoute, Map.of(key, keyConflicts.computeIfAbsent(key, ignore -> new ArrayList<>())), Collections.emptyMap());
keyConflicts.get(key).add(k);
rangeConflicts.add(assertDepsMessage(instance, rs.pick(DepsMessage.values()), rangeTxn, rangeRoute, keyConflicts, rangeConflicts(rangeConflicts, wholeRangeSlicedShard)));
}
}
});

4
test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java
View File

@ -350,7 +350,7 @@ public class AsyncOperationTest
Gen<TxnId> txnIdGen = rs -> txnId(1, clock.incrementAndGet(), 1);
qt().withPure(false)
.withSeed(-3537445084098883509L).withExamples(50)
.withExamples(50)
.forAll(Gens.random(), Gens.lists(txnIdGen).ofSizeBetween(1, 10))
.check((rs, ids) -> {
before(); // truncate tables
@ -413,7 +413,7 @@ public class AsyncOperationTest
Gen<TxnId> txnIdGen = rs -> txnId(1, clock.incrementAndGet(), 1);
AtomicInteger counter = new AtomicInteger();
qt().withPure(false).withSeed(3131884991952253478L).withExamples(100).forAll(Gens.random(), Gens.lists(txnIdGen).ofSizeBetween(1, 10)).check((rs, ids) -> {
qt().withPure(false).withExamples(100).forAll(Gens.random(), Gens.lists(txnIdGen).ofSizeBetween(1, 10)).check((rs, ids) -> {
logger.info("Test #{}", counter.incrementAndGet());
before(); // truncate tables

21
test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java
View File

@ -24,13 +24,17 @@ import org.junit.Test;
import accord.primitives.Deps;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.LocalPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.IVersionedSerializers;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.SchemaProvider;
import org.apache.cassandra.utils.AbstractTypeGenerators;
import org.apache.cassandra.utils.AccordGenerators;
import org.mockito.Mockito;
@ -51,8 +55,8 @@ public class DepsSerializerTest
public void serde()
{
DataOutputBuffer buffer = new DataOutputBuffer();
qt().withSeed(-4368731546033726179L).check(rs -> {
IPartitioner partitioner = AccordGenerators.partitioner().next(rs);
qt().check(rs -> {
IPartitioner partitioner = AccordGenerators.partitioner().map(DepsSerializerTest::normalize).next(rs);
Schema.instance = Mockito.mock(SchemaProvider.class);
DatabaseDescriptor.setPartitionerUnsafe(partitioner);
Mockito.when(Schema.instance.getExistingTablePartitioner(Mockito.any())).thenReturn(partitioner);
@ -61,4 +65,17 @@ public class DepsSerializerTest
IVersionedSerializers.testSerde(buffer, DepsSerializer.deps, deps, version.value);
});
}
private static IPartitioner normalize(IPartitioner partitioner)
{
// serializers require tokens to fit within 1 << 16, but that makes the test flakey when LocalPartitioner with a nested type is found...
if (!(partitioner instanceof LocalPartitioner)) return partitioner;
if (!shouldSimplify(partitioner.getTokenValidator())) return partitioner;
return new LocalPartitioner(Int32Type.instance);
}
private static boolean shouldSimplify(AbstractType<?> type)
{
return AbstractTypeGenerators.contains(type, t -> t.isCollection());
}
}