content: /*
   * 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;
  
  import java.io.IOException;
  import java.lang.reflect.Field;
  import java.lang.reflect.Modifier;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.EnumMap;
  import java.util.List;
  import java.util.Map;
  import java.util.TreeMap;
  import java.util.concurrent.Callable;
  import java.util.concurrent.CopyOnWriteArrayList;
  import java.util.function.Consumer;
  import java.util.function.IntSupplier;
  import java.util.function.LongConsumer;
  import java.util.function.Predicate;
  import java.util.function.Supplier;
  
  import com.google.common.util.concurrent.AsyncFunction;
  import com.google.common.util.concurrent.FutureCallback;
  
  import org.apache.cassandra.auth.PasswordSaltSupplier;
  import org.apache.cassandra.concurrent.ExecutorFactory;
  import org.apache.cassandra.config.CassandraRelevantProperties;
  import org.apache.cassandra.config.ParameterizedClass;
  import org.apache.cassandra.distributed.Cluster;
  import org.apache.cassandra.distributed.Constants;
  import org.apache.cassandra.distributed.api.Feature;
  import org.apache.cassandra.distributed.api.IInstance;
  import org.apache.cassandra.distributed.api.IInstanceConfig;
  import org.apache.cassandra.distributed.api.IInstanceInitializer;
  import org.apache.cassandra.distributed.api.IInvokableInstance;
  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;
  import org.apache.cassandra.distributed.impl.IsolatedExecutor;
  import org.apache.cassandra.io.compress.LZ4Compressor;
  import org.apache.cassandra.io.filesystem.ListenableFileSystem;
  import org.apache.cassandra.io.util.FileSystems;
  import org.apache.cassandra.net.Verb;
  import org.apache.cassandra.service.consensus.TransactionalMode;
  import org.apache.cassandra.service.paxos.BallotGenerator;
  import org.apache.cassandra.service.paxos.PaxosPrepare;
  import org.apache.cassandra.simulator.RandomSource.Choices;
  import org.apache.cassandra.simulator.asm.DeterministicChanceSupplier;
  import org.apache.cassandra.simulator.asm.InterceptAsClassTransformer;
  import org.apache.cassandra.simulator.asm.NemesisFieldSelectors;
  import org.apache.cassandra.simulator.cluster.ClusterActions;
  import org.apache.cassandra.simulator.cluster.ClusterActions.ConsensusChange;
  import org.apache.cassandra.simulator.cluster.ClusterActions.TopologyChange;
  import org.apache.cassandra.simulator.systems.Failures;
  import org.apache.cassandra.simulator.systems.InterceptedWait.CaptureSites.Capture;
  import org.apache.cassandra.simulator.systems.InterceptibleThread;
  import org.apache.cassandra.simulator.systems.InterceptingExecutorFactory;
  import org.apache.cassandra.simulator.systems.InterceptingGlobalMethods;
  import org.apache.cassandra.simulator.systems.InterceptingGlobalMethods.ThreadLocalRandomCheck;
  import org.apache.cassandra.simulator.systems.InterceptorOfGlobalMethods;
  import org.apache.cassandra.simulator.systems.InterceptorOfGlobalMethods.IfInterceptibleThread;
  import org.apache.cassandra.simulator.systems.NetworkConfig;
  import org.apache.cassandra.simulator.systems.NetworkConfig.PhaseConfig;
  import org.apache.cassandra.simulator.systems.SchedulerConfig;
  import org.apache.cassandra.simulator.systems.SimulatedBallots;
  import org.apache.cassandra.simulator.systems.SimulatedExecution;
  import org.apache.cassandra.simulator.systems.SimulatedFailureDetector;
  import org.apache.cassandra.simulator.systems.SimulatedFutureActionScheduler;
  import org.apache.cassandra.simulator.systems.SimulatedMessageDelivery;
  import org.apache.cassandra.simulator.systems.SimulatedSnitch;
  import org.apache.cassandra.simulator.systems.SimulatedSystems;
  import org.apache.cassandra.simulator.systems.SimulatedTime;
  import org.apache.cassandra.simulator.utils.ChanceRange;
  import org.apache.cassandra.simulator.utils.IntRange;
  import org.apache.cassandra.simulator.utils.KindOfSequence;
  import org.apache.cassandra.simulator.utils.LongRange;
  import org.apache.cassandra.utils.Clock;
  import org.apache.cassandra.utils.Closeable;
  import org.apache.cassandra.utils.StorageCompatibilityMode;
  import org.apache.cassandra.utils.Throwables;
  import org.apache.cassandra.utils.concurrent.Ref;
  import org.apache.cassandra.utils.memory.BufferPool;
  import org.apache.cassandra.utils.memory.BufferPools;
  import org.apache.cassandra.utils.memory.HeapPool;
  
  import static java.lang.Integer.min;
  import static java.util.Collections.emptyMap;
  import static java.util.concurrent.TimeUnit.MICROSECONDS;
  import static java.util.concurrent.TimeUnit.MILLISECONDS;
  import static java.util.concurrent.TimeUnit.NANOSECONDS;
  import static java.util.concurrent.TimeUnit.SECONDS;
  import static org.apache.cassandra.distributed.impl.AbstractCluster.getSharedClassPredicate;
  import static org.apache.cassandra.simulator.SimulatorUtils.failWithOOM;
  import static org.apache.cassandra.utils.Shared.Scope.ANY;
  import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
  
  /**
   * Wraps a Cluster and a Simulation to run upon it
   */
  @SuppressWarnings("RedundantCast")
  public class ClusterSimulation<S extends Simulation> implements AutoCloseable
  {
      static
      {
          CassandraRelevantProperties.TEST_STORAGE_COMPATIBILITY_MODE.setEnum(StorageCompatibilityMode.NONE);
      }
  
      public static final Class<?>[] SHARE = new Class[]
                                          {
                                              AsyncFunction.class,
                                              FutureCallback.class,
                                              io.netty.util.concurrent.GenericFutureListener.class,
                                              io.netty.channel.FileRegion.class,
                                              io.netty.util.ReferenceCounted.class,
                                              io.netty.util.concurrent.FastThreadLocal.class
                                          };
  
      public static final Class<?>[] ISOLATE = new Class<?>[0];
  
      public interface SimulationFactory<S extends Simulation>
      {
          S create(SimulatedSystems simulated, RunnableActionScheduler scheduler, Cluster cluster, ClusterActions.Options options);
      }
  
      public interface SchedulerFactory
      {
          RunnableActionScheduler create(RandomSource random);
      }
  
      public interface FutureActionSchedulerFactory
      {
          FutureActionScheduler create(int nodeCount, SimulatedTime time, RandomSource random);
      }
  
      public interface PerVerbFutureActionSchedulersFactory
      {
          Map<Verb, FutureActionScheduler> create(int nodeCount, SimulatedTime time, RandomSource random);
      }
  
      @SuppressWarnings("UnusedReturnValue")
      public static abstract class Builder<S extends Simulation>
      {
          protected Supplier<RandomSource> randomSupplier = RandomSource.Default::new;
          protected int uniqueNum = 0;
          protected int threadCount;
  
          protected int concurrency = 10;
          protected IntRange nodeCount = new IntRange(4, 16), dcCount = new IntRange(1, 2),
                          primaryKeySeconds = new IntRange(5, 30), withinKeyConcurrency = new IntRange(2, 5);
          protected TopologyChange[] topologyChanges = TopologyChange.values();
          protected int topologyChangeLimit = -1;
  
          protected ConsensusChange[] consensusChanges = ConsensusChange.values();
          protected int consensusChangeLimit = -1;
  
          protected int primaryKeyCount;
          protected int secondsToSimulate;
  
          protected ChanceRange normalNetworkDropChance  = new ChanceRange(randomSource -> randomSource.qlog2uniformFloat(4), 0f, 0.001f),
                                normalNetworkDelayChance = new ChanceRange(randomSource -> randomSource.qlog2uniformFloat(4), 0.01f, 0.1f),
                                  flakyNetworkDropChance = new ChanceRange(randomSource -> randomSource.qlog2uniformFloat(4), 0.01f, 0.1f),
                                 flakyNetworkDelayChance = new ChanceRange(randomSource -> randomSource.qlog2uniformFloat(4), 0.01f, 0.1f),
                                  networkPartitionChance = new ChanceRange(randomSource -> randomSource.qlog2uniformFloat(4), 0.0f, 0.1f),
                                      networkFlakyChance = new ChanceRange(randomSource -> randomSource.qlog2uniformFloat(4), 0.0f, 0.1f),
                                      monitorDelayChance = new ChanceRange(randomSource -> randomSource.qlog2uniformFloat(4), 0.01f, 0.1f),
                                    schedulerDelayChance = new ChanceRange(randomSource -> randomSource.qlog2uniformFloat(4), 0.01f, 0.1f),
                                           timeoutChance = new ChanceRange(randomSource -> randomSource.qlog2uniformFloat(4), 0.01f, 0.1f),
                                              readChance = new ChanceRange(RandomSource::uniformFloat,                        0.05f, 0.95f),
                                           nemesisChance = new ChanceRange(randomSource -> randomSource.qlog2uniformFloat(4), 0.001f, 0.01f);
  
          protected LongRange normalNetworkLatencyNanos = new LongRange(1, 2, MILLISECONDS, NANOSECONDS),
                                normalNetworkDelayNanos = new LongRange(2, 100, MILLISECONDS, NANOSECONDS),
                               flakyNetworkLatencyNanos = new LongRange(2, 100, MILLISECONDS, NANOSECONDS),
                                 flakyNetworkDelayNanos = new LongRange(2, 100, MILLISECONDS, NANOSECONDS),
                             networkReconfigureInterval = new LongRange(50, 5000, MICROSECONDS, NANOSECONDS),
                                   schedulerJitterNanos = new LongRange(100, 2000, MICROSECONDS, NANOSECONDS),
                                    schedulerDelayNanos = new LongRange(0, 50, MICROSECONDS, NANOSECONDS),
                                schedulerLongDelayNanos = new LongRange(50, 5000, MICROSECONDS, NANOSECONDS),
                                        clockDriftNanos = new LongRange(1, 5000, MILLISECONDS, NANOSECONDS),
                         clockDiscontinuitIntervalNanos = new LongRange(10, 60, SECONDS, NANOSECONDS),
                            topologyChangeIntervalNanos = new LongRange(5, 15, SECONDS, NANOSECONDS),
                           consensusChangeIntervalNanos = new LongRange(1, 5, SECONDS, NANOSECONDS);
  
  
  
          protected long contentionTimeoutNanos = MILLISECONDS.toNanos(500L),
                              writeTimeoutNanos = SECONDS.toNanos(1L),
                               readTimeoutNanos = SECONDS.toNanos(2L),
                            requestTimeoutNanos = SECONDS.toNanos(2L);
  
          protected SchedulerFactory schedulerFactory = schedulerFactory(RunnableActionScheduler.Kind.values());
  
          protected Debug debug = new Debug();
          protected Failures failures = new Failures();
          protected Capture capture = new Capture(false, false, false);
          protected HeapPool.Logged.Listener memoryListener;
          protected SimulatedTime.Listener timeListener = (i1, i2) -> {};
          protected LongConsumer onThreadLocalRandomCheck;
          protected String transactionalMode = "off";
          protected FutureActionSchedulerFactory futureActionSchedulerFactory;
          protected PerVerbFutureActionSchedulersFactory perVerbFutureActionSchedulersFactory;
  
          public Builder<S> failures(Failures failures)
          {
              this.failures = failures;
              return this;
          }
  
          public Builder<S> writeTimeoutNanos(long nanos)
          {
              this.writeTimeoutNanos = nanos;
              return this;
          }
  
          public Builder<S> readTimeoutNanos(long nanos)
          {
              this.readTimeoutNanos = nanos;
              return this;
          }
  
          public Builder<S> requestTimeoutNanos(long nanos)
          {
              this.requestTimeoutNanos = nanos;
              return this;
          }
  
          public Debug debug()
          {
              return debug;
          }
  
          public Builder<S> debug(EnumMap<Debug.Info, Debug.Levels> debug, int[] primaryKeys)
          {
              this.debug = new Debug(debug, primaryKeys);
              return this;
          }
  
          public Builder<S> unique(int num)
          {
              this.uniqueNum = num;
              return this;
          }
  
          public Builder<S> threadCount(int count)
          {
              this.threadCount = count;
              return this;
          }
  
          public Builder<S> nodes(IntRange range)
          {
              this.nodeCount = range;
              return this;
          }
  
          public Builder<S> nodes(int min, int max)
          {
              this.nodeCount = new IntRange(min, max);
              return this;
          }
  
          public Builder<S> dcs(IntRange range)
          {
              this.dcCount = range;
              return this;
          }
  
          public Builder<S> dcs(int min, int max)
          {
              this.dcCount = new IntRange(min, max);
              return this;
          }
  
          public Builder<S> concurrency(int concurrency)
          {
              this.concurrency = concurrency;
              return this;
          }
  
          public IntRange primaryKeySeconds()
          {
              return primaryKeySeconds;
          }
  
          public Builder<S> primaryKeySeconds(IntRange range)
          {
              this.primaryKeySeconds = range;
              return this;
          }
  
          public Builder<S> withinKeyConcurrency(IntRange range)
          {
              this.withinKeyConcurrency = range;
              return this;
          }
  
          public Builder<S> withinKeyConcurrency(int min, int max)
          {
              this.withinKeyConcurrency = new IntRange(min, max);
              return this;
          }
  
          public Builder<S> topologyChanges(TopologyChange[] topologyChanges)
          {
              this.topologyChanges = topologyChanges;
              return this;
          }
  
          public Builder<S> topologyChangeIntervalNanos(LongRange topologyChangeIntervalNanos)
          {
              this.topologyChangeIntervalNanos = topologyChangeIntervalNanos;
              return this;
          }
  
          public Builder<S> topologyChangeLimit(int topologyChangeLimit)
          {
              this.topologyChangeLimit = topologyChangeLimit;
              return this;
          }
  
          public Builder<S> consensusChanges(ConsensusChange[] consensusChanges)
          {
              this.consensusChanges = consensusChanges;
              return this;
          }
  
          public Builder<S> consensusChangeIntervalNanos(LongRange consensusChangeIntervalNanos)
          {
              this.consensusChangeIntervalNanos = consensusChangeIntervalNanos;
              return this;
          }
  
          public Builder<S> consensusChangeLimit(int consensusChangeLimit)
          {
              this.consensusChangeLimit = consensusChangeLimit;
              return this;
          }
  
          public int primaryKeyCount()
          {
              return primaryKeyCount;
          }
  
          public Builder<S> primaryKeyCount(int count)
          {
              this.primaryKeyCount = count;
              return this;
          }
  
          public int secondsToSimulate()
          {
              return secondsToSimulate;
          }
  
          public Builder<S> secondsToSimulate(int seconds)
          {
              this.secondsToSimulate = seconds;
              return this;
          }
  
          public Builder<S> networkPartitionChance(ChanceRange partitionChance)
          {
              this.networkPartitionChance = partitionChance;
              return this;
          }
  
          public Builder<S> networkFlakyChance(ChanceRange flakyChance)
          {
              this.networkFlakyChance = flakyChance;
              return this;
          }
  
          public Builder<S> networkReconfigureInterval(LongRange reconfigureIntervalNanos)
          {
              this.networkReconfigureInterval = reconfigureIntervalNanos;
              return this;
          }
  
          public Builder<S> networkDropChance(ChanceRange dropChance)
          {
              this.normalNetworkDropChance = dropChance;
              return this;
          }
  
          public Builder<S> networkDelayChance(ChanceRange delayChance)
          {
              this.normalNetworkDelayChance = delayChance;
              return this;
          }
  
          public Builder<S> networkLatencyNanos(LongRange networkLatencyNanos)
          {
              this.normalNetworkLatencyNanos = networkLatencyNanos;
              return this;
          }
  
          public Builder<S> networkDelayNanos(LongRange networkDelayNanos)
          {
              this.normalNetworkDelayNanos = networkDelayNanos;
              return this;
          }
  
          public Builder<S> flakyNetworkDropChance(ChanceRange dropChance)
          {
              this.flakyNetworkDropChance = dropChance;
              return this;
          }
  
          public Builder<S> flakyNetworkDelayChance(ChanceRange delayChance)
          {
              this.flakyNetworkDelayChance = delayChance;
              return this;
          }
  
          public Builder<S> flakyNetworkLatencyNanos(LongRange networkLatencyNanos)
          {
              this.flakyNetworkLatencyNanos = networkLatencyNanos;
              return this;
          }
  
          public Builder<S> flakyNetworkDelayNanos(LongRange networkDelayNanos)
          {
              this.flakyNetworkDelayNanos = networkDelayNanos;
              return this;
          }
  
          public Builder<S> clockDriftNanos(LongRange clockDriftNanos)
          {
              this.clockDriftNanos = clockDriftNanos;
              return this;
          }
  
          public Builder<S> clockDiscontinuityIntervalNanos(LongRange clockDiscontinuityIntervalNanos)
          {
              this.clockDiscontinuitIntervalNanos = clockDiscontinuityIntervalNanos;
              return this;
          }
  
          public Builder<S> schedulerDelayChance(ChanceRange delayChance)
          {
              this.schedulerDelayChance = delayChance;
              return this;
          }
  
          public Builder<S> schedulerJitterNanos(LongRange schedulerJitterNanos)
          {
              this.schedulerJitterNanos = schedulerJitterNanos;
              return this;
          }
  
          public LongRange schedulerJitterNanos()
          {
              return schedulerJitterNanos;
          }
  
          public Builder<S> schedulerDelayNanos(LongRange schedulerDelayNanos)
          {
              this.schedulerDelayNanos = schedulerDelayNanos;
              return this;
          }
  
          public Builder<S> schedulerLongDelayNanos(LongRange schedulerLongDelayNanos)
          {
              this.schedulerLongDelayNanos = schedulerLongDelayNanos;
              return this;
          }
  
          public Builder<S> timeoutChance(ChanceRange timeoutChance)
          {
              this.timeoutChance = timeoutChance;
              return this;
          }
  
          public ChanceRange readChance()
          {
              return readChance;
          }
  
          public IntRange withinKeyConcurrency()
          {
              return withinKeyConcurrency;
          }
  
          public int concurrency()
          {
              return concurrency;
          }
  
          public Builder<S> readChance(ChanceRange readChance)
          {
              this.readChance = readChance;
              return this;
          }
  
          public Builder<S> nemesisChance(ChanceRange nemesisChance)
          {
              this.nemesisChance = nemesisChance;
              return this;
          }
  
          public Builder<S> scheduler(RunnableActionScheduler.Kind... kinds)
          {
              this.schedulerFactory = schedulerFactory(kinds);
              return this;
          }
  
          public Builder<S> futureActionScheduler(FutureActionSchedulerFactory factory)
          {
              futureActionSchedulerFactory = factory;
              return this;
          }
  
          public FutureActionScheduler futureActionScheduler(int nodeCount, SimulatedTime time, RandomSource random)
          {
              if (futureActionSchedulerFactory != null)
                  return futureActionSchedulerFactory.create(nodeCount, time, random);
              KindOfSequence kind = Choices.random(random, KindOfSequence.values())
                                           .choose(random);
              return new SimulatedFutureActionScheduler(kind, nodeCount, random, time,
                                                        new NetworkConfig(new PhaseConfig(normalNetworkDropChance, normalNetworkDelayChance, normalNetworkLatencyNanos, normalNetworkDelayNanos),
                                                                          new PhaseConfig(flakyNetworkDropChance, flakyNetworkDelayChance, flakyNetworkLatencyNanos, flakyNetworkDelayNanos),
                                                                          networkPartitionChance, networkFlakyChance, networkReconfigureInterval),
                                                        new SchedulerConfig(schedulerDelayChance, schedulerDelayNanos, schedulerLongDelayNanos));
          }
  
          public Builder<S> perVerbFutureActionSchedulers(PerVerbFutureActionSchedulersFactory factory)
          {
              perVerbFutureActionSchedulersFactory = factory;
              return this;
          }
  
          public Map<Verb, FutureActionScheduler> perVerbFutureActionSchedulers(int nodeCount, SimulatedTime time, RandomSource random, FutureActionScheduler defaultScheduler)
          {
              if (perVerbFutureActionSchedulersFactory != null)
                  return perVerbFutureActionSchedulersFactory.create(nodeCount, time, random);
              return Collections.emptyMap();
          }
  
          static SchedulerFactory schedulerFactory(RunnableActionScheduler.Kind... kinds)
          {
              return (random) -> {
                  switch (Choices.random(random, kinds).choose(random))
                  {
                      default: throw new AssertionError();
                      case IMMEDIATE: return new RunnableActionScheduler.Immediate();
                      case SEQUENTIAL: return new RunnableActionScheduler.Sequential();
                      case UNIFORM: return new RunnableActionScheduler.RandomUniform(random);
                      case RANDOM_WALK: return new RunnableActionScheduler.RandomWalk(random);
                  }
              };
          }
  
          public Builder<S> scheduler(SchedulerFactory schedulerFactory)
          {
              this.schedulerFactory = schedulerFactory;
              return this;
          }
  
          public Builder<S> random(Supplier<RandomSource> randomSupplier)
          {
              this.randomSupplier = randomSupplier;
              return this;
          }
  
          public Builder<S> memoryListener(HeapPool.Logged.Listener memoryListener)
          {
              this.memoryListener = memoryListener;
              return this;
          }
  
          public Builder<S> timeListener(SimulatedTime.Listener timeListener)
          {
              this.timeListener = timeListener;
              return this;
          }
  
          public Builder<S> capture(Capture capture)
          {
              this.capture = capture;
              return this;
          }
  
          public Capture capture()
          {
              return capture;
          }
  
          public Builder<S> onThreadLocalRandomCheck(LongConsumer runnable)
          {
              this.onThreadLocalRandomCheck = runnable;
              return this;
          }
  
          public Builder<S> transactionalMode(String mode)
          {
              this.transactionalMode = mode;
              return this;
          }
  
          public TransactionalMode transactionalMode()
          {
              return TransactionalMode.fromString(transactionalMode);
          }
  
          public abstract ClusterSimulation<S> create(long seed) throws IOException;
      }
  
      static class ThreadAllocator
      {
          final RandomSource random;
          int clusterPool; // number of threads we have left for the whole cluster
          int remainingNodes; // number of nodes we still need to allocate them to
          int allocationPool; //  threads to allocate for the node we're currently processing
          int remainingAllocations; // number of _remaining_ allocations take() assumes we want to evenly allocate threads over
  
          public ThreadAllocator(RandomSource random, int threadsToAllocate, int betweenNodes)
          {
              this.random = random;
              this.clusterPool = threadsToAllocate;
              this.remainingNodes = betweenNodes;
          }
  
          // randomly set the number of threads in various thread pools
          IInstanceConfig update(IInstanceConfig config)
          {
              cycle();
              // allocate in ascending order of max, for take() correctness
              return config
                     .set("memtable_flush_writers", take(1, 1, 2))
                     .set("concurrent_compactors", take(1, 1, 4))
                     .set("concurrent_writes", take(1, 4))
                     .set("concurrent_counter_writes", take(1, 4))
                     .set("concurrent_materialized_view_writes", take(1, 4))
                     .set("concurrent_reads", take(1, 4))
                     .set("available_processors", take(3, 4));
          }
  
          // begin allocating for a new node
          void cycle()
          {
              assert remainingNodes > 0;
              // return unallocated items to the outerPool
              clusterPool += allocationPool;
              // set the curPool to allocate the next allocationPool size
              allocationPool = clusterPool;
              remainingAllocations = remainingNodes;
              // randomly select the next pool size, subtracting it from the outer pool
              allocationPool = take(1, 1);
              clusterPool -= allocationPool;
              // this is hard-coded to match the sum of the first arguments above
              remainingAllocations = 9;
              --remainingNodes;
          }
  
          /**
           * See {@link #take(int, int, int)}
           */
          int take(int times, int min)
          {
              return take(times, min, allocationPool);
          }
  
          /**
           * Allocate a random number of threads between [min..max)
           * The allocation is suitable for multiple users of the value, i.e.
           * {@code times} multiple of the result are deducted from the pool.
           *
           * If there are adequate supplies we aim to allocate threads "equally" between pools,
           * selecting a uniform value between 0.5x and 2x the fair split of the remaining pool
           * on each allocation. If the min/max bounds override that, they are preferred.
           *
           * The minimum is always honoured, regardless of available pool size.
           */
          int take(int times, int min, int max)
          {
              int remaining = remainingAllocations;
              assert remaining >= times;
              remainingAllocations -= times;
              if (remaining * min <= allocationPool)
                  return min;
              if (times == remaining)
                  return Math.max(allocationPool / remaining, min);
              if (times + 1 == remaining)
                  return random.uniform(Math.max(min, (allocationPool - max) / times), Math.min(max, (allocationPool - min) / times));
  
              int median = allocationPool / remaining;
              min = Math.max(min, Math.min(max, median) / 2);
              max = Math.min(max, median * 2);
              return min >= max ? min : random.uniform(min, max);
          }
      }
  
      public final RandomSource random;
      public final SimulatedSystems simulated;
      public final Cluster cluster;
      private final ListenableFileSystem fs;
      protected final Map<Integer, List<Closeable>> onUnexpectedShutdown = new TreeMap<>();
      protected final List<Callable<Void>> onPreShutdown = new CopyOnWriteArrayList<>();
      protected final List<Callable<Void>> onPostShutdown = new CopyOnWriteArrayList<>();
      protected final ThreadLocalRandomCheck threadLocalRandomCheck;
  
      private final RunnableActionScheduler scheduler;
      private final ClusterActions.Options options;
      private final SimulationFactory<S> factory;
  
      public ClusterSimulation(RandomSource random, long seed, int uniqueNum,
                               Builder<?> builder,
                               Consumer<IInstanceConfig> configUpdater,
                               SimulationFactory<S> factory) throws IOException
      {
          this.random = random;
          this.fs = new ListenableFileSystem(FileSystems.jimfs(Long.toHexString(seed) + 'x' + uniqueNum));
  
          final SimulatedMessageDelivery delivery;
          final SimulatedExecution execution;
          final SimulatedBallots ballots;
          final SimulatedSnitch snitch;
          final SimulatedTime time;
          final SimulatedFailureDetector failureDetector;
  
          int numOfNodes = builder.nodeCount.select(random);
          int numOfDcs = builder.dcCount.select(random, 0, numOfNodes / 4);
          int[] numInDcs = new int[numOfDcs];
          int[] nodeToDc = new int[numOfNodes];
  
          int[] minRf = new int[numOfDcs], initialRf = new int[numOfDcs], maxRf = new int[numOfDcs];
          {
              // TODO (feature): split unevenly
              int n = 0, nc = 0;
              for (int i = 0; i < numOfDcs; ++i)
              {
                  int numInDc = (numOfNodes / numOfDcs) + (numOfNodes % numOfDcs > i ? 1 : 0);
                  numInDcs[i] = numInDc;
                  minRf[i] = min(numInDc, 3);
                  maxRf[i] = min(numInDc, 9);
                  initialRf[i] = random.uniform(minRf[i], 1 + maxRf[i]);
                  nc += numInDc;
                  while (n < nc)
                      nodeToDc[n++] = i;
              }
          }
          if (builder.topologyChangeLimit < 0)
              initialRf = maxRf;
  
          snitch = new SimulatedSnitch(nodeToDc, numInDcs);
  
          execution = new SimulatedExecution();
  
          KindOfSequence kindOfDriftSequence = Choices.uniform(KindOfSequence.values()).choose(random);
          KindOfSequence kindOfDiscontinuitySequence = Choices.uniform(KindOfSequence.values()).choose(random);
          time = new SimulatedTime(numOfNodes, random, 1577836800000L /*Jan 1st UTC*/, builder.clockDriftNanos, kindOfDriftSequence,
                                   kindOfDiscontinuitySequence.period(builder.clockDiscontinuitIntervalNanos, random),
                                   builder.timeListener);
          ballots = new SimulatedBallots(random, () -> {
              long max = random.uniform(2, 16);
              return () -> random.uniform(1, max);
          });
  
          Predicate<String> sharedClassPredicate = getSharedClassPredicate(ISOLATE, SHARE, ANY, SIMULATION);
          DeterministicChanceSupplier monitorDelayChance; {
              long monitorDelayChanceSeed = random.uniform(0, Long.MAX_VALUE);
              monitorDelayChance = hash -> {
                  RandomSource subRandom = new RandomSource.Default();
                  subRandom.reset(monitorDelayChanceSeed * 31 + hash);
                  return builder.monitorDelayChance.asSupplier(subRandom);
              };
          }
          DeterministicChanceSupplier nemesisChance; {
              long nemesisChanceSeed = random.uniform(0, Long.MAX_VALUE);
              nemesisChance = hash -> {
                  RandomSource subRandom = new RandomSource.Default();
                  subRandom.reset(nemesisChanceSeed * 31 + hash);
                  return builder.nemesisChance.asSupplier(subRandom);
              };
          }
  
          InterceptAsClassTransformer interceptClasses = new InterceptAsClassTransformer(monitorDelayChance, nemesisChance, NemesisFieldSelectors.get(), ClassLoader.getSystemClassLoader(), sharedClassPredicate.negate());
          threadLocalRandomCheck = new ThreadLocalRandomCheck(builder.onThreadLocalRandomCheck);
  
          Failures failures = builder.failures;
          ThreadAllocator threadAllocator = new ThreadAllocator(random, builder.threadCount, numOfNodes);
  
          cluster = snitch.setup(Cluster.build(numOfNodes)
                           .withRoot(fs.getPath("/cassandra"))
                           .withSharedClasses(sharedClassPredicate)
                           .withConfig(config -> {
                               config.with(Feature.BLANK_GOSSIP)
                                     .set(Constants.KEY_DTEST_JOIN_RING, false)
                                     .set("read_request_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.readTimeoutNanos)))
                                     .set("write_request_timeout", String.format("%dms", NANOSECONDS.toMillis(builder.writeTimeoutNanos)))
                                     .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("file_cache_size", "16MiB")
                                     .set("use_deterministic_table_id", true)
                                     .set("accord.queue_submission_model", "ASYNC")
                                     .set("accord.range_migration", "explicit")
                                     .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("accord.journal.flush_mode", "BATCH")
                                     .set("accord.command_store_shard_count", "4");
                               // TODO: Add remove() to IInstanceConfig
                               if (config instanceof InstanceConfig)
                               {
                                   InstanceConfig instanceConfig = (InstanceConfig) config;
                                   instanceConfig.remove("commitlog_sync_period_in_ms");
                                   instanceConfig.remove("commitlog_sync_period");
                               }
  
                               configUpdater.accept(threadAllocator.update(config));
                           })
                           .withInstanceInitializer(new IInstanceInitializer()
                           {
                               @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);
                                   InterceptorOfGlobalMethods interceptorOfGlobalMethods = IsolatedExecutor.transferAdhoc((IIsolatedExecutor.SerializableQuadFunction<Capture, LongConsumer, Consumer<Throwable>, RandomSource, InterceptorOfGlobalMethods>) InterceptingGlobalMethods::new, classLoader)
                                                                                                           .apply(builder.capture, builder.onThreadLocalRandomCheck, failures, random);
                                   onShutdown.add(interceptorOfGlobalMethods);
  
                                   InterceptingExecutorFactory factory = execution.factory(interceptorOfGlobalMethods, classLoader, threadGroup);
                                   IsolatedExecutor.transferAdhoc((SerializableConsumer<ExecutorFactory>) ExecutorFactory.Global::unsafeSet, classLoader)
                                                   .accept(factory);
                                   onShutdown.add(factory);
  
                                   IsolatedExecutor.transferAdhoc((SerializableBiConsumer<InterceptorOfGlobalMethods, IntSupplier>) InterceptorOfGlobalMethods.Global::unsafeSet, classLoader)
                                                   .accept(interceptorOfGlobalMethods, () -> {
                                                       if (InterceptibleThread.isDeterministic())
                                                           throw failWithOOM();
                                                       return random.uniform(Integer.MIN_VALUE, Integer.MAX_VALUE);
                                                   });
                                   onShutdown.add(IsolatedExecutor.transferAdhoc((SerializableRunnable)InterceptorOfGlobalMethods.Global::unsafeReset, classLoader)::run);
                                   onShutdown.add(time.setup(num, classLoader));
  
                                   onUnexpectedShutdown.put(num, onShutdown);
                               }
  
                               @Override
                               public void beforeStartup(IInstance i)
                               {
                                   ((IInvokableInstance) i).unsafeAcceptOnThisThread(PasswordSaltSupplier::unsafeSet, () -> "$2a$05$rT01y27MnvpE7NgzwvYNFe");
                                   ((IInvokableInstance) i).unsafeAcceptOnThisThread(IfInterceptibleThread::setThreadLocalRandomCheck, (LongConsumer) threadLocalRandomCheck);
  
  
                                   int num = i.config().num();
                                   if (builder.memoryListener != null)
                                   {
                                      ((IInvokableInstance) i).unsafeAcceptOnThisThread(HeapPool.Logged::setListener, builder.memoryListener);
                                       onUnexpectedShutdown.get(num).add(() -> ((IInvokableInstance) i).unsafeAcceptOnThisThread(HeapPool.Logged::setListener, (ignore1, ignore2) -> {}));
                                   }
  
                                   ((IInvokableInstance) i).unsafeAcceptOnThisThread(PaxosPrepare::setOnLinearizabilityViolation, SimulatorUtils::failWithOOM);
                                   onUnexpectedShutdown.get(num).add(() -> ((IInvokableInstance) i).unsafeRunOnThisThread(() -> PaxosPrepare.setOnLinearizabilityViolation(null)));
                               }
  
                               @Override
                               public void afterStartup(IInstance i)
                               {
                                   int num = i.config().num();
                                   ((IInvokableInstance) i).unsafeAcceptOnThisThread(BallotGenerator.Global::unsafeSet, (BallotGenerator) ballots.get());
                                   onUnexpectedShutdown.get(num).add(() -> ((IInvokableInstance) i).unsafeRunOnThisThread(() -> BallotGenerator.Global.unsafeSet(new BallotGenerator.Default())));
  
                                   ((IInvokableInstance) i).unsafeAcceptOnThisThread((SerializableConsumer<BufferPool.DebugLeaks>) debug -> BufferPools.forChunkCache().debug(null, debug), failures);
                                   onUnexpectedShutdown.get(num).add(() -> ((IInvokableInstance) i).unsafeRunOnThisThread(() -> BufferPools.forChunkCache().debug(null, null)));
  
                                   ((IInvokableInstance) i).unsafeAcceptOnThisThread((SerializableConsumer<BufferPool.DebugLeaks>) debug -> BufferPools.forNetworking().debug(null, debug), failures);
                                   onUnexpectedShutdown.get(num).add(() -> ((IInvokableInstance) i).unsafeRunOnThisThread(() -> BufferPools.forNetworking().debug(null, null)));
  
                                   ((IInvokableInstance) i).unsafeAcceptOnThisThread((SerializableConsumer<Ref.OnLeak>) Ref::setOnLeak, failures);
                                   onUnexpectedShutdown.get(num).add(() -> ((IInvokableInstance) i).unsafeRunOnThisThread(() -> Ref.setOnLeak(null)));
                               }
                           }).withClassTransformer(interceptClasses)
                             .withShutdownExecutor((name, classLoader, shuttingDown, call) -> {
                                 onPostShutdown.add(call);
                                 return null;
                             })
          ).createWithoutStarting();
  
          IfInterceptibleThread.setThreadLocalRandomCheck(threadLocalRandomCheck);
          snitch.setup(cluster);
          DirectStreamingConnectionFactory.setup(cluster);
          delivery = new SimulatedMessageDelivery(cluster);
          failureDetector = new SimulatedFailureDetector(cluster);
          FutureActionScheduler futureActionScheduler = builder.futureActionScheduler(numOfNodes, time, random);
          Map<Verb, FutureActionScheduler> perVerbFutureActionScheduler = builder.perVerbFutureActionSchedulers(numOfNodes, time, random, futureActionScheduler);
          simulated = new SimulatedSystems(random, time, delivery, execution, ballots, failureDetector, snitch, futureActionScheduler, perVerbFutureActionScheduler, builder.debug, failures);
          if (futureActionScheduler instanceof SimulatedFutureActionScheduler)
              simulated.register((SimulatedFutureActionScheduler) futureActionScheduler);
  
          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),
                                               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;
  
          // during cluster shutdown ignore all failures as there is no reason to track them
          onPreShutdown.add(() -> {simulated.failures.ignoreFailures(); return null;});
      }
  
      public S simulation()
      {
          return factory.create(simulated, scheduler, cluster, options);
      }
  
      public synchronized void close() throws IOException
      {
          // Re-enable time on shutdown
          try
          {
              Field field = Clock.Global.class.getDeclaredField("instance");
              field.setAccessible(true);
  
              Field modifiersField = Field.class.getDeclaredField("modifiers");
              modifiersField.setAccessible(true);
              modifiersField.setInt(field, field.getModifiers() & ~Modifier.FINAL);
  
              field.set(null, new Clock.Default());
          }
          catch (NoSuchFieldException|IllegalAccessException e)
          {
              throw new RuntimeException(e);
          }
  
          threadLocalRandomCheck.stop();
          simulated.execution.forceStop();
          SimulatedTime.Global.disable();
  
          Throwable fail = null;
          for (int num = 1 ; num <= cluster.size() ; ++num)
          {
              if (!cluster.get(num).isShutdown())
              {
                  fail = Throwables.close(fail, onUnexpectedShutdown.get(num));
              }
          }
  
          for (Callable<Void> call : onPreShutdown)
          {
              try
              {
                  call.call();
              }
              catch (Throwable t)
              {
                  fail = Throwables.merge(fail, t);
              }
          }
  
          try
          {
              cluster.close();
          }
          catch (Throwable t)
          {
              fail = Throwables.merge(fail, t);
          }
          for (Callable<Void> call : onPostShutdown)
          {
              try
              {
                  call.call();
              }
              catch (Throwable t)
              {
                  fail = Throwables.merge(fail, t);
              }
          }
          Throwables.maybeFail(fail, IOException.class);
      }
  }
  