From ddfba19f9e547b102a54a700a0f99bcae565d06b Mon Sep 17 00:00:00 2001 From: Shruti Sekaran Date: Fri, 20 Feb 2026 20:08:10 -0800 Subject: [PATCH] Randomize Memtable type/allocation type and SSTable format in Simulator tests patch by Shruti Sekaran; reviewed by Caleb Rackliffe and Francisco Guerrero for CASSANDRA-21177 --- .../config/CassandraRelevantProperties.java | 1 + .../simulator/ClusterSimulation.java | 114 +++++++++++++++++- .../SimulatedFutureActionScheduler.java | 7 ++ .../test/ShortPaxosSimulationTest.java | 6 +- .../simulator/test/SimulationTestBase.java | 71 +++++++++-- .../simulator/test/TrivialSimulationTest.java | 7 +- 6 files changed, 186 insertions(+), 20 deletions(-) diff --git a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java index f8b983ba74..0c81d63afb 100644 --- a/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java +++ b/src/java/org/apache/cassandra/config/CassandraRelevantProperties.java @@ -480,6 +480,7 @@ public enum CassandraRelevantProperties SERIALIZATION_EMPTY_TYPE_NONEMPTY_BEHAVIOR("cassandra.serialization.emptytype.nonempty_behavior"), SET_SEP_THREAD_NAME("cassandra.set_sep_thread_name", "true"), SHUTDOWN_ANNOUNCE_DELAY_IN_MS("cassandra.shutdown_announce_in_ms", "2000"), + SIMULATOR_ITERATIONS("simulator.iterations", "3"), SIZE_RECORDER_INTERVAL("cassandra.size_recorder_interval", "300"), /** diff --git a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java index f7f9346385..1bc900a13f 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java +++ b/test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java @@ -24,6 +24,7 @@ import java.lang.reflect.Modifier; import java.util.ArrayList; import java.util.Arrays; import java.util.EnumMap; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.TreeMap; @@ -38,7 +39,11 @@ import java.util.function.Supplier; import com.google.common.util.concurrent.AsyncFunction; import com.google.common.util.concurrent.FutureCallback; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import org.apache.cassandra.concurrent.ExecutorFactory; +import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.config.ParameterizedClass; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.Feature; @@ -112,6 +117,8 @@ import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; @SuppressWarnings("RedundantCast") public class ClusterSimulation implements AutoCloseable { + private static final Logger logger = LoggerFactory.getLogger(ClusterSimulation.class); + public static final Class[] SHARE = new Class[] { AsyncFunction.class, @@ -188,6 +195,9 @@ public class ClusterSimulation implements AutoCloseable protected HeapPool.Logged.Listener memoryListener; protected SimulatedTime.Listener timeListener = (i1, i2) -> {}; protected LongConsumer onThreadLocalRandomCheck; + protected String memtableType = null; + protected String memtableAllocationType = null; + protected String sstableFormat = null; public Debug debug() { @@ -516,6 +526,24 @@ public class ClusterSimulation implements AutoCloseable return this; } + public Builder memtableType(String type) + { + this.memtableType = type; + return this; + } + + public Builder memtableAllocationType(String type) + { + this.memtableAllocationType = type; + return this; + } + + public Builder sstableFormat(String format) + { + this.sstableFormat = format; + return this; + } + public abstract ClusterSimulation create(long seed) throws IOException; } @@ -654,8 +682,68 @@ public class ClusterSimulation implements AutoCloseable execution = new SimulatedExecution(); + // Track randomized configuration for consolidated logging + Map randomizedConfig = new LinkedHashMap<>(); + randomizedConfig.put("nodes", String.valueOf(numOfNodes)); + randomizedConfig.put("dcs", String.valueOf(numOfDcs)); + + // Log replication factors + StringBuilder rfString = new StringBuilder(); + for (int i = 0; i < numOfDcs; ++i) + { + if (i > 0) + rfString.append(","); + rfString.append("dc").append(i).append(":").append(initialRf[i]); + } + randomizedConfig.put("replication_factors", rfString.toString()); + + // Randomize memtable type + String memtableType; + if (builder.memtableType != null) + { + memtableType = builder.memtableType; + } + else + { + String[] memtableTypes = {"TrieMemtable", "SkipListMemtable"}; + memtableType = memtableTypes[random.uniform(0, memtableTypes.length)]; + } + randomizedConfig.put("memtable", memtableType); + + // Randomize memtable allocation type (heap-based only to avoid InterruptibleChannel issues with offheap) + String memtableAllocationType; + if (builder.memtableAllocationType != null) + { + memtableAllocationType = builder.memtableAllocationType; + } + else + { + String[] allocationTypes = { + "heap_buffers", // Slab allocator (pooled memory) + "unslabbed_heap_buffers" // Direct heap allocation (no pooling) + }; + memtableAllocationType = allocationTypes[random.uniform(0, allocationTypes.length)]; + } + randomizedConfig.put("memtable_allocation_type", memtableAllocationType); + + // Randomize SSTable format + String sstableFormat; + if (builder.sstableFormat != null) + { + sstableFormat = builder.sstableFormat; + } + else + { + String[] formats = {"big", "bti"}; + sstableFormat = formats[random.uniform(0, formats.length)]; + } + randomizedConfig.put("sstable_format", sstableFormat); + CassandraRelevantProperties.SSTABLE_FORMAT_DEFAULT.setString(sstableFormat); + KindOfSequence kindOfDriftSequence = Choices.uniform(KindOfSequence.values()).choose(random); KindOfSequence kindOfDiscontinuitySequence = Choices.uniform(KindOfSequence.values()).choose(random); + randomizedConfig.put("clock_drift_sequence", kindOfDriftSequence.toString()); + randomizedConfig.put("clock_discontinuity_sequence", kindOfDiscontinuitySequence.toString()); time = new SimulatedTime(numOfNodes, random, 1577836800000L /*Jan 1st UTC*/, builder.clockDriftNanos, kindOfDriftSequence, kindOfDiscontinuitySequence.period(builder.clockDiscontinuitIntervalNanos, random), builder.timeListener); @@ -672,6 +760,7 @@ public class ClusterSimulation implements AutoCloseable ThreadAllocator threadAllocator = new ThreadAllocator(random, builder.threadCount, numOfNodes); List allowedDiskAccessModes = Arrays.asList("mmap", "mmap_index_only", "standard"); String disk_access_mode = allowedDiskAccessModes.get(random.uniform(0, allowedDiskAccessModes.size() - 1)); + randomizedConfig.put("disk_access_mode", disk_access_mode); boolean commitlogCompressed = random.decide(.5f); cluster = snitch.setup(Cluster.build(numOfNodes) .withRoot(fs.getPath("/cassandra")) @@ -683,13 +772,26 @@ public class ClusterSimulation implements AutoCloseable .set("cas_contention_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.contentionTimeoutNanos))) .set("request_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.requestTimeoutNanos))) .set("memtable_heap_space", "1MiB") - .set("memtable_allocation_type", builder.memoryListener != null ? "unslabbed_heap_buffers_logged" : "heap_buffers") + .set("memtable_allocation_type", builder.memoryListener != null ? "unslabbed_heap_buffers_logged" : memtableAllocationType) .set("file_cache_size", "16MiB") .set("use_deterministic_table_id", true) .set("disk_access_mode", disk_access_mode) .set("failure_detector", SimulatedFailureDetector.Instance.class.getName()) .set("commitlog_sync", "batch"); + if (memtableType.equals("TrieMemtable")) + { + config.set("memtable", Map.of( + "configurations", Map.of( + "default", Map.of("class_name", "TrieMemtable")))); + } + else + { + config.set("memtable", Map.of( + "configurations", Map.of( + "default", Map.of("class_name", "SkipListMemtable")))); + } + // TODO: Add remove() to IInstanceConfig if (config instanceof InstanceConfig) { @@ -779,10 +881,18 @@ public class ClusterSimulation implements AutoCloseable simulated.register(futureActionScheduler); RunnableActionScheduler scheduler = builder.schedulerFactory.create(random); - ClusterActions.Options options = new ClusterActions.Options(builder.topologyChangeLimit, Choices.uniform(KindOfSequence.values()).choose(random).period(builder.topologyChangeIntervalNanos, random), + KindOfSequence topologyChangeSequence = Choices.uniform(KindOfSequence.values()).choose(random); + ClusterActions.Options options = new ClusterActions.Options(builder.topologyChangeLimit, topologyChangeSequence.period(builder.topologyChangeIntervalNanos, random), Choices.random(random, builder.topologyChanges), minRf, initialRf, maxRf, null); simulation = factory.create(simulated, scheduler, cluster, options); + + // Add remaining randomization tracking + randomizedConfig.put("network_scheduler", futureActionScheduler.getKind().toString()); + randomizedConfig.put("runnable_scheduler", scheduler.getClass().getSimpleName()); + randomizedConfig.put("topology_change_sequence", topologyChangeSequence.toString()); + + logger.warn("Seed 0x{} - Randomized config: {}", Long.toHexString(seed), randomizedConfig); } public synchronized void close() throws IOException diff --git a/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedFutureActionScheduler.java b/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedFutureActionScheduler.java index f66999bb5a..99ebda73c7 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedFutureActionScheduler.java +++ b/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedFutureActionScheduler.java @@ -67,6 +67,7 @@ public class SimulatedFutureActionScheduler implements FutureActionScheduler, To final int nodeCount; final RandomSource random; final SimulatedTime time; + final KindOfSequence kind; // TODO (feature): should we produce more than two simultaneous partitions? final BitSet isInDropPartition = new BitSet(); @@ -86,6 +87,7 @@ public class SimulatedFutureActionScheduler implements FutureActionScheduler, To public SimulatedFutureActionScheduler(KindOfSequence kind, int nodeCount, RandomSource random, SimulatedTime time, NetworkConfig network, SchedulerConfig scheduler) { + this.kind = kind; this.nodeCount = nodeCount; this.random = random; this.time = time; @@ -196,4 +198,9 @@ public class SimulatedFutureActionScheduler implements FutureActionScheduler, To if (oldTopology == null || (newTopology.quorumRf < oldTopology.quorumRf && newTopology.quorumRf < isInDropPartition.cardinality())) recompute(); } + + public KindOfSequence getKind() + { + return kind; + } } diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/ShortPaxosSimulationTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/ShortPaxosSimulationTest.java index f195f1b12d..006d39f8f7 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/ShortPaxosSimulationTest.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/ShortPaxosSimulationTest.java @@ -25,19 +25,21 @@ import org.junit.Test; import org.apache.cassandra.simulator.paxos.PaxosSimulationRunner; +import static org.apache.cassandra.simulator.test.SimulationTestBase.DEFAULT_ITERATIONS; + public class ShortPaxosSimulationTest { @Test public void simulationTest() throws IOException { - PaxosSimulationRunner.main(new String[] { "run", "-n", "3..6", "-t", "1000", "-c", "2", "--cluster-action-limit", "2", "-s", "30" }); + PaxosSimulationRunner.main(new String[] { "run", "-n", "3..6", "-t", "1000", "-c", "2", "--cluster-action-limit", "2", "-s", "30", "--simulations", String.valueOf(DEFAULT_ITERATIONS) }); } @Test @Ignore("fails due to OOM DirectMemory - unclear why") public void selfReconcileTest() throws IOException { - PaxosSimulationRunner.main(new String[] { "reconcile", "-n", "3..6", "-t", "1000", "-c", "2", "--cluster-action-limit", "2", "-s", "30", "--with-self" }); + PaxosSimulationRunner.main(new String[] { "reconcile", "-n", "3..6", "-t", "1000", "-c", "2", "--cluster-action-limit", "2", "-s", "30", "--with-self", "--simulations", String.valueOf(DEFAULT_ITERATIONS) }); } } diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/SimulationTestBase.java b/test/simulator/test/org/apache/cassandra/simulator/test/SimulationTestBase.java index 4cceb8fa3e..682fe9f167 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/SimulationTestBase.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/SimulationTestBase.java @@ -28,6 +28,9 @@ import java.util.function.Predicate; import com.google.common.collect.Iterators; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + import org.apache.cassandra.concurrent.ExecutorFactory; import org.apache.cassandra.distributed.Cluster; import org.apache.cassandra.distributed.api.ConsistencyLevel; @@ -64,6 +67,7 @@ import org.apache.cassandra.utils.CloseableIterator; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; import static java.util.concurrent.TimeUnit.SECONDS; +import static org.apache.cassandra.config.CassandraRelevantProperties.SIMULATOR_ITERATIONS; import static org.apache.cassandra.simulator.ActionSchedule.Mode.STREAM_LIMITED; import static org.apache.cassandra.simulator.ActionSchedule.Mode.UNLIMITED; import static org.apache.cassandra.simulator.ClusterSimulation.ISOLATE; @@ -76,6 +80,9 @@ import static org.apache.cassandra.utils.Shared.Scope.SIMULATION; public class SimulationTestBase { + private static final Logger logger = LoggerFactory.getLogger(SimulationTestBase.class); + public static final int DEFAULT_ITERATIONS = SIMULATOR_ITERATIONS.getInt(); + static abstract class DTestClusterSimulation implements Simulation { final SimulatedSystems simulated; @@ -140,15 +147,23 @@ public class SimulationTestBase public static void simulate(Function init, Function test, Consumer> configure) throws IOException + { + simulate(init, test, configure, 1); + } + + public static void simulate(Function init, + Function test, + Consumer> configure, + int iterations) throws IOException { SimulationRunner.beforeAll(); long seed = System.currentTimeMillis(); - RandomSource random = new RandomSource.Default(); - random.reset(seed); class Factory extends ClusterSimulation.Builder { public ClusterSimulation create(long seed) throws IOException { + RandomSource random = new RandomSource.Default(); + random.reset(seed); return new ClusterSimulation<>(random, seed, 1, this, (c) -> {}, (simulated, scheduler, cluster, options) -> new DTestClusterSimulation(simulated, scheduler, cluster) { @@ -168,16 +183,21 @@ public class SimulationTestBase Factory factory = new Factory(); configure.accept(factory); - try (ClusterSimulation cluster = factory.create(seed)) + for (int i = 0; i < iterations; i++) { - try + long currentSeed = seed + i; + logger.info("Running iteration {} of {} with seed {}L", i + 1, iterations, currentSeed); + try (ClusterSimulation cluster = factory.create(currentSeed)) { - cluster.simulation.run(); - } - catch (Throwable t) - { - throw new AssertionError(String.format("Failed on seed %s", Long.toHexString(seed)), - t); + try + { + cluster.simulation.run(); + } + catch (Throwable t) + { + throw new AssertionError(String.format("Failed on seed 0x%s (base seed 0x%s + %d)", + Long.toHexString(currentSeed), Long.toHexString(seed), i), t); + } } } } @@ -185,16 +205,41 @@ public class SimulationTestBase public static void simulate(IIsolatedExecutor.SerializableRunnable run, IIsolatedExecutor.SerializableRunnable check) { - simulate(new IIsolatedExecutor.SerializableRunnable[]{run}, - check); + simulate(new IIsolatedExecutor.SerializableRunnable[]{run}, check, 1); + } + + public static void simulate(IIsolatedExecutor.SerializableRunnable run, + IIsolatedExecutor.SerializableRunnable check, + int iterations) + { + simulate(new IIsolatedExecutor.SerializableRunnable[]{run}, check, iterations); } public static void simulate(IIsolatedExecutor.SerializableRunnable[] runnables, IIsolatedExecutor.SerializableRunnable check) + { + simulate(runnables, check, 1); + } + + public static void simulate(IIsolatedExecutor.SerializableRunnable[] runnables, + IIsolatedExecutor.SerializableRunnable check, + int iterations) + { + long seed = System.currentTimeMillis(); + for (int i = 0; i < iterations; i++) + { + long currentSeed = seed + i; + logger.info("Running iteration {} of {} with seed {}L", i + 1, iterations, currentSeed); + simulate(runnables, check, currentSeed); + } + } + + public static void simulate(IIsolatedExecutor.SerializableRunnable[] runnables, + IIsolatedExecutor.SerializableRunnable check, + long seed) { Failures failures = new Failures(); RandomSource random = new RandomSource.Default(); - long seed = System.currentTimeMillis(); System.out.println("Using seed: " + seed); random.reset(seed); SimulatedTime time = new SimulatedTime(1, random, 1577836800000L /*Jan 1st UTC*/, new LongRange(1, 100, MILLISECONDS, NANOSECONDS), diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/TrivialSimulationTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/TrivialSimulationTest.java index cf8fe1b25f..1872bed12f 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/TrivialSimulationTest.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/TrivialSimulationTest.java @@ -55,7 +55,8 @@ public class TrivialSimulationTest extends SimulationTestBase (config) -> config .threadCount(10) .nodes(3, 3) - .dcs(1, 1)); + .dcs(1, 1), + DEFAULT_ITERATIONS); } @Test @@ -80,13 +81,13 @@ public class TrivialSimulationTest extends SimulationTestBase }); } }), - () -> {}); + () -> {}, DEFAULT_ITERATIONS); } @Test public void identityHashMapTest() { simulate(arr(() -> new IdentityHashMap<>().put(1, 1)), - () -> {}); + () -> {}, DEFAULT_ITERATIONS); } }