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

Get simulator working (again) 

Co-authored-by: Ariel Weisberg <aweisberg@apple.com>
Co-authored-by: Benedict Elliott Smith <benedict@apache.org>
This commit is contained in:
David Capwell 2023-10-04 15:13:37 -07:00 committed by David Capwell
parent 8de3dfc711
commit a324003c59
38 changed files with 421 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.build/checkstyle_suppressions.xml
View File

@ -21,5 +21,4 @@
"https://checkstyle.org/dtds/suppressions_1_1.dtd">
<suppressions>
<suppress checks="RegexpSinglelineJava" files="Semaphore\.java"/>
</suppressions>

18
build.xml
View File

@ -226,6 +226,24 @@
<condition property="is.java.default"><equals arg1="${ant.java.version}" arg2="${java.default}"/></condition>
<echo unless:true="${is.java.default}" message="Non default JDK version used: ${ant.java.version}"/>
<condition property="arch_x86">
<equals arg1="${os.arch}" arg2="x86" />
</condition>
<!-- On non-X86 JDK 8 (such as M1 Mac) the smallest allowed Xss is 384k; so need a larger value
when on these platforms. -->
<condition property="jvm_xss" value="-Xss256k" else="-Xss384k">
<isset property="arch_x86" />
</condition>
<condition property="java.version.11">
<!-- This includes every JDK other than 8; so JDK 9 is flagged as JDK 11, and JDK 17 is as well... at the moment this is desired behavior
and may be relooked at once JDK 8 support is dropped -->
<not><isset property="java.version.8"/></not>
</condition>
<fail message="Unsupported JDK version used: ${ant.java.version}"><condition><not><or>
<isset property="java.version.8"/>
<isset property="java.version.11"/>
</or></not></condition></fail>
<condition property="arch_x86">
<equals arg1="${os.arch}" arg2="x86" />
</condition>

5
src/java/org/apache/cassandra/concurrent/InfiniteLoopExecutor.java
View File

@ -52,6 +52,11 @@ public class InfiniteLoopExecutor implements Interruptible
@Shared(scope = Shared.Scope.SIMULATION)
public enum SimulatorSafe { SAFE, UNSAFE }
/**
* Does this loop always block on some external work provision that is going to be simulator-controlled, or does
* it loop periodically? If the latter, it may prevent simulation making progress between phases, and should be
* marked as a DAEMON process.
*/
@Shared(scope = Shared.Scope.SIMULATION)
public enum Daemon { DAEMON, NON_DAEMON }

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

@ -596,6 +596,8 @@ public enum CassandraRelevantProperties
* can be also done manually for that particular case: {@code flush(SchemaConstants.SCHEMA_KEYSPACE_NAME);}. */
TEST_FLUSH_LOCAL_SCHEMA_CHANGES("cassandra.test.flush_local_schema_changes", "true"),
TEST_HARRY_SWITCH_AFTER("cassandra.test.harry.progression.switch-after", "1"),
TEST_HISTORY_VALIDATOR_LOGGING_ENABLED("cassandra.test.history_validator.logging.enabled", "false"),
TEST_IGNORE_SIGAR("cassandra.test.ignore_sigar"),
TEST_INTERVAL_TREE_EXPENSIVE_CHECKS("cassandra.test.interval_tree_expensive_checks"),
TEST_INVALID_LEGACY_SSTABLE_ROOT("invalid-legacy-sstable-root"),
TEST_JVM_DTEST_DISABLE_SSL("cassandra.test.disable_ssl"),

6
src/java/org/apache/cassandra/db/memtable/AbstractAllocatorMemtable.java
View File

@ -220,6 +220,12 @@ public abstract class AbstractAllocatorMemtable extends AbstractMemtableWithComm
if (current instanceof AbstractAllocatorMemtable)
((AbstractAllocatorMemtable) current).flushIfPeriodExpired();
}
@Override
public String toString()
{
return "Scheduled Flush of " + owner;
}
};
ScheduledExecutors.scheduledTasks.scheduleSelfRecurring(runnable, period, TimeUnit.MILLISECONDS);
}

5
src/java/org/apache/cassandra/gms/Gossiper.java
View File

@ -2046,6 +2046,11 @@ public class Gossiper implements IFailureDetectionEventListener, GossiperMBean,
ExecutorUtils.shutdownAndWait(timeout, unit, executor);
}
public void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
{
ExecutorUtils.shutdownAndWait(timeout, unit, executor);
}
@Nullable
private String getReleaseVersionString(InetAddressAndPort ep)
{

10
src/java/org/apache/cassandra/index/IndexStatusManager.java
View File

@ -24,12 +24,13 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -46,7 +47,9 @@ import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.locator.Replica;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.utils.CassandraVersion;
import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.JsonUtils;
@ -335,4 +338,9 @@ public class IndexStatusManager
{
return keyspace + '.' + index;
}
public void shutdownAndWait(long interval, TimeUnit unit) throws InterruptedException, TimeoutException
{
ExecutorUtils.shutdownAndWait(interval, unit, statusPropagationExecutor);
}
}

9
src/java/org/apache/cassandra/journal/ActiveSegment.java
View File

@ -33,6 +33,9 @@ import org.apache.cassandra.utils.concurrent.OpOrder;
import org.apache.cassandra.utils.concurrent.Ref;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.utils.Simulate.With.MONITORS;
@Simulate(with=MONITORS)
final class ActiveSegment<K, V> extends Segment<K, V>
{
final FileChannel channel;
@ -247,6 +250,12 @@ final class ActiveSegment<K, V> extends Segment<K, V>
* Flush logic; closing and component flushing
*/
boolean shouldFlush()
{
int allocatePosition = this.allocatePosition.get();
return lastFlushedOffset < allocatePosition;
}
/**
* Possibly force a disk flush for this segment file.
* TODO FIXME: calls from outside Flusher + callbacks

29
src/java/org/apache/cassandra/journal/Flusher.java
View File

@ -28,6 +28,7 @@ import org.apache.cassandra.concurrent.Interruptible;
import org.apache.cassandra.concurrent.Interruptible.TerminateException;
import org.apache.cassandra.utils.MonotonicClock;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.Simulate;
import org.apache.cassandra.utils.concurrent.Semaphore;
import org.apache.cassandra.utils.concurrent.WaitQueue;
@ -45,6 +46,9 @@ import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
import static org.apache.cassandra.utils.MonotonicClock.Global.preciseTime;
import static org.apache.cassandra.utils.Simulate.With.GLOBAL_CLOCK;
import static org.apache.cassandra.utils.Simulate.With.LOCK_SUPPORT;
import static org.apache.cassandra.utils.Simulate.With.MONITORS;
import static org.apache.cassandra.utils.concurrent.Semaphore.newSemaphore;
import static org.apache.cassandra.utils.concurrent.WaitQueue.newWaitQueue;
@ -96,6 +100,7 @@ final class Flusher<K, V>
flushExecutor.shutdown();
}
@Simulate(with={MONITORS,GLOBAL_CLOCK,LOCK_SUPPORT})
private class FlushRunnable implements Interruptible.Task
{
private final MonotonicClock clock;
@ -151,9 +156,17 @@ final class Flusher<K, V>
if (state == SHUTTING_DOWN)
return;
long wakeUpAt = startedRunAt + flushPeriodNanos();
if (wakeUpAt > now)
haveWork.tryAcquireUntil(1, wakeUpAt);
long flushPeriodNanos = flushPeriodNanos();
if (flushPeriodNanos <= 0)
{
haveWork.acquire(1);
}
else
{
long wakeUpAt = startedRunAt + flushPeriodNanos;
if (wakeUpAt > now)
haveWork.tryAcquireUntil(1, wakeUpAt);
}
}
private void doFlush()
@ -168,6 +181,9 @@ final class Flusher<K, V>
for (ActiveSegment<K, V> segment : segmentsToFlush)
{
if (!segment.shouldFlush())
break;
syncedSegment = segment.descriptor.timestamp;
syncedOffset = segment.flush();
@ -202,8 +218,9 @@ final class Flusher<K, V>
flushCount++;
flushDuration += (finishedFlushAt - startedFlushAt);
long lag = finishedFlushAt - (startedFlushAt + flushPeriodNanos());
if (lag <= 0)
long flushPeriodNanos = flushPeriodNanos();
long lag = finishedFlushAt - (startedFlushAt + flushPeriodNanos);
if (flushPeriodNanos <= 0 || lag <= 0)
return;
lagCount++;
@ -349,7 +366,7 @@ final class Flusher<K, V>
private long flushPeriodNanos()
{
return 1_000_000L * params.flushPeriod();
return 1_000_000L * params.flushPeriodMillis();
}
private long periodicFlushLagBlockNanos()

3
src/java/org/apache/cassandra/journal/Journal.java
View File

@ -50,6 +50,7 @@ import org.apache.cassandra.journal.Segments.ReferencedSegments;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.Crc;
import org.apache.cassandra.utils.JVMStabilityInspector;
import org.apache.cassandra.utils.Simulate;
import org.apache.cassandra.utils.concurrent.WaitQueue;
import static java.lang.String.format;
@ -61,6 +62,7 @@ import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.SimulatorSafe
import static org.apache.cassandra.concurrent.Interruptible.State.NORMAL;
import static org.apache.cassandra.concurrent.Interruptible.State.SHUTTING_DOWN;
import static org.apache.cassandra.utils.Clock.Global.currentTimeMillis;
import static org.apache.cassandra.utils.Simulate.With.MONITORS;
import static org.apache.cassandra.utils.concurrent.WaitQueue.newWaitQueue;
/**
@ -77,6 +79,7 @@ import static org.apache.cassandra.utils.concurrent.WaitQueue.newWaitQueue;
* @param <K> the type of keys used to address the records;
must be fixed-size and byte-order comparable
*/
@Simulate(with=MONITORS)
public class Journal<K, V> implements Shutdownable
{
private static final Logger logger = LoggerFactory.getLogger(Journal.class);

2
src/java/org/apache/cassandra/journal/Params.java
View File

@ -41,7 +41,7 @@ public interface Params
/**
* @return milliseconds between journal flushes
*/
int flushPeriod();
int flushPeriodMillis();
/**
* @return milliseconds to block writes for while waiting for a slow disk flush to complete

6
src/java/org/apache/cassandra/metrics/AccordStateCacheMetrics.java
View File

@ -32,7 +32,7 @@ public class AccordStateCacheMetrics extends CacheAccessMetrics
public final Histogram objectSize;
private final Map<Class<?>, CacheAccessMetrics> instanceMetrics = new ConcurrentHashMap<>(2);
private final Map<String, CacheAccessMetrics> instanceMetrics = new ConcurrentHashMap<>(2);
private final String type;
@ -45,6 +45,8 @@ public class AccordStateCacheMetrics extends CacheAccessMetrics
public CacheAccessMetrics forInstance(Class<?> klass)
{
return instanceMetrics.computeIfAbsent(klass, k -> new CacheAccessMetrics(new DefaultNameFactory(TYPE_NAME, String.format("%s-%s", type, k.getSimpleName()))));
// cannot make Class<?> hashCode deterministic, as cannot rewrite - so cannot safely use as Map key if want deterministic simulation
// (or we need to create extra hoops to catch this specific case in method rewriting)
return instanceMetrics.computeIfAbsent(klass.getSimpleName(), k -> new CacheAccessMetrics(new DefaultNameFactory(TYPE_NAME, String.format("%s-%s", type, k))));
}
}

61
src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
View File

@ -20,13 +20,12 @@ package org.apache.cassandra.service.accord;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Sets;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.impl.AbstractConfigurationService;
import accord.local.Node;
@ -36,6 +35,7 @@ import accord.utils.Invariants;
import accord.utils.async.AsyncResult;
import accord.utils.async.AsyncResults;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.concurrent.Stage;
import org.apache.cassandra.gms.FailureDetector;
import org.apache.cassandra.gms.IFailureDetector;
@ -44,19 +44,23 @@ import org.apache.cassandra.net.MessageDelivery;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.accord.AccordKeyspace.EpochDiskState;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.listeners.ChangeListener;
import org.apache.cassandra.utils.Simulate;
import org.apache.cassandra.utils.concurrent.AsyncPromise;
import org.apache.cassandra.utils.concurrent.Future;
import static org.apache.cassandra.utils.Simulate.With.MONITORS;
// TODO: listen to FailureDetector and rearrange fast path accordingly
public class AccordConfigurationService extends AbstractConfigurationService<AccordConfigurationService.EpochState, AccordConfigurationService.EpochHistory> implements ChangeListener, AccordEndpointMapper, AccordSyncPropagator.Listener
@Simulate(with=MONITORS)
public class AccordConfigurationService extends AbstractConfigurationService<AccordConfigurationService.EpochState, AccordConfigurationService.EpochHistory> implements ChangeListener, AccordEndpointMapper, AccordSyncPropagator.Listener, Shutdownable
{
private static final Logger logger = LoggerFactory.getLogger(AccordConfigurationService.class);
private final AccordSyncPropagator syncPropagator;
private EpochDiskState diskState = EpochDiskState.EMPTY;
private enum State { INITIALIZED, LOADING, STARTED }
private enum State { INITIALIZED, LOADING, STARTED, SHUTDOWN }
private State state = State.INITIALIZED;
private volatile EndpointMapping mapping = EndpointMapping.EMPTY;
@ -150,6 +154,35 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
receiveRedundant(redundant, epoch);
}));
state = State.STARTED;
ClusterMetadataService.instance().log().addListener(this);
}
@Override
public synchronized boolean isTerminated()
{
return state == State.SHUTDOWN;
}
@Override
public synchronized void shutdown()
{
if (isTerminated())
return;
ClusterMetadataService.instance().log().removeListener(this);
state = State.SHUTDOWN;
}
@Override
public Object shutdownNow()
{
shutdown();
return null;
}
@Override
public boolean awaitTermination(long timeout, TimeUnit units) throws InterruptedException
{
return isTerminated();
}
@Override
@ -262,7 +295,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
}
@Override
protected void receiveRemoteSyncCompletePreListenerNotify(Node.Id node, long epoch)
protected synchronized void receiveRemoteSyncCompletePreListenerNotify(Node.Id node, long epoch)
{
if (state == State.STARTED)
diskState = AccordKeyspace.markRemoteTopologySync(node, epoch, diskState);
@ -271,7 +304,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
@Override
public synchronized void reportEpochClosed(Ranges ranges, long epoch)
{
Invariants.checkState(state == State.STARTED);
checkStarted();
Topology topology = getTopologyForEpoch(epoch);
syncPropagator.reportClosed(epoch, topology.nodes(), ranges);
}
@ -279,7 +312,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
@Override
public synchronized void reportEpochRedundant(Ranges ranges, long epoch)
{
Invariants.checkState(state == State.STARTED);
checkStarted();
// TODO (expected): ensure we aren't fetching a truncated epoch; otherwise this should be non-null
Topology topology = getTopologyForEpoch(epoch);
syncPropagator.reportRedundant(epoch, topology.nodes(), ranges);
@ -300,18 +333,24 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
}
@Override
protected void truncateTopologiesPreListenerNotify(long epoch)
protected synchronized void truncateTopologiesPreListenerNotify(long epoch)
{
Invariants.checkState(state == State.STARTED);
checkStarted();
}
@Override
protected void truncateTopologiesPostListenerNotify(long epoch)
protected synchronized void truncateTopologiesPostListenerNotify(long epoch)
{
if (state == State.STARTED)
diskState = AccordKeyspace.truncateTopologyUntil(epoch, diskState);
}
private void checkStarted()
{
State state = this.state;
Invariants.checkState(state == State.STARTED, "Expected state to be STARTED but was %s", state);
}
@VisibleForTesting
public static class EpochSnapshot
{

2
src/java/org/apache/cassandra/service/accord/AccordFastPathCoordinator.java
View File

@ -253,7 +253,7 @@ public abstract class AccordFastPathCoordinator implements ChangeListener, Confi
private void scheduleMaintenanceTask(long delayMillis)
{
ScheduledExecutors.scheduledTasks.schedule(this::maintenance, delayMillis, TimeUnit.MILLISECONDS);
ScheduledExecutors.scheduledTasks.scheduleSelfRecurring(this::maintenance, delayMillis, TimeUnit.MILLISECONDS);
}
synchronized void maintenance()

9
src/java/org/apache/cassandra/service/accord/AccordJournal.java
View File

@ -127,6 +127,7 @@ import static org.apache.cassandra.concurrent.InfiniteLoopExecutor.SimulatorSafe
import static accord.messages.MessageType.STABLE_FAST_PATH_REQ;
import static accord.messages.MessageType.STABLE_MAXIMAL_REQ;
import static accord.messages.MessageType.STABLE_SLOW_PATH_REQ;
import static org.apache.cassandra.concurrent.Interruptible.State.NORMAL;
import static org.apache.cassandra.db.TypeSizes.BYTE_SIZE;
import static org.apache.cassandra.db.TypeSizes.INT_SIZE;
import static org.apache.cassandra.db.TypeSizes.LONG_SIZE;
@ -172,9 +173,9 @@ public class AccordJournal implements Shutdownable
}
@Override
public int flushPeriod()
public int flushPeriodMillis()
{
return 1000;
return DatabaseDescriptor.getCommitLogSyncPeriod();
}
@Override
@ -1098,7 +1099,9 @@ public class AccordJournal implements Shutdownable
{
if (!unframedRequests.isEmpty() || !delayedRequests.isEmpty())
doRun();
haveWork.acquire(1);
if (state == NORMAL)
haveWork.acquire(1);
}
private void doRun()

33
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -26,11 +26,14 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import javax.annotation.Nonnull;
import javax.annotation.concurrent.GuardedBy;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import accord.coordinate.TopologyMismatch;
import org.apache.cassandra.cql3.statements.RequestValidations;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.transformations.AddAccordTable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -91,7 +94,6 @@ import org.apache.cassandra.service.accord.interop.AccordInteropExecution;
import org.apache.cassandra.service.accord.interop.AccordInteropPersist;
import org.apache.cassandra.service.accord.txn.TxnResult;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Epoch;
import org.apache.cassandra.tcm.membership.NodeId;
import org.apache.cassandra.transport.Dispatcher;
@ -117,6 +119,10 @@ public class AccordService implements IAccordService, Shutdownable
{
private static final Logger logger = LoggerFactory.getLogger(AccordService.class);
private enum State { INIT, STARTED, SHUTDOWN}
public static final AccordClientRequestMetrics readMetrics = new AccordClientRequestMetrics("AccordRead");
public static final AccordClientRequestMetrics writeMetrics = new AccordClientRequestMetrics("AccordWrite");
private static final Future<Void> BOOTSTRAP_SUCCESS = ImmediateFuture.success(null);
private final Node node;
@ -129,6 +135,8 @@ public class AccordService implements IAccordService, Shutdownable
private final AccordJournal journal;
private final AccordVerbHandler<? extends Request> verbHandler;
private final LocalConfig configuration;
@GuardedBy("this")
private State state = State.INIT;
private static final IAccordService NOOP_SERVICE = new IAccordService()
{
@ -308,13 +316,16 @@ public class AccordService implements IAccordService, Shutdownable
}
@Override
public void startup()
public synchronized void startup()
{
if (state != State.INIT)
return;
journal.start(node);
configService.start();
ClusterMetadataService.instance().log().addListener(configService);
fastPathCoordinator.start();
ClusterMetadataService.instance().log().addListener(fastPathCoordinator);
state = State.STARTED;
}
@Override
@ -526,15 +537,18 @@ public class AccordService implements IAccordService, Shutdownable
}
@Override
public void shutdown()
public synchronized void shutdown()
{
ExecutorUtils.shutdown(Arrays.asList(scheduler, nodeShutdown, journal));
if (state != State.STARTED)
return;
ExecutorUtils.shutdown(shutdownableSubsystems());
state = State.SHUTDOWN;
}
@Override
public Object shutdownNow()
{
ExecutorUtils.shutdownNow(Arrays.asList(scheduler, nodeShutdown, journal));
shutdown();
return null;
}
@ -543,7 +557,7 @@ public class AccordService implements IAccordService, Shutdownable
{
try
{
ExecutorUtils.awaitTermination(timeout, units, Arrays.asList(scheduler, nodeShutdown, journal));
ExecutorUtils.awaitTermination(timeout, units, shutdownableSubsystems());
return true;
}
catch (TimeoutException e)
@ -552,11 +566,16 @@ public class AccordService implements IAccordService, Shutdownable
}
}
private List<Shutdownable> shutdownableSubsystems()
{
return Arrays.asList(scheduler, nodeShutdown, journal, configService);
}
@VisibleForTesting
@Override
public void shutdownAndWait(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException
{
scheduler.shutdownNow();
shutdown();
ExecutorUtils.shutdownAndWait(timeout, unit, this);
}

3
src/java/org/apache/cassandra/utils/concurrent/Semaphore.java
View File

@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.cassandra.utils.Intercept;
import org.apache.cassandra.utils.Shared;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
@Shared(scope = SIMULATION)
@ -139,7 +140,7 @@ public interface Semaphore
*/
public boolean tryAcquireUntil(int acquire, long nanoTimeDeadline) throws InterruptedException
{
long wait = nanoTimeDeadline - System.nanoTime();
long wait = nanoTimeDeadline - nanoTime();
return tryAcquire(acquire, Math.max(0, wait), TimeUnit.NANOSECONDS);
}

5
test/conf/logback-simulator.xml
View File

@ -19,7 +19,8 @@
<configuration debug="false" scan="true" scanPeriod="60 seconds">
<define name="run_start" class="org.apache.cassandra.simulator.logging.RunStartDefiner" scope="system"/>
<define name="run_seed" class="org.apache.cassandra.simulator.logging.SeedDefiner" scope="system"/>
<define name="instance_id" class="org.apache.cassandra.distributed.impl.InstanceIDDefiner" />
<define name="cluster_id" class="org.apache.cassandra.distributed.impl.ClusterIDDefiner"/>
<define name="instance_id" class="org.apache.cassandra.distributed.impl.InstanceIDDefiner"/>
<!-- Shutdown hook ensures that async appender flushes -->
<shutdownHook class="ch.qos.logback.core.hook.DelayingShutdownHook"/>
@ -38,7 +39,7 @@
</appender>
<appender name="INSTANCEFILE" class="ch.qos.logback.core.FileAppender">
<file>./build/test/logs/simulator/${run_start}-${run_seed}/${instance_id}/system.log</file>
<file>./build/test/logs/simulator/${run_start}-${run_seed}/cluster-${cluster_id}/${instance_id}/system.log</file>
<encoder>
<pattern>%-5level [%thread] ${instance_id} %replace(CS:%X{command_store} ){'CS\:\s+', ''}%replace(OP:%X{async_op} ){'OP\:\s+', ''}%date{ISO8601} %msg%n</pattern>
</encoder>

10
test/distributed/org/apache/cassandra/distributed/impl/Instance.java
View File

@ -102,6 +102,7 @@ import org.apache.cassandra.exceptions.StartupException;
import org.apache.cassandra.gms.Gossiper;
import org.apache.cassandra.hints.DTestSerializer;
import org.apache.cassandra.hints.HintsService;
import org.apache.cassandra.index.IndexStatusManager;
import org.apache.cassandra.index.SecondaryIndexManager;
import org.apache.cassandra.io.IVersionedAsymmetricSerializer;
import org.apache.cassandra.io.sstable.format.SSTableReader;
@ -628,6 +629,9 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
{
assert config.networkTopology().contains(config.broadcastAddress()) : String.format("Network topology %s doesn't contain the address %s",
config.networkTopology(), config.broadcastAddress());
// org.apache.cassandra.distributed.impl.AbstractCluster.startup sets the exception handler for the thread
// so extract it to populate ExecutorFactory.Global
ExecutorFactory.Global.tryUnsafeSet(new ExecutorFactory.Default(Thread.currentThread().getContextClassLoader(), null, Thread.getDefaultUncaughtExceptionHandler()));
DistributedTestInitialLocationProvider.assign(config.networkTopology());
CassandraDaemon.getInstanceForTesting().activate(false);
// TODO: filters won't work for the messages dispatched during startup
@ -927,6 +931,11 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
error = parallelRun(error, executor,
() -> Gossiper.instance.stopShutdownAndWait(1L, MINUTES));
}
else
{
error = parallelRun(error, executor,
() -> Gossiper.instance.shutdownAndWait(1L, MINUTES));
}
error = parallelRun(error, executor, StorageService.instance::disableAutoCompaction);
@ -970,6 +979,7 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
() -> ActiveRepairService.instance().shutdownNowAndWait(1L, MINUTES),
() -> EpochAwareDebounce.instance.close(),
SnapshotManager.instance::close,
() -> IndexStatusManager.instance.shutdownAndWait(1L, MINUTES),
DiskErrorsHandlerService::close
);

2
test/distributed/org/apache/cassandra/distributed/impl/IsolatedExecutor.java
View File

@ -126,7 +126,7 @@ public class IsolatedExecutor implements IIsolatedExecutor
public Future<Void> shutdown()
{
isolatedExecutor.shutdownNow();
isolatedExecutor.shutdown();
return shutdownExecutor.shutdown(name, classLoader, isolatedExecutor, () -> {
// Shutdown logging last - this is not ideal as the logging subsystem is initialized

4
test/simulator/asm/org/apache/cassandra/simulator/asm/ClassTransformer.java
View File

@ -189,6 +189,10 @@ class ClassTransformer extends ClassVisitor implements MethodWriterSink
{
if (dependentTypes != null)
Utils.visitIfRefType(descriptor, dependentTypes);
// org.apache.cassandra.simulator.systems.SimulatedTime.InstanceTime.nanoTime does not change between invokes which causes AbstractQueuedSynchronizer to loop forever,
// so need to make the threshold negative to avoid the spin loop.
if (className.equals("java/util/concurrent/locks/AbstractQueuedSynchronizer") && name.equals("SPIN_FOR_TIMEOUT_THRESHOLD"))
return super.visitField(makePublic(access), name, descriptor, signature, Long.MIN_VALUE);
return super.visitField(makePublic(access), name, descriptor, signature, value);
}

104
test/simulator/asm/org/apache/cassandra/simulator/asm/InterceptAgent.java
View File

@ -30,6 +30,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumSet;
import java.util.List;
import java.util.Objects;
import java.util.function.BiFunction;
import java.util.regex.Pattern;
@ -93,6 +94,9 @@ public class InterceptAgent
if (className.equals("java/lang/Object"))
return transformObject(bytecode);
if (className.equals("java/lang/Class"))
return transformClass(bytecode);
if (className.equals("java/lang/Enum"))
return transformEnum(bytecode);
@ -103,10 +107,14 @@ public class InterceptAgent
return transformThreadLocalRandom(bytecode);
if (className.startsWith("java/util/concurrent/ConcurrentHashMap"))
return transformConcurrent(className, bytecode, DETERMINISTIC, NO_PROXY_METHODS);
return InterceptAgent.transform(className, bytecode, DETERMINISTIC, NO_PROXY_METHODS);
if (className.startsWith("java/util/concurrent/locks"))
return transformConcurrent(className, bytecode, SYSTEM_CLOCK, LOCK_SUPPORT, NO_PROXY_METHODS);
{
if (className.equals("java/util/concurrent/locks/AbstractQueuedSynchronizer"))
return InterceptAgent.transformAbstractQueuedSynchronizer(className, bytecode, SYSTEM_CLOCK, LOCK_SUPPORT, NO_PROXY_METHODS);
return InterceptAgent.transform(className, bytecode, SYSTEM_CLOCK, LOCK_SUPPORT, NO_PROXY_METHODS);
}
return null;
}
@ -172,6 +180,29 @@ public class InterceptAgent
return transform(bytes, ObjectVisitor::new);
}
/**
* We don't want Object.toString() to invoke our overridden identityHashCode by virtue of invoking some overridden hashCode()
* So we overwrite Object.toString() to replace calls to Object.hashCode() with direct calls to System.identityHashCode()
*/
private static byte[] transformClass(byte[] bytes)
{
class ClazzVisitor extends ClassVisitor
{
public ClazzVisitor(int api, ClassVisitor classVisitor)
{
super(api, classVisitor);
}
@Override
public void visitEnd()
{
new StringHashcode(api).accept(this);
super.visitEnd();
}
}
return transform(bytes, ClazzVisitor::new);
}
/**
* We want Enum to have a deterministic hashCode() so we simply forward calls to ordinal()
*/
@ -314,7 +345,7 @@ public class InterceptAgent
else
{
MethodVisitor mv = super.visitMethod(access, name, descriptor, signature, exceptions);
if (determinismCheck && (name.equals("nextSeed") || name.equals("nextSecondarySeed")))
if (determinismCheck && (name.equals("nextSeed") || name.equals("nextSecondarySeed") || name.equals("advanceProbe")))
mv = new ThreadLocalRandomCheckTransformer(api, mv);
return mv;
}
@ -323,7 +354,61 @@ public class InterceptAgent
return transform(bytes, ThreadLocalRandomVisitor::new);
}
private static byte[] transform(byte[] bytes, BiFunction<Integer, ClassWriter, ClassVisitor> constructor)
/**
* We require ThreadLocalRandom to be deterministic, so we modify its initialisation method to invoke a
* global deterministic random value generator
*/
private static byte[] transformAbstractQueuedSynchronizer(String className, byte[] bytes, Flag flag, Flag ... flags)
{
class AbstractQueuedSynchronizerVisitor extends ClassVisitor
{
private long defaultSpinForTimeoutThreshold = 1000L;
public AbstractQueuedSynchronizerVisitor(int api, ClassVisitor classVisitor)
{
super(api, classVisitor);
}
@Override
public FieldVisitor visitField(int access, String name, String descriptor, String signature, Object value)
{
if (name.equals("SPIN_FOR_TIMEOUT_THRESHOLD"))
{
defaultSpinForTimeoutThreshold = (Long)value;
return super.visitField(access, name, descriptor, signature, 0L);
}
return super.visitField(access, name, descriptor, signature, value);
}
@Override
public MethodVisitor visitMethod(int access, String name, String descriptor, String signature, String[] exceptions)
{
/// !!!!! WARNING !!!!!
/// THIS IS SUPER BRITTLE BECAUSE rt.jar INLINES GETSTATIC AS LDC
// TODO (desired): visit constructor to fetch actual value of constant in case changes in future release -
// but this is brittle enough changes upstream will likely need revisiting anyway
MethodVisitor mv = super.visitMethod(access, name, descriptor, signature, exceptions);
if (!name.equals("doAcquireNanos") && !name.equals("doAcquireSharedNanos"))
return mv;
return new MethodVisitor(api, mv)
{
@Override
public void visitLdcInsn(Object value)
{
if (Objects.equals(defaultSpinForTimeoutThreshold, value))
super.visitLdcInsn(0L);
else
super.visitLdcInsn(value);
}
};
}
}
return transform(className, bytes, AbstractQueuedSynchronizerVisitor::new, flag, flags);
}
private static byte[] transform(byte[] bytes, BiFunction<Integer, ClassVisitor, ClassVisitor> constructor)
{
ClassWriter out = new ClassWriter(0);
ClassReader in = new ClassReader(bytes);
@ -332,7 +417,7 @@ public class InterceptAgent
return out.toByteArray();
}
private static byte[] transformConcurrent(String className, byte[] bytes, Flag flag, Flag ... flags)
private static byte[] transform(String className, byte[] bytes, Flag flag, Flag ... flags)
{
ClassTransformer transformer = new ClassTransformer(BYTECODE_VERSION, className, EnumSet.of(flag, flags), null);
transformer.readAndTransform(bytes);
@ -340,4 +425,13 @@ public class InterceptAgent
return null;
return transformer.toBytes();
}
private static byte[] transform(String className, byte[] bytes, BiFunction<Integer, ClassVisitor, ClassVisitor> constructor, Flag flag, Flag ... flags)
{
ClassReader in = new ClassReader(bytes);
ClassTransformer transformer = new ClassTransformer(BYTECODE_VERSION, className, EnumSet.of(flag, flags), null);
ClassVisitor extraTransformer = constructor.apply(BYTECODE_VERSION, transformer);
in.accept(extraTransformer, 0);
return transformer.toBytes();
}
}

2
test/simulator/asm/org/apache/cassandra/simulator/asm/InterceptClasses.java
View File

@ -62,6 +62,8 @@ public class InterceptClasses implements BiFunction<String, byte[], byte[]>
"|org[/.]apache[/.]cassandra[/.]distributed[/.]impl[/.]DirectStreamingConnectionFactory.*" +
"|org[/.]apache[/.]cassandra[/.]db[/.]commitlog[/.].*" +
"|org[/.]apache[/.]cassandra[/.]service[/.]paxos[/.].*" +
"|org[/.]apache[/.]cassandra[/.]service[/.]accord[/.].*" +
"|org[/.]apache[/.]cassandra[/.]journal[/.].*" +
"|accord[/.].*"
);

3
test/simulator/asm/org/apache/cassandra/simulator/asm/MonitorMethodTransformer.java
View File

@ -122,8 +122,7 @@ class MonitorMethodTransformer extends MethodNode
}
int invokeCode;
if (isInstanceMethod && (access & Opcodes.ACC_PRIVATE) != 0) invokeCode = Opcodes.INVOKESPECIAL;
else if (isInstanceMethod) invokeCode = Opcodes.INVOKEVIRTUAL;
if (isInstanceMethod) invokeCode = Opcodes.INVOKESPECIAL;
else invokeCode = Opcodes.INVOKESTATIC;
return invokeCode;
}

43
test/simulator/asm/org/apache/cassandra/simulator/asm/StringHashcode.java Normal file
View File

@ -0,0 +1,43 @@
/*
* 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.simulator.asm;
import org.objectweb.asm.Opcodes;
import org.objectweb.asm.tree.InsnNode;
import org.objectweb.asm.tree.LabelNode;
import org.objectweb.asm.tree.MethodInsnNode;
import org.objectweb.asm.tree.MethodNode;
/**
* Generate a new hashCode method in the class that invokes a deterministic hashCode generator
*/
class StringHashcode extends MethodNode
{
StringHashcode(int api)
{
super(api, Opcodes.ACC_PUBLIC, "hashCode", "()I", null, null);
maxLocals = 1;
maxStack = 1;
instructions.add(new LabelNode());
instructions.add(new MethodInsnNode(Opcodes.INVOKEVIRTUAL, "java/lang/Object", "toString", "()Ljava/lang/String;", false));
instructions.add(new LabelNode());
instructions.add(new MethodInsnNode(Opcodes.INVOKEVIRTUAL, "java/lang/Object", "hashCode", "(Ljava/lang/Object;)I", false));
instructions.add(new InsnNode(Opcodes.IRETURN));
}
}

17
test/simulator/main/org/apache/cassandra/simulator/ActionSchedule.java
View File

@ -281,6 +281,12 @@ public class ActionSchedule implements CloseableIterator<Object>, LongConsumer
if (!runnable.isEmpty() || !scheduled.isEmpty())
return true;
while (moreWork())
{
if (!runnable.isEmpty() || !scheduled.isEmpty())
return true;
}
if (!sequences.isEmpty())
{
// TODO (feature): detection of which action is blocking progress, and logging of its stack trace only
@ -313,15 +319,12 @@ public class ActionSchedule implements CloseableIterator<Object>, LongConsumer
throw failWithOOM();
}
while (moreWork())
{
if (!runnable.isEmpty() || !scheduled.isEmpty())
return true;
}
return false;
}
// NOTE: this is only here for debugging, its a quick way to see if pre (0), interleave (1), or post (2) is active
private int step = -1;
private boolean moreWork()
{
if (!moreWork.hasNext())
@ -347,6 +350,8 @@ public class ActionSchedule implements CloseableIterator<Object>, LongConsumer
work.actors.forEach(runnableScheduler::attachTo);
work.actors.forEach(a -> a.forEach(Action::setConsequence));
work.actors.forEach(this::add);
step++;
return true;
}

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

@ -52,6 +52,7 @@ import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableBiConsumer;
import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableConsumer;
import org.apache.cassandra.distributed.api.IIsolatedExecutor.SerializableRunnable;
import org.apache.cassandra.distributed.impl.ClusterIDDefiner;
import org.apache.cassandra.distributed.impl.DirectStreamingConnectionFactory;
import org.apache.cassandra.distributed.impl.InstanceConfig;
import org.apache.cassandra.distributed.impl.InstanceIDDefiner;
@ -775,7 +776,8 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
.set("disk_access_mode", "standard")
.set("failure_detector", SimulatedFailureDetector.Instance.class.getName())
.set("commitlog_compression", new ParameterizedClass(LZ4Compressor.class.getName(), emptyMap()))
.set("commitlog_sync", "batch");
.set("commitlog_sync", "batch")
.set("lwt_strategy", builder.lwtStrategy);
// TODO: Add remove() to IInstanceConfig
if (config instanceof InstanceConfig)
{
@ -791,6 +793,11 @@ public class ClusterSimulation<S extends Simulation> implements AutoCloseable
@Override
public void initialise(ClassLoader classLoader, ThreadGroup threadGroup, int num, int generation)
{
IsolatedExecutor.transferAdhoc((IIsolatedExecutor.SerializableConsumer<String>) ClusterIDDefiner::setId, classLoader)
.accept(threadGroup.getParent().getName());
IsolatedExecutor.transferAdhoc((IIsolatedExecutor.SerializableConsumer<Integer>) InstanceIDDefiner::setInstanceId, classLoader)
.accept(num);
List<Closeable> onShutdown = new ArrayList<>();
IsolatedExecutor.transferAdhoc((SerializableConsumer<Integer>) InstanceIDDefiner::setInstanceId, classLoader)
.accept(num);

22
test/simulator/main/org/apache/cassandra/simulator/SimulationRunner.java
View File

@ -25,6 +25,7 @@ import java.util.List;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.function.ToDoubleFunction;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@ -51,6 +52,7 @@ import org.apache.cassandra.simulator.systems.InterceptibleThread;
import org.apache.cassandra.simulator.systems.InterceptorOfGlobalMethods;
import org.apache.cassandra.simulator.utils.ChanceRange;
import org.apache.cassandra.utils.Clock;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Hex;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
@ -75,7 +77,10 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.MEMTABLE_O
import static org.apache.cassandra.config.CassandraRelevantProperties.PAXOS_REPAIR_RETRY_TIMEOUT_IN_MS;
import static org.apache.cassandra.config.CassandraRelevantProperties.RING_DELAY;
import static org.apache.cassandra.config.CassandraRelevantProperties.SHUTDOWN_ANNOUNCE_DELAY_IN_MS;
import static org.apache.cassandra.config.CassandraRelevantProperties.SIMULATOR_STARTED;
import static org.apache.cassandra.config.CassandraRelevantProperties.SYSTEM_AUTH_DEFAULT_RF;
import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_CASSANDRA_SUITENAME;
import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_CASSANDRA_TESTTAG;
import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_JVM_DTEST_DISABLE_SSL;
import static org.apache.cassandra.simulator.debug.Reconcile.reconcileWith;
import static org.apache.cassandra.simulator.debug.Record.record;
@ -135,6 +140,7 @@ public class SimulationRunner
IGNORE_MISSING_NATIVE_FILE_HINTS.setBoolean(true);
ORG_APACHE_CASSANDRA_DISABLE_MBEAN_REGISTRATION.setBoolean(true);
TEST_JVM_DTEST_DISABLE_SSL.setBoolean(true); // to support easily running without netty from dtest-jar
SIMULATOR_STARTED.setString(Long.toString(TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())));
if (Thread.currentThread() instanceof InterceptibleThread); // load InterceptibleThread class to avoid infinite loop in InterceptorOfGlobalMethods
new InterceptedWait.CaptureSites(Thread.currentThread())
@ -344,8 +350,11 @@ public class SimulationRunner
{
long seed = parseHex(Optional.ofNullable(this.seed)).orElse(new Random(System.nanoTime()).nextLong());
SeedDefiner.setSeed(seed);
logger();
beforeAll();
// TODO (expected): this doesn't work properly for multiple seeds in a single JVM
TEST_CASSANDRA_TESTTAG.setString("simulator");
TEST_CASSANDRA_SUITENAME.setString(SIMULATOR_STARTED.getString() + '-' + CassandraRelevantProperties.SIMULATOR_SEED.getString());
logger();
Thread.setDefaultUncaughtExceptionHandler((th, e) -> {
boolean isInterrupt = false;
Throwable t = e;
@ -378,6 +387,7 @@ public class SimulationRunner
protected void run(long seed, B builder) throws IOException
{
logger().error("Seed 0x{}", Long.toHexString(seed));
logger().info("Cassandra {} / {}", FBUtilities.getReleaseVersionString(), FBUtilities.getGitSHA());
try (ClusterSimulation<?> cluster = builder.create(seed))
{
@ -456,6 +466,16 @@ public class SimulationRunner
}
}
@Command(name = "version", description = "Display version information")
protected static class VersionCommand<B extends ClusterSimulation.Builder<?>> implements ICommand<B>
{
@Override
public void run(B builder) throws IOException
{
System.out.println(FBUtilities.getReleaseVersionString());
System.out.println(FBUtilities.getGitSHA());
}
}
public static Optional<Long> parseHex(Optional<String> value)
{

4
test/simulator/main/org/apache/cassandra/simulator/debug/Record.java
View File

@ -46,6 +46,7 @@ import org.apache.cassandra.simulator.SimulationRunner.RecordOption;
import org.apache.cassandra.simulator.systems.SimulatedTime;
import org.apache.cassandra.utils.Closeable;
import org.apache.cassandra.utils.CloseableIterator;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.concurrent.Threads;
import static org.apache.cassandra.io.util.File.WriteMode.OVERWRITE;
@ -58,7 +59,7 @@ public class Record
{
private static final Logger logger = LoggerFactory.getLogger(Record.class);
private static final Pattern NORMALISE_THREAD_RECORDING_OUT = Pattern.compile("(Thread\\[[^]]+:[0-9]+),[0-9](,node[0-9]+)_[0-9]+]");
private static final Pattern NORMALISE_LAMBDA = Pattern.compile("((\\$\\$Lambda\\$[0-9]+/[0-9]+)?(@[0-9a-f]+)?)");
private static final Pattern NORMALISE_LAMBDA = Pattern.compile("((\\$\\$Lambda\\$[0-9]+/(0x)?[a-f0-9]+)?(@[0-9a-f]+)?)");
public static void record(String saveToDir, long seed, RecordOption withRng, RecordOption withTime, ClusterSimulation.Builder<?> builder)
{
@ -81,6 +82,7 @@ public class Record
if (builder.capture().wakeSites)
modifiers.add("WakeSites");
logger.error("Seed 0x{} ({}) (With: {})", Long.toHexString(seed), eventFile, modifiers);
logger.info("Cassandra {} / {}", FBUtilities.getReleaseVersionString(), FBUtilities.getGitSHA());
}
try (PrintWriter eventOut = new PrintWriter(new GZIPOutputStream(eventFile.newOutputStream(OVERWRITE), 1 << 16));

2
test/simulator/main/org/apache/cassandra/simulator/debug/SelfReconcile.java
View File

@ -43,6 +43,7 @@ import org.apache.cassandra.simulator.systems.InterceptibleThread;
import org.apache.cassandra.simulator.systems.InterceptorOfConsequences;
import org.apache.cassandra.simulator.systems.SimulatedTime;
import org.apache.cassandra.utils.CloseableIterator;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
import org.apache.cassandra.utils.memory.HeapPool;
@ -248,6 +249,7 @@ public class SelfReconcile
public static void reconcileWithSelf(long seed, RecordOption withRng, RecordOption withTime, boolean withAllocations, ClusterSimulation.Builder<?> builder)
{
logger.error("Seed 0x{}", Long.toHexString(seed));
logger.info("Cassandra {} / {}", FBUtilities.getReleaseVersionString(), FBUtilities.getGitSHA());
InterceptReconciler reconciler = new InterceptReconciler(withRng == WITH_CALLSITES);
if (withRng != NONE) builder.random(reconciler);

6
test/simulator/main/org/apache/cassandra/simulator/logging/RunStartDefiner.java
View File

@ -18,8 +18,7 @@
package org.apache.cassandra.simulator.logging;
import java.util.concurrent.TimeUnit;
import accord.utils.Invariants;
import ch.qos.logback.core.PropertyDefinerBase;
import org.apache.cassandra.config.CassandraRelevantProperties;
@ -27,8 +26,7 @@ public class RunStartDefiner extends PropertyDefinerBase
{
static
{
if (CassandraRelevantProperties.SIMULATOR_STARTED.getString() == null)
CassandraRelevantProperties.SIMULATOR_STARTED.setString(Long.toString(TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())));
Invariants.checkState(CassandraRelevantProperties.SIMULATOR_STARTED.getString() != null);
}
@Override

11
test/simulator/main/org/apache/cassandra/simulator/paxos/AbstractPairOfSequencesPaxosSimulation.java
View File

@ -39,11 +39,10 @@ import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.distributed.api.LogAction;
import org.apache.cassandra.distributed.api.LogResult;
import org.apache.cassandra.distributed.impl.FileLogAction;
import org.apache.cassandra.distributed.impl.Instance;
import org.apache.cassandra.distributed.shared.Metrics;
import org.apache.cassandra.io.util.File;
import org.apache.cassandra.simulator.Action;
import org.apache.cassandra.simulator.ActionList;
import org.apache.cassandra.simulator.ActionPlan;
@ -53,8 +52,6 @@ import org.apache.cassandra.simulator.RandomSource;
import org.apache.cassandra.simulator.RunnableActionScheduler;
import org.apache.cassandra.simulator.cluster.ClusterActions;
import org.apache.cassandra.simulator.cluster.KeyspaceActions;
import org.apache.cassandra.simulator.logging.RunStartDefiner;
import org.apache.cassandra.simulator.logging.SeedDefiner;
import org.apache.cassandra.simulator.systems.SimulatedActionTask;
import org.apache.cassandra.simulator.systems.SimulatedSystems;
import org.apache.cassandra.simulator.utils.IntRange;
@ -131,11 +128,7 @@ abstract class AbstractPairOfSequencesPaxosSimulation extends PaxosSimulation
@Override
protected ActionList performSimple()
{
// can't use inst.logs as that runs in the class loader, which uses in-memory file system
String suite = new RunStartDefiner().getPropertyValue() + "-" + new SeedDefiner().getPropertyValue();
String instanceId = "node" + inst.config().num();
File logFile = new File(String.format("build/test/logs/simulator/%s/%s/system.log", suite, instanceId));
FileLogAction logs = new FileLogAction(logFile);
LogAction logs = inst.logs();
LogResult<List<String>> errors = logs.grepForErrors();
if (!errors.getResult().isEmpty())

5
test/simulator/main/org/apache/cassandra/simulator/paxos/AccordClusterSimulation.java
View File

@ -43,7 +43,10 @@ class AccordClusterSimulation extends ClusterSimulation<PaxosSimulation> impleme
public void applyHandicaps()
{
/**
* TODO: remove after partial replication patch
* TODO (required): remove
* We currently require coordinators to have a CommandStore to coordinate a query, but not every node
* is a replica under standard simulation
*
* The current homekey implementation isn't compatible with the C* commands per key implementation when
* a non-replica coordinates a query.
*

27
test/simulator/main/org/apache/cassandra/simulator/paxos/AccordSimulationRunner.java
View File

@ -21,12 +21,22 @@ package org.apache.cassandra.simulator.paxos;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.BeforeClass;
import io.airlift.airline.Cli;
import io.airlift.airline.Command;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.simulator.SimulationRunner;
import org.apache.cassandra.utils.StorageCompatibilityMode;
public class AccordSimulationRunner extends SimulationRunner
{
@BeforeClass
public static void beforeAll()
{
CassandraRelevantProperties.JUNIT_STORAGE_COMPATIBILITY_MODE.setString(StorageCompatibilityMode.NONE.toString());
}
@Command(name = "run")
public static class Run extends SimulationRunner.Run<AccordClusterSimulation.Builder>
{
@ -35,6 +45,7 @@ public class AccordSimulationRunner extends SimulationRunner
@Override
protected void run(long seed, AccordClusterSimulation.Builder builder) throws IOException
{
beforeAll();
builder.applyHandicaps();
super.run(seed, builder);
}
@ -44,12 +55,28 @@ public class AccordSimulationRunner extends SimulationRunner
public static class Record extends SimulationRunner.Record<AccordClusterSimulation.Builder>
{
public Record() {}
@Override
protected void run(long seed, AccordClusterSimulation.Builder builder) throws IOException
{
beforeAll();
builder.applyHandicaps();
super.run(seed, builder);
}
}
@Command(name = "reconcile")
public static class Reconcile extends SimulationRunner.Reconcile<AccordClusterSimulation.Builder>
{
public Reconcile() {}
@Override
protected void run(long seed, AccordClusterSimulation.Builder builder) throws IOException
{
beforeAll();
builder.applyHandicaps();
super.run(seed, builder);
}
}
public static class Help extends HelpCommand<AccordClusterSimulation.Builder> {}

6
test/simulator/main/org/apache/cassandra/simulator/paxos/PairOfSequencesAccordSimulation.java
View File

@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory;
import com.carrotsearch.hppc.IntArrayList;
import com.carrotsearch.hppc.IntHashSet;
import com.carrotsearch.hppc.cursors.IntCursor;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.rows.Cell;
@ -139,7 +140,10 @@ public class PairOfSequencesAccordSimulation extends AbstractPairOfSequencesPaxo
seed, primaryKeys,
runForNanos, jitter);
this.writeRatio = 1F - readRatio;
validator = new LoggingHistoryValidator(new StrictSerializabilityValidator(primaryKeys));
HistoryValidator validator = new StrictSerializabilityValidator(primaryKeys);
if (CassandraRelevantProperties.TEST_HISTORY_VALIDATOR_LOGGING_ENABLED.getBoolean())
validator = new LoggingHistoryValidator(validator);
this.validator = validator;
}
@Override

2
test/simulator/main/org/apache/cassandra/simulator/paxos/PaxosSimulationRunner.java
View File

@ -137,6 +137,7 @@ public class PaxosSimulationRunner extends SimulationRunner
}
public static class Help extends HelpCommand<PaxosClusterSimulation.Builder> {}
public static class Version extends VersionCommand<PaxosClusterSimulation.Builder> {}
static void propagateTo(String consistency, boolean withStateCache, boolean withoutStateCache, String variant, String toVariant, PaxosClusterSimulation.Builder builder)
{
@ -163,6 +164,7 @@ public class PaxosSimulationRunner extends SimulationRunner
.withCommand(Run.class)
.withCommand(Reconcile.class)
.withCommand(Record.class)
.withCommand(Version.class)
.withCommand(Help.class)
.withDefaultCommand(Help.class)
.build()

3
test/simulator/main/org/apache/cassandra/simulator/systems/InterceptingMonitors.java
View File

@ -698,10 +698,7 @@ public abstract class InterceptingMonitors implements InterceptorOfGlobalMethods
{
if (!thread.isIntercepting() && disabled) return;
else if (!thread.isIntercepting())
{
throw new AssertionError("Thread " + thread + " is running but is not simulated");
}
checkForDeadlock(thread, state.heldBy);
InterceptedMonitorWait wait = new InterceptedMonitorWait(UNBOUNDED_WAIT, 0L, state, thread, captureWaitSite(thread));

2
test/unit/org/apache/cassandra/journal/TestParams.java
View File

@ -42,7 +42,7 @@ public class TestParams implements Params
}
@Override
public int flushPeriod()
public int flushPeriodMillis()
{
return 1000;
}