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.UUID;
  import java.util.function.Consumer;
  import java.util.function.LongConsumer;
  
  import javax.annotation.Nullable;
  
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import org.apache.cassandra.simulator.RandomSource;
  import org.apache.cassandra.simulator.systems.InterceptedWait.CaptureSites.Capture;
  import org.apache.cassandra.simulator.systems.InterceptedWait.InterceptedConditionWait;
  import org.apache.cassandra.utils.Clock;
  import org.apache.cassandra.utils.concurrent.Condition;
  import org.apache.cassandra.utils.concurrent.CountDownLatch;
  import org.apache.cassandra.utils.concurrent.Semaphore;
  import org.apache.cassandra.utils.concurrent.WaitQueue;
  
  import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_SIMULATOR_DETERMINISM_CHECK;
  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.NonInterceptible.Permit.OPTIONAL;
  import static org.apache.cassandra.simulator.systems.NonInterceptible.Permit.REQUIRED;
  
  @PerClassLoader
  public class InterceptingGlobalMethods extends InterceptingMonitors implements InterceptorOfGlobalMethods
  {
      private static final Logger logger = LoggerFactory.getLogger(InterceptingGlobalMethods.class);
      private static final boolean isDeterminismCheckStrict = TEST_SIMULATOR_DETERMINISM_CHECK.convert(name -> name.equals("strict"));
  
      private final @Nullable LongConsumer onThreadLocalRandomCheck;
      private final Capture capture;
      private int uniqueUuidCounter = 0;
      private final Consumer<Throwable> onUncaughtException;
  
      public InterceptingGlobalMethods(Capture capture, LongConsumer onThreadLocalRandomCheck, Consumer<Throwable> onUncaughtException, RandomSource random)
      {
          super(random);
          this.capture = capture.any() ? capture : null;
          this.onThreadLocalRandomCheck = onThreadLocalRandomCheck;
          this.onUncaughtException = onUncaughtException;
      }
  
      @Override
      public Semaphore newSemaphore(int count)
      {
          return new InterceptingSemaphore(count, false);
      }
  
      @Override
      public Semaphore newFairSemaphore(int count)
      {
          return new InterceptingSemaphore(count, true);
      }
  
      @Override
      public WaitQueue newWaitQueue()
      {
          return new InterceptingWaitQueue();
      }
  
      @Override
      public CountDownLatch newCountDownLatch(int count)
      {
          return new InterceptingAwaitable.InterceptingCountDownLatch(count);
      }
  
      @Override
      public Condition newOneTimeCondition()
      {
          return new InterceptingAwaitable.InterceptingCondition();
      }
  
      @Override
      public InterceptedWait.CaptureSites captureWaitSite(Thread thread)
      {
          if (capture == null)
              return null;
  
          return new InterceptedWait.CaptureSites(thread, capture);
      }
  
      @Override
      public InterceptibleThread ifIntercepted()
      {
          Thread thread = Thread.currentThread();
          if (thread instanceof InterceptibleThread)
          {
              InterceptibleThread interceptibleThread = (InterceptibleThread) thread;
              if (interceptibleThread.isIntercepting())
                  return interceptibleThread;
          }
  
          if (NonInterceptible.isPermitted())
              return null;
  
          if (!disabled)
          {
              logger().error("Caught a non-intercepted thread! " + thread, new RuntimeException());
              throw failWithOOM();
          }
  
          return null;
      }
  
      @Override
      public void uncaughtException(Thread thread, Throwable throwable)
      {
          onUncaughtException.accept(throwable);
      }
  
      @Override
      public void nemesis(float chance)
      {
          InterceptibleThread thread = ifIntercepted();
          if (thread == null || thread.isEvaluationDeterministic() || !random.decide(chance))
              return;
  
          InterceptedConditionWait signal = new InterceptedConditionWait(NEMESIS, 0L, thread, captureWaitSite(thread), null);
          thread.interceptWait(signal);
  
          // save interrupt state to restore afterwards - new ones only arrive if terminating simulation
          boolean restoreInterrupt = Thread.interrupted();
          try
          {
              while (true)
              {
                  try
                  {
                      signal.awaitDeclaredUninterruptible();
                      return;
                  }
                  catch (InterruptedException e)
                  {
                      restoreInterrupt = true;
                      if (disabled)
                          return;
                  }
              }
          }
          finally
          {
              if (restoreInterrupt)
                  thread.interrupt();
          }
      }
  
      @Override
      public long randomSeed()
      {
          InterceptibleThread thread = ifIntercepted();
          if (thread == null || thread.isEvaluationDeterministic())
              return Thread.currentThread().getName().hashCode();
  
          return random.uniform(Long.MIN_VALUE, Long.MAX_VALUE);
      }
  
      @Override
      public synchronized UUID randomUUID()
      {
          long msb = random.uniform(0, 1L << 60);
          msb = ((msb << 4) & 0xffffffffffff0000L) | 0x4000 | (msb & 0xfff);
          return new UUID(msb, (1L << 63) | uniqueUuidCounter++);
      }
  
      @Override
      public void threadLocalRandomCheck(long seed)
      {
          if (onThreadLocalRandomCheck != null)
              onThreadLocalRandomCheck.accept(seed);
      }
  
      public static class ThreadLocalRandomCheck implements LongConsumer
      {
          final LongConsumer wrapped;
          private boolean disabled;
  
          public ThreadLocalRandomCheck(LongConsumer wrapped)
          {
              this.wrapped = wrapped;
          }
  
          @Override
          public void accept(long value)
          {
              if (wrapped != null)
                  wrapped.accept(value);
  
              Thread thread = Thread.currentThread();
              if (thread instanceof InterceptibleThread)
              {
                  InterceptibleThread interceptibleThread = (InterceptibleThread) thread;
                  if (interceptibleThread.isIntercepting())
                      return;
              }
  
              if (NonInterceptible.isPermitted(isDeterminismCheckStrict ? OPTIONAL : REQUIRED))
                  return;
  
              if (!disabled)
                  throw failWithOOM();
          }
  
          public void stop()
          {
              disabled = true;
          }
      }
  
      @Override
      public long nanoTime()
      {
          return Clock.Global.nanoTime();
      }
  
      @Override
      public long currentTimeMillis()
      {
          return Clock.Global.currentTimeMillis();
      }
  }
  