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

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
This commit is contained in:
Shruti Sekaran 2026-02-20 20:08:10 -08:00 committed by Caleb Rackliffe
parent b5b1c529f7
commit ddfba19f9e
6 changed files with 186 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/java/org/apache/cassandra/config/CassandraRelevantProperties.java
View File

@ -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"),
/**

114
test/simulator/main/org/apache/cassandra/simulator/ClusterSimulation.java
View File

@ -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<S extends Simulation> 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<S extends Simulation> 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<S extends Simulation> implements AutoCloseable
return this;
}
public Builder<S> memtableType(String type)
{
this.memtableType = type;
return this;
}
public Builder<S> memtableAllocationType(String type)
{
this.memtableAllocationType = type;
return this;
}
public Builder<S> sstableFormat(String format)
{
this.sstableFormat = format;
return this;
}
public abstract ClusterSimulation<S> create(long seed) throws IOException;
}
@ -654,8 +682,68 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
execution = new SimulatedExecution();
// Track randomized configuration for consolidated logging
Map<String, String> 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<S extends Simulation> implements AutoCloseable
ThreadAllocator threadAllocator = new ThreadAllocator(random, builder.threadCount, numOfNodes);
List<String> 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<S extends Simulation> 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<S extends Simulation> 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

7
test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedFutureActionScheduler.java
View File

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

6
test/simulator/test/org/apache/cassandra/simulator/test/ShortPaxosSimulationTest.java
View File

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

71
test/simulator/test/org/apache/cassandra/simulator/test/SimulationTestBase.java
View File

@ -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<DTestClusterSimulation, ActionList> init,
Function<DTestClusterSimulation, ActionList> test,
Consumer<ClusterSimulation.Builder<DTestClusterSimulation>> configure) throws IOException
{
simulate(init, test, configure, 1);
}
public static void simulate(Function<DTestClusterSimulation, ActionList> init,
Function<DTestClusterSimulation, ActionList> test,
Consumer<ClusterSimulation.Builder<DTestClusterSimulation>> configure,
int iterations) throws IOException
{
SimulationRunner.beforeAll();
long seed = System.currentTimeMillis();
RandomSource random = new RandomSource.Default();
random.reset(seed);
class Factory extends ClusterSimulation.Builder<DTestClusterSimulation>
{
public ClusterSimulation<DTestClusterSimulation> 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),

7
test/simulator/test/org/apache/cassandra/simulator/test/TrivialSimulationTest.java
View File

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