diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java index a5215eadea..2b60bfcd51 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java @@ -179,6 +179,7 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore keys = (AbstractKeys) 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 mapReduceForKey(Routables keysOrRanges, Ranges slice, BiFunction map, O accumulate) diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandsForRanges.java index 848df1d270..1a90c0a700 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandsForRanges.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandsForRanges.java @@ -49,7 +49,7 @@ public class AccordSafeCommandsForRanges extends ImmutableAccordSafeState> 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 diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java index 720013762a..4da915448c 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java @@ -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 map; - public CommandsForRanges(Ranges ranges, NavigableMap map) + private CommandsForRanges(Ranges ranges, NavigableMap map) { this.ranges = ranges; - this.map = (NavigableMap) (NavigableMap) map; + this.map = map; + } + + public static CommandsForRanges create(Ranges ranges, NavigableMap map) + { + return new CommandsForRanges(ranges, (NavigableMap) (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 copy = new TreeMap<>(); + for (Map.Entry e : map.entrySet()) + { + if (!e.getValue().ranges.intersects(slice)) continue; + copy.put(e.getKey(), e.getValue().slice(slice)); + } + return new CommandsForRanges(ranges, copy); + } } diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java index f90d57b32e..0b4b357128 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java @@ -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 depsIds; - private Summary(TxnId txnId, @Nullable Timestamp executeAt, SaveStatus saveStatus, Ranges ranges, List depsIds) + @VisibleForTesting + Summary(TxnId txnId, @Nullable Timestamp executeAt, SaveStatus saveStatus, Ranges ranges, List 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() { diff --git a/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java b/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java index 7a39b0393f..34d80250a2 100644 --- a/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java +++ b/src/java/org/apache/cassandra/service/accord/async/AsyncOperation.java @@ -304,7 +304,7 @@ public abstract class AsyncOperation extends AsyncChains.Head implements R } catch (Throwable t) { - logger.error(String.format("Operation %s failed", this), t); + logger.error("Operation {} failed", this, t); fail(t); } finally diff --git a/test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java b/test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java new file mode 100644 index 0000000000..5993d1d133 --- /dev/null +++ b/test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java @@ -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 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_GEN = AccordGens.ranges(SMALL_INTS, AccordGens.intRoutingKey(), ROUTING_RANGE_FACTORY); + private static final Gen TXN_ID_GEN = AccordGens.txnIds(); + private static final Gen CFK_GEN = rs -> { + Ranges ranges = RANGES_GEN.next(rs); + int numTxn = 10; + TreeMap 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 allOutside(Ranges ranges) + { + if (ranges.isEmpty()) return Collections.emptyList(); + List 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; + } +} diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java index dfcfbebdd4..6e947051b3 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java @@ -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); diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java index 2c313566df..1c05c0a0ad 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java @@ -133,30 +133,6 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester ServerTestUtils.markCMS(); } - protected static void safeBlock(List> 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> 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> keyConflicts(List list, Keys keys) { + if (list.isEmpty()) return Collections.emptyMap(); Map> kc = Maps.newHashMapWithExpectedSize(keys.size()); for (Key key : keys) - { - if (list.isEmpty()) - continue; kc.put(key, list); - } return kc; } protected static Map> rangeConflicts(List list, Ranges ranges) { + if (list.isEmpty()) return Collections.emptyMap(); Map> 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> assertDepsMessageAsync(SimulatedAccordCommandStore instance, - DepsMessage messageType, - Txn txn, FullRoute route, - Map> keyConflicts) - { - return assertDepsMessageAsync(instance, messageType, txn, route, keyConflicts, Collections.emptyMap()); - } - protected static Pair> assertDepsMessageAsync(SimulatedAccordCommandStore instance, DepsMessage messageType, Txn txn, FullRoute route, diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java index 66485cf8d6..9fda5cd16f 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedDepsTest.java @@ -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 conflicts = new ArrayList<>(numSamples); - boolean concurrent = rs.nextBoolean(); - List> 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 keyConflicts = new ArrayList<>(numSamples); List outOfRangeKeyConflicts = new ArrayList<>(numSamples); List rangeConflicts = new ArrayList<>(numSamples); - List> asyncs = new ArrayList<>(numSamples * 2 + numSamples * numConflictKeyTxns); - List 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 keyConflicts = new ArrayList<>(numSamples); List rangeConflicts = new ArrayList<>(numSamples); - boolean concurrent = rs.nextBoolean(); - List> 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 keyConflicts = new ArrayList<>(numSamples); Map> rangeConflicts = new HashMap<>(); - boolean concurrent = rs.nextBoolean(); - List> asyncs = !concurrent ? null : new ArrayList<>(numSamples); - List 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); - } } }); } diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedMultiKeyAndRangeTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedMultiKeyAndRangeTest.java index c345a26866..11497133b3 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedMultiKeyAndRangeTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedMultiKeyAndRangeTest.java @@ -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 msgGen = msgDistribution.next(rs); Map> keyConflicts = new HashMap<>(); RangeTree rangeConflicts = RTree.create(RangeTreeRangeAccessor.instance); - List> 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> 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); } }); } diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java index b3df25bdfb..31880d3297 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedRandomKeysWithRangeConflictTest.java @@ -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> keyConflicts = new HashMap<>(); List rangeConflicts = new ArrayList<>(numSamples); - boolean concurrent = rs.nextBoolean(); - List> 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))); } } }); diff --git a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java index ffdb46eaaf..11ed33b42f 100644 --- a/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java +++ b/test/unit/org/apache/cassandra/service/accord/async/AsyncOperationTest.java @@ -350,7 +350,7 @@ public class AsyncOperationTest Gen 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 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 diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java index 4ee49b24b9..4238f8a687 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/DepsSerializerTest.java @@ -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()); + } } \ No newline at end of file