Deterministic Simulator thread ids

Also: Don't ignore self reconcile test

patch by Ariel Weisberg; reviewed by David Capwell for CASSANDRA-20841

Author: aweisberg

test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThread.java

@@ -189,6 +189,7 @@ public String toString()
     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 String toString()
     // 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 InterceptibleThread(ThreadGroup group, Runnable target, String name, Obje
         // 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)

test/simulator/main/org/apache/cassandra/simulator/systems/InterceptibleThreadFactory.java

@@ -20,6 +20,7 @@
 
 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 static class ConcreteInterceptibleThreadFactory extends NamedThreadFactor
         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 @@ protected synchronized InterceptibleThread newThread(ThreadGroup threadGroup, Ru
         {
             // 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 static class PlainThreadFactory extends NamedThreadFactory
         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;

test/simulator/main/org/apache/cassandra/simulator/systems/InterceptingExecutorFactory.java

@@ -27,6 +27,7 @@
 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 E build()
     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 @@ <F extends ThreadFactory> F factory(String name, Object extraInfo, ThreadGroup t
         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

test/simulator/main/org/apache/cassandra/simulator/systems/SimulatedExecution.java

@@ -21,6 +21,7 @@
 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 @@
 
 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 SimulatedExecution()
 
     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()

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 void simulationTest()
     }
 
     @Test
-    @Ignore("fails due to OOM DirectMemory - unclear why")
     public void selfReconcileTest()
     {
         PaxosSimulationRunner.executeWithExceptionThrowing(new String[] { "reconcile",
@@ -114,7 +112,9 @@ public void selfReconcileTest()
                                                                           "-c", "2",
                                                                           "--cluster-action-limit", "2",
                                                                           "-s", "30",
-                                                                          "--with-self" });
+                                                                          "--with-self",
+                                                                          "--with-rng", "0",
+                                                                          "--with-time", "0",});
     }
 }