diff --git a/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThread.java b/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThread.java index 1300a0a47d..0d1e38b298 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThread.java +++ b/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThread.java @@ -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) diff --git a/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThreadFactory.java b/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThreadFactory.java index d9d443b946..aa3b8e6d2c 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThreadFactory.java +++ b/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThreadFactory.java @@ -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 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 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 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 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 idSupplier) { super(id, priority, contextClassLoader, threadGroup, uncaughtExceptionHandler); this.onTermination = onTermination; diff --git a/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptingExecutorFactory.java b/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptingExecutorFactory.java index aa22e3ef86..d4b58ef890 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptingExecutorFactory.java +++ b/test/simulator/main/org/apache/cassandra/simulator/systems/InterceptingExecutorFactory.java @@ -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 transferToInstance; + final Supplier idSupplier; volatile boolean isClosed; - InterceptingExecutorFactory(SimulatedExecution simulatedExecution, InterceptorOfGlobalMethods interceptorOfGlobalMethods, ClassLoader classLoader, ThreadGroup threadGroup) + InterceptingExecutorFactory(SimulatedExecution simulatedExecution, InterceptorOfGlobalMethods interceptorOfGlobalMethods, ClassLoader classLoader, ThreadGroup threadGroup, Supplier 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) 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 diff --git a/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedExecution.java b/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedExecution.java index a56aa84ea6..deff67312d 100644 --- a/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedExecution.java +++ b/test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedExecution.java @@ -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() diff --git a/test/simulator/test/org/apache/cassandra/simulator/test/ShortPaxosSimulationTest.java b/test/simulator/test/org/apache/cassandra/simulator/test/ShortPaxosSimulationTest.java index bd2e26e595..4f281dd7e8 100644 --- a/test/simulator/test/org/apache/cassandra/simulator/test/ShortPaxosSimulationTest.java +++ b/test/simulator/test/org/apache/cassandra/simulator/test/ShortPaxosSimulationTest.java @@ -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",}); } }