diff --git a/modules/accord b/modules/accord index aa08b04e82..555337a7d4 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit aa08b04e8235d9c0b7e2bbe3bbe8773184bbbe46 +Subproject commit 555337a7d41158f74033818facf94fed6904bf5a diff --git a/src/java/org/apache/cassandra/service/accord/AccordCache.java b/src/java/org/apache/cassandra/service/accord/AccordCache.java index 704ed31881..b7340d3a66 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCache.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCache.java @@ -1132,6 +1132,9 @@ public class AccordCache implements CacheSize if (value.isEmpty()) return null; + if (value.isLoadingPruned()) + return value; + return Serialize.toBytesWithoutKey(value.maximalPrune()); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index db64d55da2..72bca48480 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -224,15 +224,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier public Command loadCommand(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore) { Builder builder = load(commandStoreId, txnId); - Cleanup cleanup = builder.maybeCleanup(true, FULL, redundantBefore, durableBefore); - switch (cleanup) - { - case ERASE: - return Command.Truncated.erased(txnId); - case EXPUNGE: - return null; - } - + builder.maybeCleanup(true, FULL, redundantBefore, durableBefore); return builder.construct(redundantBefore); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java index 66ae868231..cf9e5db835 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java @@ -55,6 +55,7 @@ import org.apache.cassandra.service.accord.serializers.KeySerializers; import org.apache.cassandra.service.accord.serializers.TopologySerializers; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.utils.CollectionSerializers; +import org.apache.cassandra.utils.NoSpamLogger; /** * Receives information about closed, retired ranges, and about sync completion, and @@ -66,6 +67,7 @@ import org.apache.cassandra.utils.CollectionSerializers; public class AccordSyncPropagator { private static final Logger logger = LoggerFactory.getLogger(AccordSyncPropagator.class); + private static final NoSpamLogger noSpamLogger = NoSpamLogger.getLogger(logger, 1L, TimeUnit.MINUTES); public static final IVerbHandler verbHandler = message -> { if (!AccordService.isSetup()) @@ -352,7 +354,7 @@ public class AccordSyncPropagator cb.onResponse(msg.responseWith(SimpleReply.Ok)); return true; } - logger.warn("Node{} is not alive, unable to notify of {}", to, notification); + noSpamLogger.warn("Node{} is not alive, unable to notify of {}", to, notification); scheduler.schedule(() -> notify(to, notification), 1, TimeUnit.MINUTES); return false; } diff --git a/src/java/org/apache/cassandra/service/accord/JournalKey.java b/src/java/org/apache/cassandra/service/accord/JournalKey.java index b299f91613..9bd42ebaf1 100644 --- a/src/java/org/apache/cassandra/service/accord/JournalKey.java +++ b/src/java/org/apache/cassandra/service/accord/JournalKey.java @@ -234,7 +234,8 @@ public final class JournalKey boolean equals(JournalKey other) { - return this.id.equals(other.id) && + return other != null && + this.id.equals(other.id) && this.type == other.type && this.commandStoreId == other.commandStoreId; } diff --git a/src/java/org/apache/cassandra/utils/btree/BTree.java b/src/java/org/apache/cassandra/utils/btree/BTree.java index 201554512c..08372f26a5 100644 --- a/src/java/org/apache/cassandra/utils/btree/BTree.java +++ b/src/java/org/apache/cassandra/utils/btree/BTree.java @@ -4029,6 +4029,12 @@ public class BTree if (leaf().count == 0 && upos == 0) { int prev = 0; + if (!remove.hasNext() || comparator.compare((K)unode[usz - 1], remove.peek()) < 0) + { + // short-circuit common case of removal not intersecting the current node, by comparing with last key + upos = usz; + } + while (upos < usz) { // fast path - buffer is empty and input unconsumed, so may be able to propagate original diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java index 2a093887b6..66e188df5c 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordLoadTest.java @@ -70,8 +70,8 @@ public class AccordLoadTest extends AccordTestBase // AccordTestBase.setupCluster(builder -> builder, 3); AccordTestBase.setupCluster(builder -> builder.withConfig(config -> config .with(Feature.NETWORK, Feature.GOSSIP) - .set("accord.shard_durability_target_splits", "64") - .set("accord.shard_durability_cycle", "5m") + .set("accord.shard_durability_target_splits", "32") + .set("accord.shard_durability_cycle", "1m") // .set("accord.ephemeral_read_enabled", "true") .set("accord.gc_delay", "30s")), 3); } @@ -103,8 +103,9 @@ public class AccordLoadTest extends AccordTestBase // final int flushInterval = 50_000; final int flushInterval = 500; final int compactionPeriodSeconds = 0; -// final int restartInterval = 100_000; - final int restartInterval = Integer.MAX_VALUE; + int restartInterval = 30_000; + final int restartDecay = 2; +// final int restartInterval = Integer.MAX_VALUE; final int batchSizeLimit = 200; final long batchTime = TimeUnit.SECONDS.toNanos(10); final int concurrency = 100; @@ -176,7 +177,8 @@ public class AccordLoadTest extends AccordTestBase int index = 1 + random.nextInt(cluster.size()); logger.info("Picking new coordinator ... {}", index); coordinator = cluster.coordinator(index); - break; + if (cluster.get(index).callOnInstance(() -> AccordService.started())) + break; } catch (Throwable t) { logger.info("Failed to select coordinator", t); } } @@ -211,7 +213,8 @@ public class AccordLoadTest extends AccordTestBase try { i.runOnInstance(() -> { - ((AccordService) AccordService.instance()).journal().closeCurrentSegmentForTestingIfNonEmpty(); + if (AccordService.started()) + ((AccordService) AccordService.instance()).journal().closeCurrentSegmentForTestingIfNonEmpty(); }); } catch (Throwable t) @@ -224,6 +227,7 @@ public class AccordLoadTest extends AccordTestBase if ((nextRestartAt -= batchSize) <= 0) { nextRestartAt += restartInterval; + restartInterval = Math.max(restartInterval, restartInterval * restartDecay); int nodeIdx = 1 + random.nextInt(cluster.size()); restartExecutor.submit(() -> { System.out.printf("restarting node %d...\n", nodeIdx); @@ -231,6 +235,8 @@ public class AccordLoadTest extends AccordTestBase { cluster.get(nodeIdx).shutdown().get(); cluster.get(nodeIdx).startup(); + while (!cluster.get(nodeIdx).callOnInstance(() -> AccordService.started())) + Thread.sleep(1000); return null; } catch (InterruptedException | ExecutionException e) diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java index 50cc6c65de..2101ba8d14 100644 --- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java +++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java @@ -24,7 +24,10 @@ import java.util.ArrayList; import java.util.Collection; import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Objects; import java.util.Set; +import java.util.TreeMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import javax.annotation.Nullable; @@ -39,6 +42,8 @@ import accord.burn.SimulationException; import accord.impl.TopologyFactory; import accord.impl.basic.Cluster; import accord.impl.basic.RandomDelayQueue; +import accord.local.Command; +import accord.local.CommandStore; import accord.local.CommandStores; import accord.local.DurableBefore; import accord.local.Node; @@ -63,6 +68,7 @@ import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.io.sstable.ISSTableScanner; import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.File; +import org.apache.cassandra.journal.Journal; import org.apache.cassandra.journal.SegmentCompactor; import org.apache.cassandra.journal.StaticSegment; import org.apache.cassandra.journal.TestParams; @@ -78,6 +84,7 @@ import org.apache.cassandra.service.accord.serializers.ResultSerializers; import org.apache.cassandra.service.accord.serializers.TopologySerializers; import org.apache.cassandra.service.accord.serializers.Version; import org.apache.cassandra.tools.FieldUtil; +import org.apache.cassandra.utils.CloseableIterator; import static accord.impl.PrefixedIntHashKey.ranges; import static org.apache.cassandra.utils.TimeUUID.Generator.nextTimeUUID; @@ -129,7 +136,7 @@ public class AccordJournalBurnTest extends BurnTestBase public void testOne() { long seed = System.nanoTime(); - int operations = 5000; + int operations = 1000; logger.info("Seed: {}", seed); Cluster.trace.trace("Seed: {}", seed); @@ -243,9 +250,11 @@ public class AccordJournalBurnTest extends BurnTestBase return new DefaultCompactionWriter(cfs, directories, transaction, nonExpiredSSTables, false, 0); } + int counter; @Override public void purge(CommandStores commandStores, EpochSupplier minEpoch) { + ++counter; this.journal.closeCurrentSegmentForTestingIfNonEmpty(); this.journal.runCompactorForTesting(); @@ -272,6 +281,7 @@ public class AccordJournalBurnTest extends BurnTestBase return; List scanners = selected.stream().map(SSTableReader::getScanner).collect(Collectors.toList()); + TreeMap before = read(commandStores); Collection newSStables; try (LifecycleTransaction txn = cfs.getTracker().tryModify(selected, OperationType.COMPACTION); CompactionController controller = new CompactionController(cfs, selected, 0); @@ -298,10 +308,45 @@ public class AccordJournalBurnTest extends BurnTestBase throw new RuntimeException(e); } } - + TreeMap after = read(commandStores); + for (Map.Entry e : before.entrySet()) + { + Command b = e.getValue(); + Command a = after.get(e.getKey()); + Invariants.require(Objects.equals(a, b)); + } + if (before.size() != after.size()) + { + for (Map.Entry e : after.entrySet()) + Invariants.require(null != before.get(e.getKey())); + Invariants.require(false); + } Invariants.require(!orig.equals(cfs.getLiveSSTables())); } + private TreeMap read(CommandStores commandStores) + { + TreeMap result = new TreeMap<>(JournalKey.SUPPORT::compare); + try (CloseableIterator> iter = journalTable.keyIterator()) + { + JournalKey prev = null; + while (iter.hasNext()) + { + Journal.KeyRefs ref = iter.next(); + if (ref.key().type != JournalKey.Type.COMMAND_DIFF) + continue; + + JournalKey key = ref.key(); + if (key.equals(prev)) continue; + CommandStore commandStore = commandStores.forId(ref.key().commandStoreId); + Command command = loadCommand(key.commandStoreId, key.id, commandStore.unsafeGetRedundantBefore(), commandStore.durableBefore()); + if (command != null) + result.put(key, command); + prev = key; + } + } + return result; + } @Override public void replay(CommandStores commandStores)