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.debug;
  
  import java.io.BufferedReader;
  import java.io.ByteArrayInputStream;
  import java.io.IOException;
  import java.io.InputStreamReader;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.LinkedHashSet;
  import java.util.List;
  import java.util.Set;
  import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
  import java.util.function.Supplier;
  import java.util.regex.Pattern;
  import java.util.zip.GZIPInputStream;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.Util;
  import org.apache.cassandra.io.util.DataInputPlus;
  import org.apache.cassandra.io.util.DataInputPlus.DataInputStreamPlus;
  import org.apache.cassandra.io.util.File;
  import org.apache.cassandra.simulator.ClusterSimulation;
  import org.apache.cassandra.simulator.RandomSource;
  import org.apache.cassandra.simulator.Simulation;
  import org.apache.cassandra.simulator.SimulationRunner.RecordOption;
  import org.apache.cassandra.simulator.systems.InterceptedWait.CaptureSites.Capture;
  import org.apache.cassandra.simulator.systems.SimulatedTime;
  import org.apache.cassandra.utils.Closeable;
  import org.apache.cassandra.utils.CloseableIterator;
  import org.apache.cassandra.utils.concurrent.Threads;
  
  import static org.apache.cassandra.simulator.SimulationRunner.RecordOption.NONE;
  import static org.apache.cassandra.simulator.SimulationRunner.RecordOption.VALUE;
  import static org.apache.cassandra.simulator.SimulationRunner.RecordOption.WITH_CALLSITES;
  import static org.apache.cassandra.simulator.SimulatorUtils.failWithOOM;
  
  public class Reconcile
  {
      private static final Logger logger = LoggerFactory.getLogger(Reconcile.class);
  
      private static final Pattern STRIP_TRACES = Pattern.compile("(Wakeup|Continue|Timeout|Waiting)\\[(((([a-zA-Z]\\.)*[a-zA-Z0-9_$]+\\.[a-zA-Z0-9_<>$]+:[\\-0-9]+; )*(([a-zA-Z]\\.)*[a-zA-Z0-9_$]+\\.[a-zA-Z0-9_<>$]+:[\\-0-9]+))( #\\[.*?]#)?) ?(by\\[.*?])?]");
      private static final Pattern STRIP_NOW_TRACES = Pattern.compile("( #\\[.*?]#)");
      private static final Pattern NORMALISE_THREAD_RECORDING_IN = Pattern.compile("(Thread\\[[^]]+:[0-9]+),?[0-9]+(,node[0-9]+)]");
      static final Pattern NORMALISE_LAMBDA = Pattern.compile("((\\$\\$Lambda\\$[0-9]+/(0x)?[a-f0-9]+)?(@[0-9a-f]+)?)");
      static final Pattern NORMALISE_THREAD = Pattern.compile("(Thread\\[[^]]+:[0-9]+),[0-9](,node[0-9]+)(_[0-9]+)?]");
  
      public static class AbstractReconciler
      {
          private static final Logger logger = LoggerFactory.getLogger(AbstractReconciler.class);
  
          final DataInputPlus in;
          final List<String> strings = new ArrayList<>();
          final boolean inputHasCallSites;
          final boolean reconcileCallSites;
          int line;
  
          public AbstractReconciler(DataInputPlus in, boolean inputHasCallSites, RecordOption reconcile)
          {
              this.in = in;
              this.inputHasCallSites = inputHasCallSites;
              this.reconcileCallSites = reconcile == WITH_CALLSITES;
          }
  
          String readInterned() throws IOException
          {
              int id = in.readVInt32();
              if (id == strings.size()) strings.add(in.readUTF());
              else if (id > strings.size()) throw failWithOOM();
              return strings.get(id);
          }
  
          private String readCallSite() throws IOException
          {
              if (!inputHasCallSites)
                  return "";
  
              String trace = in.readUTF();
              for (int i = trace.indexOf('\n') ; i >= 0 ; i = trace.indexOf('\n', i + 1))
                  ++line;
              return reconcileCallSites ? trace : "";
          }
  
          private String ourCallSite()
          {
              if (!reconcileCallSites)
                  return "";
  
              StackTraceElement[] ste = Thread.currentThread().getStackTrace();
              return Arrays.stream(ste, 4, ste.length)
                           .filter(st -> !st.getClassName().equals("org.apache.cassandra.simulator.debug.Reconcile")
                                         && !st.getClassName().equals("org.apache.cassandra.simulator.SimulationRunner$Reconcile")
                                         && !st.getClassName().equals("sun.reflect.NativeMethodAccessorImpl") // depends on async compile thread
                                         && !st.getClassName().startsWith("sun.reflect.GeneratedMethodAccessor")) // depends on async compile thread
                           .collect(new Threads.StackTraceCombiner(true, "", "\n", ""));
          }
  
          public void checkThread() throws IOException
          {
              // normalise lambda also strips Object.toString() inconsistencies for some Thread objects
              String thread = NORMALISE_LAMBDA.matcher(readInterned()).replaceAll("");
              String ourThread = NORMALISE_LAMBDA.matcher(Thread.currentThread().toString()).replaceAll("");
              String callSite = NORMALISE_LAMBDA.matcher(readCallSite()).replaceAll("");
              String ourCallSite = NORMALISE_LAMBDA.matcher(ourCallSite()).replaceAll("");
              if (!thread.equals(ourThread) || !callSite.equals(ourCallSite))
              {
                  logger.error(String.format("(%s,%s) != (%s,%s)", thread, callSite, ourThread, ourCallSite));
                  throw failWithOOM();
              }
          }
      }
  
      public static class TimeReconciler extends AbstractReconciler implements SimulatedTime.Listener, Closeable
      {
          boolean disabled;
  
          public TimeReconciler(DataInputPlus in, boolean inputHasCallSites, RecordOption reconcile)
          {
              super(in, inputHasCallSites, reconcile);
          }
  
          @Override
          public void close()
          {
              disabled = true;
          }
  
          @Override
          public synchronized void accept(String kind, long value)
          {
              if (disabled)
                  return;
  
              try
              {
                  String testKind = readInterned();
                  long testValue = in.readUnsignedVInt();
                  checkThread();
                  if (!kind.equals(testKind) || value != testValue)
                  {
                      logger.error("({},{}) != ({},{})", kind, value, testKind, testValue);
                      throw failWithOOM();
                  }
              }
              catch (IOException e)
              {
                  throw new RuntimeException(e);
              }
          }
      }
  
      public static class RandomSourceReconciler extends RandomSource.Abstract implements Supplier<RandomSource>, Closeable
      {
          private static final Logger logger = LoggerFactory.getLogger(RandomSourceReconciler.class);
          private static final AtomicReferenceFieldUpdater<Reconcile.RandomSourceReconciler, Thread> lockedUpdater = AtomicReferenceFieldUpdater.newUpdater(Reconcile.RandomSourceReconciler.class, Thread.class, "locked");
          final DataInputPlus in;
          final RandomSource wrapped;
          final AbstractReconciler threads;
          int count;
          volatile Thread locked;
          volatile boolean disabled;
  
          public RandomSourceReconciler(DataInputPlus in, RandomSource wrapped, boolean inputHasCallSites, RecordOption reconcile)
          {
              this.in = in;
              this.wrapped = wrapped;
              this.threads = new AbstractReconciler(in, inputHasCallSites, reconcile);
          }
  
          private void enter()
          {
              if (!lockedUpdater.compareAndSet(this, null, Thread.currentThread()))
              {
                  if (disabled)
                      return;
  
                  disabled = true;
                  logger.error("Race within RandomSourceReconciler - means we have a Simulator bug permitting two threads to run at once");
                  throw failWithOOM();
              }
          }
  
          private void exit()
          {
              locked = null;
          }
  
          public void onDeterminismCheck(long value)
          {
              if (disabled)
                  return;
  
              enter();
              try
              {
                  byte type = in.readByte();
                  int c = in.readVInt32();
                  long v = in.readLong();
                  threads.checkThread();
                  if (type != 7 || c != count || value != v)
                  {
                      logger.error(String.format("(%d,%d,%d) != (%d,%d,%d)", 7, count, value, type, c, v));
                      throw failWithOOM();
                  }
                  ++count;
              }
              catch (IOException e)
              {
                  throw new RuntimeException(e);
              }
              finally
              {
                  exit();
              }
          }
  
          public int uniform(int min, int max)
          {
              int v = wrapped.uniform(min, max);
              if (disabled)
                  return v;
  
              enter();
              try
              {
                  byte type = in.readByte();
                  int c = in.readVInt32();
                  threads.checkThread();
                  int min1 = in.readVInt32();
                  int max1 = in.readVInt32() + min1;
                  int v1 = in.readVInt32() + min1;
                  if (type != 1 || min != min1 || max != max1 || v != v1 || c != count)
                  {
                      logger.error(String.format("(%d,%d,%d[%d,%d]) != (%d,%d,%d[%d,%d])", 1, count, v, min, max, type, c, v1, min1, max1));
                      throw failWithOOM();
                  }
                  ++count;
              }
              catch (IOException e)
              {
                  throw new RuntimeException(e);
              }
              finally
              {
                  exit();
              }
              return v;
          }
  
          public long uniform(long min, long max)
          {
              long v = wrapped.uniform(min, max);
              if (disabled)
                  return v;
  
              enter();
              try
              {
                  byte type = in.readByte();
                  int c = in.readVInt32();
                  threads.checkThread();
                  long min1 = in.readVInt();
                  long max1 = in.readVInt() + min1;
                  long v1 = in.readVInt() + min1;
                  if (type != 2 || min != min1 || max != max1 || v != v1 || c != count)
                  {
                      logger.error(String.format("(%d,%d,%d[%d,%d]) != (%d,%d,%d[%d,%d])", 2, count, v, min, max, type, c, v1, min1, max1));
                      throw failWithOOM();
                  }
                  ++count;
              }
              catch (IOException e)
              {
                  throw new RuntimeException(e);
              }
              finally
              {
                  exit();
              }
              return v;
          }
  
          public float uniformFloat()
          {
              float v = wrapped.uniformFloat();
              if (disabled)
                  return v;
  
              enter();
              try
              {
                  byte type = in.readByte();
                  int c = in.readVInt32();
                  threads.checkThread();
                  float v1 = in.readFloat();
                  if (type != 3 || v != v1 || c != count)
                  {
                      logger.error(String.format("(%d,%d,%f) != (%d,%d,%f)", 3, count, v, type, c, v1));
                      throw failWithOOM();
                  }
                  ++count;
              }
              catch (IOException e)
              {
                  throw new RuntimeException(e);
              }
              finally
              {
                  exit();
              }
              return v;
          }
  
          @Override
          public double uniformDouble()
          {
              double v = wrapped.uniformDouble();
              if (disabled)
                  return v;
  
              enter();
              try
              {
                  byte type = in.readByte();
                  int c = in.readVInt32();
                  threads.checkThread();
                  double v1 = in.readDouble();
                  if (type != 6 || v != v1 || c != count)
                  {
                      logger.error(String.format("(%d,%d,%f) != (%d,%d,%f)", 6, count, v, type, c, v1));
                      throw failWithOOM();
                  }
                  ++count;
              }
              catch (IOException e)
              {
                  throw new RuntimeException(e);
              }
              finally
              {
                  exit();
              }
              return v;
          }
  
          public synchronized void reset(long seed)
          {
              wrapped.reset(seed);
              if (disabled)
                  return;
  
              enter();
              try
              {
                  byte type = in.readByte();
                  int c = in.readVInt32();
                  long v1 = in.readVInt();
                  if (type != 4 || seed != v1 || c != count)
                      throw failWithOOM();
                  ++count;
              }
              catch (IOException e)
              {
                  throw new RuntimeException(e);
              }
              finally
              {
                  exit();
              }
          }
  
          public synchronized long reset()
          {
              long v = wrapped.reset();
              if (disabled)
                  return v;
  
              enter();
              try
              {
                  byte type = in.readByte();
                  int c = in.readVInt32();
                  long v1 = in.readVInt();
                  if (type != 5 || v != v1 || c != count)
                      throw failWithOOM();
                  ++count;
              }
              catch (IOException e)
              {
                  throw new RuntimeException(e);
              }
              finally
              {
                  exit();
              }
              return v;
          }
  
          public synchronized RandomSource get()
          {
              if (count++ > 0)
                  throw failWithOOM();
              return this;
          }
  
          @Override
          public void close()
          {
              disabled = true;
          }
      }
  
      public static void reconcileWith(String loadFromDir, long seed, RecordOption withRng, RecordOption withTime, ClusterSimulation.Builder<?> builder)
      {
          File eventFile = new File(new File(loadFromDir), Long.toHexString(seed) + ".gz");
          File rngFile = new File(new File(loadFromDir), Long.toHexString(seed) + ".rng.gz");
          File timeFile = new File(new File(loadFromDir), Long.toHexString(seed) + ".time.gz");
  
          try (BufferedReader eventIn = new BufferedReader(new InputStreamReader(new GZIPInputStream(eventFile.newInputStream())));
               DataInputStreamPlus rngIn = Util.DataInputStreamPlusImpl.wrap(rngFile.exists() && withRng != NONE ? new GZIPInputStream(rngFile.newInputStream()) : new ByteArrayInputStream(new byte[0]));
               DataInputStreamPlus timeIn = Util.DataInputStreamPlusImpl.wrap(timeFile.exists() && withTime != NONE ? new GZIPInputStream(timeFile.newInputStream()) : new ByteArrayInputStream(new byte[0])))
          {
              boolean inputHasWaitSites, inputHasWakeSites, inputHasRngCallSites, inputHasTimeCallSites;
              {
                  String modifiers = eventIn.readLine();
                  if (!modifiers.startsWith("modifiers:"))
                      throw new IllegalStateException();
  
                  builder.capture(new Capture(
                      builder.capture().waitSites & (inputHasWaitSites = modifiers.contains("waitSites")),
                      builder.capture().wakeSites & (inputHasWakeSites = modifiers.contains("wakeSites")),
                      builder.capture().nowSites)
                  );
                  inputHasRngCallSites = modifiers.contains("rngCallSites");
                  if (!modifiers.contains("rng")) withRng = NONE;
                  if (withRng == WITH_CALLSITES && !inputHasRngCallSites) withRng = VALUE;
  
                  inputHasTimeCallSites = modifiers.contains("timeCallSites");
                  if (!modifiers.contains("time")) withTime = NONE;
                  if (withTime == WITH_CALLSITES && !inputHasTimeCallSites) withTime = VALUE;
              }
              if (withRng != NONE && !rngFile.exists())
                  throw new IllegalStateException();
              if (withTime != NONE && !timeFile.exists())
                  throw new IllegalStateException();
  
              {
                  Set<String> modifiers = new LinkedHashSet<>();
                  if (withRng == WITH_CALLSITES)
                      modifiers.add("rngCallSites");
                  else if (withRng == VALUE)
                      modifiers.add("rng");
                  if (withTime == WITH_CALLSITES)
                      modifiers.add("timeCallSites");
                  else if (withTime == VALUE)
                      modifiers.add("time");
                  if (builder.capture().waitSites)
                      modifiers.add("WaitSites");
                  if (builder.capture().wakeSites)
                      modifiers.add("WakeSites");
                  logger.error("Seed 0x{} ({}) (With: {})", Long.toHexString(seed), eventFile, modifiers);
              }
  
              RandomSourceReconciler random = null;
              TimeReconciler time = null;
              if (withRng != NONE)
              {
                  builder.random(random = new RandomSourceReconciler(rngIn, new RandomSource.Default(), inputHasRngCallSites, withRng));
                  builder.onThreadLocalRandomCheck(random::onDeterminismCheck);
              }
              if (withTime != NONE)
                  builder.timeListener(time = new TimeReconciler(timeIn, inputHasTimeCallSites, withTime));
  
              class Line { int line = 1; } Line line = new Line(); // box for heap dump analysis
              try (ClusterSimulation<?> cluster = builder.create(seed);
                   Simulation simulation = cluster.simulation();
                   CloseableIterator<?> iter = simulation.iterator())
              {
                  try
                  {
                      while (iter.hasNext())
                      {
                          ++line.line;
                          String rawInput = eventIn.readLine();
                          String input = (inputHasWaitSites != builder.capture().waitSites || inputHasWakeSites != builder.capture().wakeSites)
                                         ? normaliseRecordingInWithoutWaitOrWakeSites(rawInput, inputHasWaitSites && !builder.capture().waitSites, inputHasWakeSites && !builder.capture().wakeSites)
                                         : normaliseRecordingIn(rawInput);
                          Object next = iter.next();
                          String rawOutput = next.toString();
                          String output = normaliseReconcileWithRecording(rawOutput);
                          if (!input.equals(output))
                              failWithHeapDump(line.line, input, output);
                      }
                      if (random != null)
                          random.close();
                      if (time != null)
                          time.close();
                  }
                  catch (Throwable t)
                  {
                      t.printStackTrace();
                      throw t;
                  }
              }
          }
          catch (Throwable t)
          {
              if (t instanceof Error)
                  throw (Error) t;
              throw new RuntimeException("Failed on seed " + Long.toHexString(seed), t);
          }
      }
  
      private static String normaliseRecordingIn(String input)
      {
          return STRIP_NOW_TRACES.matcher(
              NORMALISE_THREAD_RECORDING_IN.matcher(
                  NORMALISE_LAMBDA.matcher(input).replaceAll("")
              ).replaceAll("$1$2]")
          ).replaceAll("");
      }
  
      private static String normaliseRecordingInWithoutWaitOrWakeSites(String input, boolean stripWaitSites, boolean stripWakeSites)
      {
          return STRIP_TRACES.matcher(
              NORMALISE_THREAD_RECORDING_IN.matcher(
                  NORMALISE_LAMBDA.matcher(input).replaceAll("")
              ).replaceAll("$1$2]")
          ).replaceAll(stripWaitSites && stripWakeSites ? "$1[]" : stripWaitSites ? "$1[$9]" : "$1[$3]");
      }
  
      private static String normaliseReconcileWithRecording(String input)
      {
          return STRIP_NOW_TRACES.matcher(
              NORMALISE_THREAD.matcher(
                  NORMALISE_LAMBDA.matcher(input).replaceAll("")
              ).replaceAll("$1$2]")
          ).replaceAll("");
      }
  
      static void failWithHeapDump(int line, Object input, Object output)
      {
          logger.error("Line {}", line);
          logger.error("Input {}", input);
          logger.error("Output {}", output);
          throw failWithOOM();
      }
  }
  