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

Merge branch 'cassandra-5.0' into trunk 

* cassandra-5.0:
  Randomize Memtable type/allocation type and SSTable format in Simulator tests
This commit is contained in:
Caleb Rackliffe 2026-03-11 16:06:29 -05:00
parent cdb67f62ff ddfba19f9e
commit 87d79f30a0
6 changed files with 189 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -528,6 +528,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"),
SIMULATOR_SEED("cassandra.simulator.seed"),
SIMULATOR_STARTED("cassandra.simulator.started"),
SIZE_RECORDER_INTERVAL("cassandra.size_recorder_interval", "300"),
@ -559,7 +560,6 @@ public enum CassandraRelevantProperties
SNAPSHOT_MIN_ALLOWED_TTL_SECONDS("cassandra.snapshot.min_allowed_ttl_seconds", "60"),
SSL_ENABLE("ssl.enable"),
SSL_STORAGE_PORT("cassandra.ssl_storage_port"),
SSTABLE_FORMAT_DEFAULT("cassandra.sstable.format.default"),
START_GOSSIP("cassandra.start_gossip", "true"),
START_NATIVE_TRANSPORT("cassandra.start_native_transport"),
STORAGE_DIR("cassandra.storagedir"),

117
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.Collections;
import java.util.EnumMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
@ -38,6 +39,9 @@ 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.auth.PasswordSaltSupplier;
import org.apache.cassandra.concurrent.ExecutorFactory;
import org.apache.cassandra.config.CassandraRelevantProperties;
@ -121,6 +125,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);
static
{
CassandraRelevantProperties.TEST_STORAGE_COMPATIBILITY_MODE.setEnum(StorageCompatibilityMode.NONE);
@ -220,6 +226,9 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
protected String transactionalMode = "off";
protected FutureActionSchedulerFactory futureActionSchedulerFactory;
protected PerVerbFutureActionSchedulersFactory perVerbFutureActionSchedulersFactory;
protected String memtableType = null;
protected String memtableAllocationType = null;
protected String sstableFormat = null;
public Builder<S> failures(Failures failures)
{
@ -623,6 +632,24 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
return TransactionalMode.fromString(transactionalMode);
}
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;
}
@ -767,8 +794,67 @@ 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);
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);
@ -800,7 +886,6 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
Failures failures = builder.failures;
ThreadAllocator threadAllocator = new ThreadAllocator(random, builder.threadCount, numOfNodes);
cluster = snitch.setup(Cluster.build(numOfNodes)
.withRoot(fs.getPath("/cassandra"))
.withSharedClasses(sharedClassPredicate)
@ -812,7 +897,7 @@ 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("accord.queue_submission_model", "ASYNC")
@ -822,7 +907,22 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
.set("commitlog_compression", new ParameterizedClass(LZ4Compressor.class.getName(), emptyMap()))
.set("commitlog_sync", "batch")
.set("accord.journal.flush_mode", "BATCH")
.set("accord.command_store_shard_count", "4");
.set("accord.command_store_shard_count", "4")
.set("sstable", Map.of("selected_format", sstableFormat));
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)
{
@ -921,13 +1021,22 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
scheduler = builder.schedulerFactory.create(random);
// TODO (required): we aren't passing paxos variant change parameter anymore
options = new ClusterActions.Options(builder.topologyChangeLimit, Choices.uniform(KindOfSequence.values()).choose(random).period(builder.topologyChangeIntervalNanos, random),
KindOfSequence topologyChangeSequence = Choices.uniform(KindOfSequence.values()).choose(random);
options = new ClusterActions.Options(builder.topologyChangeLimit, topologyChangeSequence.period(builder.topologyChangeIntervalNanos, random),
Choices.random(random, builder.topologyChanges),
builder.consensusChangeLimit, Choices.uniform(KindOfSequence.values()).choose(random).period(builder.consensusChangeIntervalNanos, random),
Choices.random(random, builder.consensusChanges),
minRf, initialRf, maxRf, null);
this.factory = factory;
// Add remaining randomization tracking
if (futureActionScheduler instanceof SimulatedFutureActionScheduler)
randomizedConfig.put("network_scheduler", ((SimulatedFutureActionScheduler) 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);
// during cluster shutdown ignore all failures as there is no reason to track them
onPreShutdown.add(() -> {simulated.failures.ignoreFailures(); return null;});
}

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

@ -64,6 +64,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();
@ -83,6 +84,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;
@ -193,4 +195,9 @@ public class SimulatedFutureActionScheduler implements FutureActionScheduler, To
if (oldTopology == null || (newTopology.quorumRf < oldTopology.quorumRf && newTopology.quorumRf < isInDropPartition.cardinality()))
recompute();
}
public KindOfSequence getKind()
{
return kind;
}
}

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

@ -22,6 +22,8 @@ import org.junit.Test;
import org.apache.cassandra.simulator.paxos.PaxosSimulationRunner;
import static org.apache.cassandra.simulator.test.SimulationTestBase.DEFAULT_ITERATIONS;
/**
* In order to run these tests in your IDE, you need to first build a simulator jara
*
@ -100,7 +102,8 @@ public class ShortPaxosSimulationTest
"-t", "1000",
"-c", "2",
"--cluster-action-limit", "2",
"-s", "30" });
"-s", "30",
"--simulations", String.valueOf(DEFAULT_ITERATIONS) });
}
@Test
@ -114,7 +117,8 @@ public class ShortPaxosSimulationTest
"-s", "30",
"--with-self",
"--with-rng", "0",
"--with-time", "0",});
"--with-time", "0",
"--simulations", String.valueOf(DEFAULT_ITERATIONS) });
}
}

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

@ -84,6 +84,7 @@ import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_GLOBAL;
import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_MONOTONIC_APPROX;
import static org.apache.cassandra.config.CassandraRelevantProperties.CLOCK_MONOTONIC_PRECISE;
import static org.apache.cassandra.config.CassandraRelevantProperties.SIMULATOR_ITERATIONS;
import static org.apache.cassandra.simulator.ActionSchedule.Mode.TIME_LIMITED;
import static org.apache.cassandra.simulator.ActionSchedule.Mode.UNLIMITED;
import static org.apache.cassandra.simulator.ClusterSimulation.ISOLATE;
@ -96,6 +97,8 @@ import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
public class SimulationTestBase
{
public static final int DEFAULT_ITERATIONS = SIMULATOR_ITERATIONS.getInt();
@BeforeClass
public static void beforeAll()
{
@ -251,6 +254,16 @@ public class SimulationTestBase
simulate(init, test, teardown, configure, (i1, i2) -> {});
}
static void simulate(Function<SimpleSimulation, ActionList> init,
Function<SimpleSimulation, ActionList> test,
Function<SimpleSimulation, ActionList> teardown,
Consumer<ClusterSimulation.Builder<SimpleSimulation>> configure,
int iterations) throws IOException
{
simulate(System::currentTimeMillis, new DTestClusterSimulationBuilder(init, test, teardown, (i1, i2) -> {}),
configure, iterations);
}
@SuppressWarnings("unused")
static void simulate(long seed,
Function<SimpleSimulation, ActionList> init,
@ -325,6 +338,14 @@ public class SimulationTestBase
public static <T extends Simulation> void simulate(LongSupplier seedGen,
ClusterSimulation.Builder<T> factory,
Consumer<ClusterSimulation.Builder<T>> configure) throws IOException
{
simulate(seedGen, factory, configure, 1);
}
public static <T extends Simulation> void simulate(LongSupplier seedGen,
ClusterSimulation.Builder<T> factory,
Consumer<ClusterSimulation.Builder<T>> configure,
int iterations) throws IOException
{
SimulationRunner.beforeAll();
long seed = seedGen.getAsLong();
@ -332,36 +353,60 @@ public class SimulationTestBase
//long seed = 1687184561194L;
System.out.printf("Simulation seed: %dL%n", seed);
configure.accept(factory);
try (ClusterSimulation<?> cluster = factory.create(seed))
for (int i = 0; i < iterations; i++)
{
try (Simulation simulation = cluster.simulation())
long currentSeed = seed + i;
System.out.printf("Running iteration %d of %d with seed %dL%n", i + 1, iterations, currentSeed);
try (ClusterSimulation<?> cluster = factory.create(currentSeed))
{
simulation.run();
try (Simulation simulation = cluster.simulation())
{
simulation.run();
}
catch (Throwable t)
{
throw new SimulationException(currentSeed, t);
}
}
catch (Throwable t)
{
throw new SimulationException(seed, t);
if (t instanceof SimulationException)
throw t;
throw new SimulationException(currentSeed, t);
}
}
catch (Throwable t)
{
if (t instanceof SimulationException)
throw t;
throw new SimulationException(seed, t);
}
}
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, System.currentTimeMillis());
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;
System.out.printf("Running iteration %d of %d with seed %dL%n", i + 1, iterations, currentSeed);
simulate(runnables, check, currentSeed);
}
}
public static void simulate(IIsolatedExecutor.SerializableRunnable[] runnables,

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

@ -38,7 +38,7 @@ public class TrivialSimulationTest extends SimulationTestBase
public void identityHashMapTest()
{
simulate(arr(() -> new IdentityHashMap<>().put(1, 1)),
() -> {});
() -> {}, DEFAULT_ITERATIONS);
}
@Test
@ -56,7 +56,8 @@ public class TrivialSimulationTest extends SimulationTestBase
(config) -> config
.threadCount(10)
.nodes(3, 3)
.dcs(1, 1));
.dcs(1, 1),
DEFAULT_ITERATIONS);
}
@Test
@ -81,8 +82,6 @@ public class TrivialSimulationTest extends SimulationTestBase
});
}
}),
() -> {});
() -> {}, DEFAULT_ITERATIONS);
}
}