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.ArrayList;
  import java.util.List;
  import java.util.concurrent.TimeUnit;
  import java.util.function.Predicate;
  import java.util.stream.Stream;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.utils.Shared;
  import org.apache.cassandra.utils.concurrent.Condition;
  import org.apache.cassandra.utils.concurrent.Threads;
  import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;
  
  import static org.apache.cassandra.simulator.SimulatorUtils.failWithOOM;
  import static org.apache.cassandra.simulator.systems.InterceptedWait.CaptureSites.Capture.WAKE;
  import static org.apache.cassandra.simulator.systems.InterceptedWait.CaptureSites.Capture.WAKE_AND_NOW;
  import static org.apache.cassandra.simulator.systems.InterceptedWait.Trigger.SIGNAL;
  import static org.apache.cassandra.simulator.systems.InterceptibleThread.interceptorOrDefault;
  import static org.apache.cassandra.utils.Shared.Recursive.ALL;
  import static org.apache.cassandra.utils.Shared.Recursive.INTERFACES;
  import static org.apache.cassandra.utils.Shared.Scope.SIMULATION;
  
  /**
   * A general abstraction for intercepted thread wait events, either
   * generated by the program execution or our nemesis system.
   */
  @Shared(scope = SIMULATION, inner = INTERFACES)
  public interface InterceptedWait extends NotifyThreadPaused
  {
      enum Kind { SLEEP_UNTIL, WAIT_UNTIL, UNBOUNDED_WAIT, NEMESIS }
      enum Trigger { TIMEOUT, INTERRUPT, SIGNAL }
  
      interface TriggerListener
      {
          /**
           * Invoked when the wait is triggered, permitting any dependent Action to be invalidated.
           * This is particularly useful for thread timeouts, which are often logically invalidated
           * but may otherwise hold up scheduling of further events until their scheduled time passes.
           * @param triggered the wait that has been triggered, and is no longer valid
           */
          void onTrigger(InterceptedWait triggered);
      }
  
      /**
       * The kind of simulated wait
       */
      Kind kind();
  
      /**
       * true if the signal has already been triggered by another simulation action
       */
      boolean isTriggered();
  
      /**
       * true if this wait can be interrupted
       */
      boolean isInterruptible();
  
      /**
       * If kind() == TIMED_WAIT or ABSOLUTE_TIMED_WAIT this returns the relative and absolute
       * period to wait in nanos
       */
      long waitTime();
  
      /**
       * Intercept a wakeup signal on this wait
       */
      void interceptWakeup(Trigger trigger, Thread by);
  
      /**
       * Signal the waiter immediately, and have the caller wait until its simulation has terminated.
       *
       * @param interceptor the interceptor to relay events to
       * @param trigger if SIGNAL, propagate the signal to the wrapped condition we are waiting on
       */
      void triggerAndAwaitDone(InterceptorOfConsequences interceptor, Trigger trigger);
  
      /**
       * Signal all waiters immediately, bypassing the simulation
       */
      void triggerBypass();
  
      /**
       * Add a trigger listener, to notify the wait is no longer valid
       */
      void addListener(TriggerListener onTrigger);
  
      Thread waiting();
  
      /**
       * A general purpose superclass for implementing an intercepted/simulated thread wait event.
       * All share this implementation except for monitor waits, which must use the monitor they are waiting on
       * in order to release its lock.
       */
      class InterceptedConditionWait extends NotInterceptedSyncCondition implements InterceptedWait
      {
          static final Logger logger = LoggerFactory.getLogger(InterceptedConditionWait.class);
  
          final Kind kind;
          final InterceptibleThread waiting;
          final CaptureSites captureSites;
          final InterceptorOfConsequences interceptedBy;
          final Condition propagateSignal;
          final List<TriggerListener> onTrigger = new ArrayList<>(3);
          final long waitTime;
          boolean isInterruptible, isSignalPending, isTriggered, isDone, hasExited;
  
          public InterceptedConditionWait(Kind kind, long waitTime, InterceptibleThread waiting, CaptureSites captureSites, Condition propagateSignal)
          {
              this.kind = kind;
              this.waitTime = waitTime;
              this.waiting = waiting;
              this.captureSites = captureSites;
              this.interceptedBy = waiting.interceptedBy();
              this.propagateSignal = propagateSignal;
          }
  
          public synchronized void triggerAndAwaitDone(InterceptorOfConsequences interceptor, Trigger trigger)
          {
              if (isTriggered)
                  return;
  
              if (hasExited)
              {
                  logger.error("{} exited without trigger {}", waiting, captureSites == null ? new CaptureSites(waiting, WAKE_AND_NOW) : captureSites);
                  throw failWithOOM();
              }
  
              waiting.beforeInvocation(interceptor, this);
              isTriggered = true;
              onTrigger.forEach(listener -> listener.onTrigger(this));
  
              if (!waiting.preWakeup(this) || !isInterruptible)
                  super.signal();
  
              if (isSignalPending && propagateSignal != null)
                  propagateSignal.signal();
  
              try
              {
                  while (!isDone)
                      wait();
              }
              catch (InterruptedException ie)
              {
                  throw new UncheckedInterruptedException(ie);
              }
          }
  
          public synchronized void triggerBypass()
          {
              if (isTriggered)
                  return;
  
              isTriggered = true;
              super.signal();
              if (propagateSignal != null)
                  propagateSignal.signal();
          }
  
          @Override
          public void addListener(TriggerListener onTrigger)
          {
              this.onTrigger.add(onTrigger);
          }
  
          @Override
          public Thread waiting()
          {
              return waiting;
          }
  
          @Override
          public synchronized void notifyThreadPaused()
          {
              isDone = true;
              notifyAll();
          }
  
          @Override
          public Kind kind()
          {
              return kind;
          }
  
          @Override
          public long waitTime()
          {
              return waitTime;
          }
  
          @Override
          public void interceptWakeup(Trigger trigger, Thread by)
          {
              assert !isTriggered;
              isSignalPending |= trigger == SIGNAL;
              if (captureSites != null)
                  captureSites.registerWakeup(by);
              interceptorOrDefault(by).interceptWakeup(this, trigger, interceptedBy);
          }
  
          public boolean isTriggered()
          {
              return isTriggered;
          }
  
          @Override
          public boolean isInterruptible()
          {
              return isInterruptible;
          }
  
          // ignore return value; always false as can only represent artificial (intercepted) signaled status
          public boolean await(long time, TimeUnit unit) throws InterruptedException
          {
              try
              {
                  isInterruptible = true;
                  super.await();
              }
              finally
              {
                  hasExited = true;
              }
              return false;
          }
  
          // ignore return value; always false as can only represent artificial (intercepted) signaled status
          public boolean awaitUntil(long until) throws InterruptedException
          {
              try
              {
                  isInterruptible = true;
                  super.await();
              }
              finally
              {
                  hasExited = true;
              }
              return false;
          }
  
          // ignore return value; always false as can only represent artificial (intercepted) signaled status
          public boolean awaitUntilUninterruptibly(long until)
          {
              try
              {
                  isInterruptible = false;
                  super.awaitUninterruptibly();
              }
              finally
              {
                  hasExited = true;
              }
              return false;
          }
  
          // ignore return value; always false as can only represent artificial (intercepted) signaled status
          public boolean awaitUninterruptibly(long time, TimeUnit units)
          {
              try
              {
                  isInterruptible = false;
                  super.awaitUninterruptibly(time, units);
              }
              finally
              {
                  hasExited = true;
              }
              return false;
          }
  
          public Condition await() throws InterruptedException
          {
              try
              {
                  isInterruptible = true;
                  super.await();
              }
              finally
              {
                  hasExited = true;
              }
              return this;
          }
  
          // declared as uninterruptible to the simulator to avoid unnecessary wakeups, but handles interrupts if they arise
          public Condition awaitDeclaredUninterruptible() throws InterruptedException
          {
              try
              {
                  isInterruptible = false;
                  super.await();
              }
              finally
              {
                  hasExited = true;
              }
              return this;
          }
  
          public Condition awaitUninterruptibly()
          {
              try
              {
                  isInterruptible = false;
                  super.awaitUninterruptibly();
              }
              finally
              {
                  hasExited = true;
              }
              return this;
          }
  
          public String toString()
          {
              return captureSites == null ? "" : "[" + captureSites + ']';
          }
      }
  
      // debug the place at which a thread waits or is signalled
      @Shared(scope = SIMULATION, members = ALL)
      public static class CaptureSites
      {
          public static class Capture
          {
              public static final Capture NONE = new Capture(false, false, false);
              public static final Capture WAKE = new Capture(true, false, false);
              public static final Capture WAKE_AND_NOW = new Capture(true, true, false);
  
              public final boolean waitSites;
              public final boolean wakeSites;
              public final boolean nowSites;
  
              public Capture(boolean waitSites, boolean wakeSites, boolean nowSites)
              {
                  this.waitSites = waitSites;
                  this.wakeSites = wakeSites;
                  this.nowSites = nowSites;
              }
  
              public boolean any()
              {
                  return waitSites | wakeSites | nowSites;
              }
          }
  
          final Thread waiting;
          final StackTraceElement[] waitSite;
          final Capture capture;
          @SuppressWarnings("unused") Thread waker;
          StackTraceElement[] wakeupSite;
  
          public CaptureSites(Thread waiting, StackTraceElement[] waitSite, Capture capture)
          {
              this.waiting = waiting;
              this.waitSite = waitSite;
              this.capture = capture;
          }
  
          public CaptureSites(Thread waiting, Capture capture)
          {
              this.waiting = waiting;
              this.waitSite = waiting.getStackTrace();
              this.capture = capture;
          }
  
          public CaptureSites(Thread waiting)
          {
              this.waiting = waiting;
              this.waitSite = waiting.getStackTrace();
              this.capture = WAKE;
          }
  
          public void registerWakeup(Thread waking)
          {
              this.waker = waking;
              if (capture.wakeSites)
                  this.wakeupSite = waking.getStackTrace();
          }
  
          public String toString(Predicate<StackTraceElement> include)
          {
              String tail;
              if (wakeupSite != null)
                  tail = Threads.prettyPrint(Stream.of(wakeupSite).filter(include), true, capture.nowSites ? "]# by[" : waitSite != null ? " by[" : "by[", "; ", "]");
              else if (capture.nowSites)
                  tail = "]#";
              else
                  tail = "";
              if (capture.nowSites)
                  tail = Threads.prettyPrint(Stream.of(waiting.getStackTrace()).filter(include), true, waitSite != null ? " #[" : "#[", "; ", tail);
              if (waitSite != null)
                  tail =Threads.prettyPrint(Stream.of(waitSite).filter(include), true, "", "; ", tail);
              return tail;
          }
  
          public String toString()
          {
              return toString(ignore -> true);
          }
      }
  
  }
  