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

Deterministic Simulator thread ids 

Also: Don't ignore self reconcile test

patch by Ariel Weisberg; reviewed by David Capwell for CASSANDRA-20841
This commit is contained in:
Ariel Weisberg 2025-08-11 12:37:13 -04:00 committed by Benedict Elliott Smith
parent dddbe1ad87
commit d7a46b52ef
5 changed files with 27 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThread.java
View File

@ -189,6 +189,7 @@ public class InterceptibleThread extends FastThreadLocalThread implements Interc
final Runnable onTermination;
private final InterceptorOfGlobalMethods interceptorOfGlobalMethods;
private final LocalTime time;
private final long id;
// this is set before the thread's execution begins/continues; events and cessation are reported back to this
private InterceptorOfConsequences interceptor;
@ -205,7 +206,7 @@ public class InterceptibleThread extends FastThreadLocalThread implements Interc
// perform any non-deterministic actions
private int determinismDepth;
public InterceptibleThread(ThreadGroup group, Runnable target, String name, Object extraToStringInfo, Runnable onTermination, InterceptorOfGlobalMethods interceptorOfGlobalMethods, LocalTime time)
public InterceptibleThread(ThreadGroup group, Runnable target, String name, Object extraToStringInfo, Runnable onTermination, InterceptorOfGlobalMethods interceptorOfGlobalMethods, LocalTime time, long id)
{
super(group, target, name);
this.onTermination = onTermination;
@ -214,6 +215,13 @@ public class InterceptibleThread extends FastThreadLocalThread implements Interc
// group is nulled on termination, and we need it for reporting purposes, so save the toString
this.toString = "Thread[" + name + ',' + getPriority() + ',' + group.getName() + ']';
this.extraToStringInfo = extraToStringInfo;
this.id = id;
}
@Override
public long getId()
{
return id;
}
public boolean park(long waitTime, WaitTimeKind waitTimeKind)

11
test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThreadFactory.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.simulator.systems;
import java.io.Serializable;
import java.util.concurrent.ThreadFactory;
import java.util.function.Supplier;
import org.apache.cassandra.concurrent.NamedThreadFactory;
@ -28,7 +29,7 @@ public interface InterceptibleThreadFactory extends ThreadFactory
public interface MetaFactory<F extends ThreadFactory> extends Serializable
{
F create(String id, int priority, ClassLoader contextClassLoader, Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
ThreadGroup threadGroup, Runnable onTermination, SimulatedTime.LocalTime time, InterceptingExecutorFactory parent, Object extraToStringInfo);
ThreadGroup threadGroup, Runnable onTermination, SimulatedTime.LocalTime time, InterceptingExecutorFactory parent, Object extraToStringInfo, Supplier<Long> idSupplier);
}
public static class ConcreteInterceptibleThreadFactory extends NamedThreadFactory implements InterceptibleThreadFactory
@ -37,16 +38,18 @@ public interface InterceptibleThreadFactory extends ThreadFactory
final Runnable onTermination;
final SimulatedTime.LocalTime time;
final Object extraToStringInfo;
final Supplier<Long> idSupplier;
public ConcreteInterceptibleThreadFactory(String id, int priority, ClassLoader contextClassLoader, Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
ThreadGroup threadGroup, Runnable onTermination, SimulatedTime.LocalTime time,
InterceptingExecutorFactory parent, Object extraToStringInfo)
InterceptingExecutorFactory parent, Object extraToStringInfo, Supplier<Long> idSupplier)
{
super(id, priority, contextClassLoader, threadGroup, uncaughtExceptionHandler);
this.onTermination = onTermination;
this.time = time;
this.parent = parent;
this.extraToStringInfo = extraToStringInfo;
this.idSupplier = idSupplier;
}
@Override
@ -60,7 +63,7 @@ public interface InterceptibleThreadFactory extends ThreadFactory
{
// Can not use NamedThreadFactory.globalPrefix() as this method runs in the App class loader and not the Instance class loader; the ThreadGroup's name can act as a proxy for this.
String threadName = threadGroup.getName() + '_' + name;
InterceptibleThread thread = new InterceptibleThread(threadGroup, runnable, threadName, extraToStringInfo, onTermination, parent.interceptorOfGlobalMethods, time);
InterceptibleThread thread = new InterceptibleThread(threadGroup, runnable, threadName, extraToStringInfo, onTermination, parent.interceptorOfGlobalMethods, time, idSupplier.get());
if (parent.isClosed)
thread.trapInterrupts(false);
return setupThread(thread);
@ -72,7 +75,7 @@ public interface InterceptibleThreadFactory extends ThreadFactory
final Runnable onTermination;
public PlainThreadFactory(String id, int priority, ClassLoader contextClassLoader, Thread.UncaughtExceptionHandler uncaughtExceptionHandler,
ThreadGroup threadGroup, Runnable onTermination, SimulatedTime.LocalTime time, InterceptingExecutorFactory parent, Object extraToStringInfo)
ThreadGroup threadGroup, Runnable onTermination, SimulatedTime.LocalTime time, InterceptingExecutorFactory parent, Object extraToStringInfo, Supplier<Long> idSupplier)
{
super(id, priority, contextClassLoader, threadGroup, uncaughtExceptionHandler);
this.onTermination = onTermination;

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

@ -27,6 +27,7 @@ import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Supplier;
import com.google.common.annotations.VisibleForTesting;
@ -187,15 +188,17 @@ public class InterceptingExecutorFactory implements ExecutorFactory, Closeable
final ClassLoader classLoader;
final ThreadGroup threadGroup;
final IIsolatedExecutor.DynamicFunction<Serializable> transferToInstance;
final Supplier<Long> idSupplier;
volatile boolean isClosed;
InterceptingExecutorFactory(SimulatedExecution simulatedExecution, InterceptorOfGlobalMethods interceptorOfGlobalMethods, ClassLoader classLoader, ThreadGroup threadGroup)
InterceptingExecutorFactory(SimulatedExecution simulatedExecution, InterceptorOfGlobalMethods interceptorOfGlobalMethods, ClassLoader classLoader, ThreadGroup threadGroup, Supplier<Long> idSupplier)
{
this.simulatedExecution = simulatedExecution;
this.interceptorOfGlobalMethods = interceptorOfGlobalMethods;
this.classLoader = classLoader;
this.threadGroup = threadGroup;
this.transferToInstance = IsolatedExecutor.transferTo(classLoader);
this.idSupplier = idSupplier;
}
public InterceptibleThreadFactory factory(String name)
@ -232,7 +235,7 @@ public class InterceptingExecutorFactory implements ExecutorFactory, Closeable
else if (!this.threadGroup.parentOf(threadGroup)) throw new IllegalArgumentException();
Runnable onTermination = transferToInstance.apply((SerializableRunnable)FastThreadLocal::removeAll);
LocalTime time = transferToInstance.apply((SerializableCallable<LocalTime>) SimulatedTime.Global::current).call();
return factory.create(name, Thread.NORM_PRIORITY, classLoader, uncaughtExceptionHandler, threadGroup, onTermination, time, this, extraInfo);
return factory.create(name, Thread.NORM_PRIORITY, classLoader, uncaughtExceptionHandler, threadGroup, onTermination, time, this, extraInfo, idSupplier);
}
@Override

4
test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedExecution.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.simulator.systems;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import org.apache.cassandra.simulator.ActionList;
@ -41,6 +42,7 @@ import static org.apache.cassandra.simulator.systems.SimulatedAction.Kind.TASK;
public class SimulatedExecution implements InterceptorOfExecution
{
private final AtomicLong idSupplier = new AtomicLong(0);
static class NoExecutorMarker implements InterceptingExecutor
{
static final NoExecutorMarker INFINITE_LOOP = new NoExecutorMarker();
@ -160,7 +162,7 @@ public class SimulatedExecution implements InterceptorOfExecution
public InterceptingExecutorFactory factory(InterceptorOfGlobalMethods interceptorOfGlobalMethods, ClassLoader classLoader, ThreadGroup threadGroup)
{
return new InterceptingExecutorFactory(this, interceptorOfGlobalMethods, classLoader, threadGroup);
return new InterceptingExecutorFactory(this, interceptorOfGlobalMethods, classLoader, threadGroup, idSupplier::incrementAndGet);
}
public InterceptExecution intercept()

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

@ -18,7 +18,6 @@
package org.apache.cassandra.simulator.test;
import org.junit.Ignore;
import org.junit.Test;
import org.apache.cassandra.simulator.paxos.PaxosSimulationRunner;
@ -105,7 +104,6 @@ public class ShortPaxosSimulationTest
}
@Test
@Ignore("fails due to OOM DirectMemory - unclear why")
public void selfReconcileTest()
{
PaxosSimulationRunner.executeWithExceptionThrowing(new String[] { "reconcile",
@ -114,7 +112,9 @@ public class ShortPaxosSimulationTest
"-c", "2",
"--cluster-action-limit", "2",
"-s", "30",
"--with-self" });
"--with-self",
"--with-rng", "0",
"--with-time", "0",});
}
}