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.systems;
  
  import java.util.ArrayDeque;
  import java.util.ArrayList;
  import java.util.Deque;
  import java.util.IdentityHashMap;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.TimeUnit;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import net.nicoulaj.compilecommand.annotations.Inline;
  import org.apache.cassandra.simulator.RandomSource;
  import org.apache.cassandra.simulator.systems.InterceptedWait.InterceptedConditionWait;
  import org.apache.cassandra.utils.Clock;
  import org.apache.cassandra.utils.Closeable;
  import org.apache.cassandra.utils.concurrent.Awaitable.SyncAwaitable;
  import org.apache.cassandra.utils.concurrent.Threads;
  import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
  
  import static java.util.concurrent.TimeUnit.MILLISECONDS;
  import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_SIMULATOR_DEBUG;
  import static org.apache.cassandra.simulator.SimulatorUtils.failWithOOM;
  import static org.apache.cassandra.simulator.systems.InterceptedWait.Kind.NEMESIS;
  import static org.apache.cassandra.simulator.systems.InterceptedWait.Kind.SLEEP_UNTIL;
  import static org.apache.cassandra.simulator.systems.InterceptedWait.Kind.UNBOUNDED_WAIT;
  import static org.apache.cassandra.simulator.systems.InterceptedWait.Kind.WAIT_UNTIL;
  import static org.apache.cassandra.simulator.systems.InterceptedWait.Trigger.SIGNAL;
  import static org.apache.cassandra.simulator.systems.InterceptibleThread.interceptorOrDefault;
  import static org.apache.cassandra.simulator.systems.InterceptingMonitors.WaitListAccessor.LOCK;
  import static org.apache.cassandra.simulator.systems.InterceptingMonitors.WaitListAccessor.NOTIFY;
  import static org.apache.cassandra.simulator.systems.SimulatedTime.Global.relativeToGlobalNanos;
  
  @PerClassLoader
  @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
  public abstract class InterceptingMonitors implements InterceptorOfGlobalMethods, Closeable
  {
      private static final boolean DEBUG_MONITOR_STATE = TEST_SIMULATOR_DEBUG.getBoolean();
  
      // eagerly initializing the logger prevents the dtest instance variables
      // from being setup correctly, which causes all nodes to log as <main>
      private static class LoggerHandle
      {
          private static final Logger logger = LoggerFactory.getLogger(InterceptingMonitors.class);
      }
  
      protected static Logger logger()
      {
          return LoggerHandle.logger;
      }
  
  
      static class MonitorState
      {
          InterceptedMonitorWait waitingOnNotify;
          InterceptedMonitorWait waitingOnLock;
          /**
           * The thread we have assigned lock ownership to.
           * This may not be actively holding the lock, if
           * we found it waiting for the monitor and assigned
           * it to receive
           */
          InterceptibleThread heldBy;
          int depth;
          int suspended;
          Deque<Object> recentActions = DEBUG_MONITOR_STATE ? new ArrayDeque<>() : null;
  
          boolean isEmpty()
          {
              return depth == 0 && waitingOnLock == null && waitingOnNotify == null && suspended == 0;
          }
  
          InterceptedMonitorWait removeAllWaitingOn(WaitListAccessor list)
          {
              InterceptedMonitorWait result = list.head(this);
              list.setHead(this, null);
  
              InterceptedMonitorWait cur = result;
              while (cur != null)
              {
                  InterceptedMonitorWait next = cur.next;
                  cur.waitingOn = null;
                  cur.next = null;
                  cur = next;
              }
              return result;
          }
  
          void removeWaitingOn(InterceptedMonitorWait remove)
          {
              if (remove.waitingOn != null)
              {
                  InterceptedMonitorWait head = remove.waitingOn.head(this);
                  remove.waitingOn.setHead(this, head.remove(remove));
                  assert remove.next == null;
              }
          }
  
          @Inline
          InterceptedMonitorWait removeOneWaitingOn(WaitListAccessor list, RandomSource random)
          {
              InterceptedMonitorWait head = list.head(this);
              if (head == null)
                  return null;
  
              if (head.next == null)
              {
                  list.setHead(this, null);
                  head.waitingOn = null;
                  return head;
              }
  
              int i = random.uniform(0, 1 + head.nextLength);
              if (i == 0)
              {
                  list.setHead(this, head.next);
                  head.next.nextLength = head.nextLength - 1;
                  head.next = null;
                  head.waitingOn = null;
                  return head;
              }
  
              InterceptedMonitorWait pred = head;
              while (--i > 0)
                  pred = pred.next;
  
              InterceptedMonitorWait result = pred.next;
              pred.next = result.next;
              --head.nextLength;
              result.next = null;
              result.waitingOn = null;
              return result;
          }
  
          void waitOn(WaitListAccessor list, InterceptedMonitorWait wait)
          {
              assert wait.waitingOn == null;
              wait.waitingOn = list;
  
              assert wait.next == null;
              InterceptedMonitorWait head = list.head(this);
              if (head != null)
              {
                  wait.next = head.next;
                  head.next = wait;
                  ++head.nextLength;
              }
              else
              {
                  list.setHead(this, wait);
                  wait.nextLength = 0;
              }
          }
  
          void suspend(InterceptedMonitorWait wait)
          {
              assert heldBy == wait.waiting;
              wait.suspendMonitor(depth);
              ++suspended;
              heldBy = null;
              depth = 0;
          }
  
          void restore(InterceptedMonitorWait wait)
          {
              assert heldBy == null || heldBy == wait.waiting;
              assert depth == 0;
              assert suspended > 0;
              heldBy = wait.waiting;
              depth = wait.unsuspendMonitor();
              --suspended;
          }
  
          void claim(InterceptedMonitorWait wait)
          {
              assert heldBy == null || heldBy == wait.waiting;
              assert depth == 0;
              heldBy = wait.waiting;
              depth = wait.unsuspendMonitor();
          }
  
          void log(Object event, Thread toThread, Thread byThread)
          {
              if (recentActions != null)
                  log(event + " " + toThread + " by " + byThread);
          }
  
          void log(Object event, Thread toThread)
          {
              if (recentActions != null)
                  log(event + " " + toThread);
          }
  
          void log(Object event)
          {
              if (recentActions == null)
                  return;
  
              if (recentActions.size() > 20)
                  recentActions.poll();
              recentActions.add(event + " " + depth);
          }
      }
  
      interface WaitListAccessor
      {
          static final WaitListAccessor NOTIFY = new WaitListAccessor()
          {
              @Override public InterceptedMonitorWait head(MonitorState state) { return state.waitingOnNotify; }
              @Override public void setHead(MonitorState state, InterceptedMonitorWait newHead) { state.waitingOnNotify = newHead; }
          };
  
          static final WaitListAccessor LOCK = new WaitListAccessor()
          {
              @Override public InterceptedMonitorWait head(MonitorState state) { return state.waitingOnLock; }
              @Override public void setHead(MonitorState state, InterceptedMonitorWait newHead) { state.waitingOnLock = newHead; }
          };
  
          InterceptedMonitorWait head(MonitorState state);
          void setHead(MonitorState state, InterceptedMonitorWait newHead);
      }
  
      static class InterceptedMonitorWait implements InterceptedWait
      {
          Kind kind;
          final long waitTime;
          final InterceptibleThread waiting;
          final CaptureSites captureSites;
          final InterceptorOfConsequences interceptedBy;
          final MonitorState state;
          final Object monitor;
          int suspendedMonitorDepth;
  
          Trigger trigger;
          boolean isTriggered;
          final List<TriggerListener> onTrigger = new ArrayList<>(3);
  
          boolean notifiedOfPause;
          boolean waitingOnRelinquish;
  
          WaitListAccessor waitingOn;
          volatile InterceptedMonitorWait next;
          int nextLength;
          boolean hasExited;
  
          InterceptedMonitorWait(Kind kind, long waitTime, MonitorState state, InterceptibleThread waiting, CaptureSites captureSites)
          {
              this.kind = kind;
              this.waitTime = waitTime;
              this.waiting = waiting;
              this.captureSites = captureSites;
              this.interceptedBy = waiting.interceptedBy();
              this.state = state;
              this.monitor = this;
          }
  
          InterceptedMonitorWait(Kind kind, long waitTime, MonitorState state, InterceptibleThread waiting, CaptureSites captureSites, Object object)
          {
              this.kind = kind;
              this.waitTime = waitTime;
              this.waiting = waiting;
              this.captureSites = captureSites;
              this.interceptedBy = waiting.interceptedBy();
              this.state = state;
              this.monitor = object;
          }
  
          @Override
          public Kind kind()
          {
              return kind;
          }
  
          void suspendMonitor(int depth)
          {
              assert suspendedMonitorDepth == 0;
              suspendedMonitorDepth = depth;
          }
  
          int unsuspendMonitor()
          {
              assert suspendedMonitorDepth > 0;
              int result = suspendedMonitorDepth;
              suspendedMonitorDepth = 0;
              return result;
          }
  
          public boolean isTriggered()
          {
              return isTriggered;
          }
  
          public boolean isInterruptible()
          {
              return true;
          }
  
          @Override
          public long waitTime()
          {
              return waitTime;
          }
  
          @Override
          public void interceptWakeup(Trigger trigger, Thread by)
          {
              if (this.trigger != null && this.trigger.compareTo(trigger) >= 0)
                  return;
  
              this.trigger = trigger;
              if (captureSites != null)
                  captureSites.registerWakeup(by);
              interceptorOrDefault(by).interceptWakeup(this, trigger, interceptedBy);
          }
  
          public void triggerAndAwaitDone(InterceptorOfConsequences interceptor, Trigger trigger)
          {
              if (isTriggered)
                  return;
  
              if (hasExited)
                  throw failWithOOM();
  
              state.removeWaitingOn(this); // if still present, remove
  
              // we may have been assigned ownership of the lock if we attempted to trigger but found the lock held
              if (state.heldBy != null && state.heldBy != waiting)
              {   // reset this condition to wait on lock release
                  state.waitOn(LOCK, this);
                  this.kind = UNBOUNDED_WAIT;
                  this.trigger = null;
                  interceptor.beforeInvocation(waiting);
                  interceptor.interceptWait(this);
                  return;
              }
  
              try
              {
                  synchronized (monitor)
                  {
                      waiting.beforeInvocation(interceptor, this);
  
                      isTriggered = true;
                      onTrigger.forEach(listener -> listener.onTrigger(this));
  
                      if (!waiting.preWakeup(this))
                          monitor.notifyAll(); // TODO: could use interrupts to target waiting anyway, avoiding notifyAll()
  
                      while (!notifiedOfPause)
                          monitor.wait();
  
                      if (waitingOnRelinquish)
                      {
                          waitingOnRelinquish = false;
                          monitor.notifyAll(); // TODO: could use interrupts to target waiting anyway, avoiding notifyAll()
                      }
                  }
              }
              catch (InterruptedException ie)
              {
                  throw new UncheckedInterruptedException(ie);
              }
          }
  
          @Override
          public void triggerBypass()
          {
              if (isTriggered)
                  return;
  
              synchronized (monitor)
              {
                  isTriggered = true;
                  monitor.notifyAll();
                  state.removeWaitingOn(this);
              }
          }
  
          @Override
          public void addListener(TriggerListener onTrigger)
          {
              this.onTrigger.add(onTrigger);
          }
  
          @Override
          public Thread waiting()
          {
              return waiting;
          }
  
          @Override
          public void notifyThreadPaused()
          {
              notifiedOfPause = true;
              if (Thread.holdsLock(monitor))
              {
                  monitor.notifyAll();
                  waitingOnRelinquish = true;
                  try { while (waitingOnRelinquish) monitor.wait(); }
                  catch (InterruptedException e) { throw new UncheckedInterruptedException(e); }
              }
              else
              {
                  synchronized (monitor)
                  {
                      monitor.notifyAll();
                  }
              }
          }
  
          void await() throws InterruptedException
          {
              try
              {
                  while (!isTriggered())
                      monitor.wait();
              }
              finally
              {
                  hasExited = true;
              }
          }
  
          InterceptedMonitorWait remove(InterceptedMonitorWait remove)
          {
              remove.waitingOn = null;
  
              if (remove == this)
              {
                  InterceptedMonitorWait next = this.next;
                  if (next != null)
                  {
                      next.nextLength = nextLength - 1;
                      remove.next = null;
                  }
  
                  return next;
              }
  
              InterceptedMonitorWait cur = this;
              while (cur != null && cur.next != remove)
                  cur = cur.next;
  
              if (cur != null)
              {
                  cur.next = remove.next;
                  remove.next = null;
                  --nextLength;
              }
              return this;
          }
  
          public String toString()
          {
              return captureSites == null ? "" : "[" + captureSites + ']';
          }
      }
  
      final RandomSource random;
      private final Map<Object, MonitorState> monitors = new IdentityHashMap<>();
      private final Map<Thread, Object> waitingOn = new IdentityHashMap<>();
      protected boolean disabled;
  
      public InterceptingMonitors(RandomSource random)
      {
          this.random = random;
      }
  
      private MonitorState state(Object monitor)
      {
          return monitors.computeIfAbsent(monitor, ignore -> new MonitorState());
      }
  
      private MonitorState maybeState(Object monitor)
      {
          return monitors.get(monitor);
      }
  
      private void maybeClear(Object monitor, MonitorState state)
      {
          if (state.isEmpty())
              monitors.remove(monitor, state);
      }
  
      @Override
      public void waitUntil(long deadline) throws InterruptedException
      {
          InterceptibleThread thread = ifIntercepted();
          if (thread == null)
          {
              Clock.waitUntil(deadline);
              return;
          }
  
          if (Thread.interrupted())
              throw new InterruptedException();
  
          InterceptedMonitorWait trigger = new InterceptedMonitorWait(SLEEP_UNTIL, deadline, new MonitorState(), thread, captureWaitSite(thread));
          thread.interceptWait(trigger);
          synchronized (trigger)
          {
              try
              {
                  trigger.await();
              }
              catch (InterruptedException e)
              {
                  if (!trigger.isTriggered)
                      throw e;
              }
          }
      }
  
      @Override
      public void sleep(long period, TimeUnit units) throws InterruptedException
      {
          waitUntil(nanoTime() + units.toNanos(period));
      }
  
      @Override
      public void sleepUninterriptibly(long period, TimeUnit units)
      {
          try
          {
              sleep(period, units);
          }
          catch (InterruptedException e)
          {
              // instead of looping uninterruptibly
              throw new UncheckedInterruptedException(e);
          }
      }
  
      public boolean waitUntil(Object monitor, long deadline) throws InterruptedException
      {
          InterceptibleThread thread = ifIntercepted();
          if (thread == null) return SyncAwaitable.waitUntil(monitor, deadline);
          else return wait(monitor, thread, WAIT_UNTIL, deadline);
      }
  
      @Override
      public void wait(Object monitor) throws InterruptedException
      {
          InterceptibleThread thread = ifIntercepted();
          if (thread == null) monitor.wait();
          else wait(monitor, thread, UNBOUNDED_WAIT, -1L);
      }
  
      @Override
      public void wait(Object monitor, long millis) throws InterruptedException
      {
          InterceptibleThread thread = ifIntercepted();
          if (thread == null) monitor.wait(millis);
          else wait(monitor, thread, WAIT_UNTIL, relativeToGlobalNanos(MILLISECONDS.toNanos(millis)));
      }
  
      @Override
      public void wait(Object monitor, long millis, int nanos) throws InterruptedException
      {
          InterceptibleThread thread = ifIntercepted();
          if (thread == null) monitor.wait(millis, nanos);
          else wait(monitor, thread, WAIT_UNTIL, relativeToGlobalNanos(MILLISECONDS.toNanos(millis) + nanos));
      }
  
      private boolean wait(Object monitor, InterceptibleThread thread, InterceptedWait.Kind kind, long waitNanos) throws InterruptedException
      {
          if (Thread.interrupted())
              throw new InterruptedException();
  
          MonitorState state = state(monitor);
          InterceptedMonitorWait trigger = new InterceptedMonitorWait(kind, waitNanos, state, thread, captureWaitSite(thread), monitor);
          state.log("enterwait", thread);
          state.suspend(trigger);
          state.waitOn(NOTIFY, trigger);
          wakeOneWaitingOnLock(thread, state);
          thread.interceptWait(trigger);
          try
          {
              trigger.await();
          }
          finally
          {
              state.restore(trigger);
              state.log("exitwait", thread);
          }
          return trigger.trigger == SIGNAL;
      }
  
      public void notify(Object monitor)
      {
          MonitorState state = state(monitor);
          if (state != null)
          {
              InterceptedMonitorWait wake = state.removeOneWaitingOn(NOTIFY, random);
              if (wake != null)
              {
                  // TODO: assign ownership on monitorExit
                  assert wake.waitingOn == null;
                  Thread waker = Thread.currentThread();
                  wake.interceptWakeup(SIGNAL, waker);
                  state.log("notify", wake.waiting, waker);
                  return;
              }
          }
          monitor.notify();
      }
  
      @Override
      public void notifyAll(Object monitor)
      {
          MonitorState state = state(monitor);
          if (state != null)
          {
              InterceptedMonitorWait wake = state.removeAllWaitingOn(NOTIFY);
              if (wake != null)
              {
                  Thread waker = Thread.currentThread();
                  wake.interceptWakeup(SIGNAL, waker);
                  state.log("notify", wake.waiting, waker);
  
                  wake = wake.next;
                  while (wake != null)
                  {
                      InterceptedMonitorWait next = wake.next;
                      state.waitOn(LOCK, wake);
                      state.log("movetowaitonlock ", wake.waiting, waker);
                      wake = next;
                  }
                  return;
              }
          }
          monitor.notifyAll();
      }
  
      @Override
      public void preMonitorEnter(Object monitor, float preMonitorDelayChance)
      {
          if (disabled)
              return;
  
          Thread anyThread = Thread.currentThread();
          if (!(anyThread instanceof InterceptibleThread))
              return;
  
          // save any interrupt before testing random.decide, in case we are trapping these for verification
          InterceptibleThread thread = (InterceptibleThread) anyThread;
          boolean restoreInterrupt = Thread.interrupted();
          try
          {
              if (   !thread.isEvaluationDeterministic()
                  && random.decide(preMonitorDelayChance))
              {
                  // TODO (feature): hold a stack of threads already paused by the nemesis, and, if one of the threads
                  //        is entering the monitor, put the contents of this stack into `waitingOn` for this monitor.
                  InterceptedConditionWait signal = new InterceptedConditionWait(NEMESIS, 0L, thread, captureWaitSite(thread), null);
                  thread.interceptWait(signal);
  
                  while (true)
                  {
                      try
                      {
                          signal.awaitDeclaredUninterruptible();
                          break;
                      }
                      catch (InterruptedException e)
                      {
                          if (disabled)
                              throw new UncheckedInterruptedException(e);
                          restoreInterrupt = true;
                      }
                  }
              }
  
              MonitorState state = state(monitor);
              if (state.heldBy != thread)
              {
                  if (state.heldBy != null)
                  {
                      if (!thread.isIntercepting() && disabled) return;
                      else if (!thread.isIntercepting())
                          throw new AssertionError("Thread " + thread + " is running but is not simulated");
  
                      checkForDeadlock(thread, state.heldBy);
                      InterceptedMonitorWait wait = new InterceptedMonitorWait(UNBOUNDED_WAIT, 0L, state, thread, captureWaitSite(thread));
                      wait.suspendedMonitorDepth = 1;
                      state.log("monitorenter_wait", thread);
                      state.waitOn(LOCK, wait);
                      thread.interceptWait(wait);
                      synchronized (wait)
                      {
                          waitingOn.put(thread, monitor);
                          restoreInterrupt |= Thread.interrupted();
                          try
                          {
                              while (true)
                              {
                                  try
                                  {
                                      wait.await();
                                      break;
                                  }
                                  catch (InterruptedException e)
                                  {
                                      if (disabled)
                                      {
                                          if (state.heldBy == thread)
                                          {
                                              state.heldBy = null;
                                              state.depth = 0;
                                          }
                                          throw new UncheckedInterruptedException(e);
                                      }
  
                                      restoreInterrupt = true;
                                      if (wait.isTriggered)
                                          break;
                                  }
                              }
                          }
                          finally
                          {
                              waitingOn.remove(thread);
                          }
                      }
                      state.claim(wait);
                      state.log("monitorenter_claim", thread);
                  }
                  else
                  {
                      state.log("monitorenter_free", thread);
                      state.heldBy = thread;
                      state.depth = 1;
                  }
              }
              else
              {
                  state.log("monitorreenter", thread);
                  state.depth++;
              }
          }
          finally
          {
              if (restoreInterrupt)
                  thread.interrupt();
          }
      }
  
      @Override
      public void preMonitorExit(Object monitor)
      {
          if (disabled)
              return;
  
          Thread thread = Thread.currentThread();
          if (!(thread instanceof InterceptibleThread))
              return;
  
          MonitorState state = maybeState(monitor);
          if (state == null)
              return;
  
          if (state.heldBy != thread)
              throw new AssertionError();
  
          if (--state.depth > 0)
          {
              state.log("monitorreexit", thread);
              return;
          }
  
          state.log("monitorexit", thread);
          state.heldBy = null;
  
          if (!wakeOneWaitingOnLock(thread, state))
          {
              maybeClear(monitor, state);
          }
      }
  
      private boolean wakeOneWaitingOnLock(Thread waker, MonitorState state)
      {
          InterceptedMonitorWait wake = state.removeOneWaitingOn(LOCK, random);
          if (wake != null)
          {
              assert wake.waitingOn == null;
              assert !wake.isTriggered();
  
              wake.interceptWakeup(SIGNAL, waker);
  
              // assign them the lock, so they'll definitely get it when they wake
              assert state.heldBy == null;
              state.heldBy = wake.waiting;
              state.log("wake", wake.waiting);
              return true;
          }
          return false;
      }
  
      // TODO (feature): integrate LockSupport waits into this deadlock check
      private void checkForDeadlock(Thread waiting, Thread blockedBy)
      {
          Thread cur = blockedBy;
          while (true)
          {
              Object monitor = waitingOn.get(cur);
              if (monitor == null)
                  return;
              MonitorState state = monitors.get(monitor);
              if (state == null)
                  return;
              Thread next = state.heldBy;
              if (next == cur)
                  return; // not really waiting, just hasn't woken up yet
              if (next == waiting)
              {
                  logger().error("Deadlock between {}{} and {}{}", waiting, Threads.prettyPrintStackTrace(waiting, true, ";"), cur, Threads.prettyPrintStackTrace(cur, true, ";"));
                  throw failWithOOM();
              }
              cur = next;
          }
      }
  
      @Override
      public void park()
      {
          InterceptibleThread.park();
      }
  
      @Override
      public void parkNanos(long nanos)
      {
          InterceptibleThread.parkNanos(nanos);
      }
  
      @Override
      public void parkUntil(long millis)
      {
          InterceptibleThread.parkUntil(millis);
      }
  
      @Override
      public void park(Object blocker)
      {
          InterceptibleThread.park(blocker);
      }
  
      @Override
      public void parkNanos(Object blocker, long nanos)
      {
          InterceptibleThread.parkNanos(blocker, nanos);
      }
  
      @Override
      public void parkUntil(Object blocker, long millis)
      {
          InterceptibleThread.parkUntil(blocker, millis);
      }
  
      @Override
      public void unpark(Thread thread)
      {
          InterceptibleThread.unpark(thread);
      }
  
      public void close()
      {
          disabled = true;
      }
  }
  